From ca2e71c3814025dea776f84ab72c1d563eb888d5 Mon Sep 17 00:00:00 2001
From: Amit Singh Chowdhery <amit.f.singh@ericsson.com>
Date: Mon, 28 Mar 2016 19:06:42 +0200
Subject: [PATCH] Add/drop multiple columns in one ALTER TABLE statement

patch by Amit Singh Chowdhery; reviewed by Robert Stupp for CASSANDRA-10411
---
 CHANGES.txt                                   |   1 +
 doc/cql3/CQL.textile                          |   3 +
 src/java/org/apache/cassandra/cql3/Cql.g      |  26 +-
 .../cql3/statements/AlterTableStatement.java  | 283 ++++++++++--------
 .../statements/AlterTableStatementColumn.java |  53 ++++
 .../cql3/validation/operations/AlterTest.java |  81 +++++
 6 files changed, 310 insertions(+), 137 deletions(-)
 create mode 100644 src/java/org/apache/cassandra/cql3/statements/AlterTableStatementColumn.java

diff --git a/CHANGES.txt b/CHANGES.txt
index b80fdf3cf4..8e9cc44684 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 3.6
+ * Add/drop multiple columns in one ALTER TABLE statement (CASSANDRA-10411)
  * Add require_endpoint_verification opt for internode encryption (CASSANDRA-9220)
  * Add auto import java.util for UDF code block (CASSANDRA-11392)
  * Add --hex-format option to nodetool getsstables (CASSANDRA-11337)
diff --git a/doc/cql3/CQL.textile b/doc/cql3/CQL.textile
index 336a64cded..1ee2537e8b 100644
--- a/doc/cql3/CQL.textile
+++ b/doc/cql3/CQL.textile
@@ -396,7 +396,9 @@ bc(syntax)..
 
 <instruction> ::= ALTER <identifier> TYPE <type>
                 | ADD   <identifier> <type>
+                | ADD   ( <identifier> <type> ( , <identifier> <type> )* )
                 | DROP  <identifier>
+                | DROP  ( <identifier> ( , <identifier> )* )
                 | WITH  <option> ( AND <option> )*
 p. 
 __Sample:__
@@ -2312,6 +2314,7 @@ The following describes the changes in each version of CQL.
 h3. 3.4.2
 
 * "@INSERT/UPDATE options@":#updateOptions for tables having a default_time_to_live specifying a TTL of 0 will remove the TTL from the inserted or updated values
+* "@ALTER TABLE@":#alterTableStmt @ADD@ and @DROP@ now allow mutiple columns to be added/removed
 
 h3. 3.4.1
 
diff --git a/src/java/org/apache/cassandra/cql3/Cql.g b/src/java/org/apache/cassandra/cql3/Cql.g
index 5cb479cf56..f7841fd1c0 100644
--- a/src/java/org/apache/cassandra/cql3/Cql.g
+++ b/src/java/org/apache/cassandra/cql3/Cql.g
@@ -848,8 +848,8 @@ alterKeyspaceStatement returns [AlterKeyspaceStatement expr]
 
 /**
  * ALTER COLUMN FAMILY <CF> ALTER <column> TYPE <newtype>;
- * ALTER COLUMN FAMILY <CF> ADD <column> <newtype>;
- * ALTER COLUMN FAMILY <CF> DROP <column>;
+ * ALTER COLUMN FAMILY <CF> ADD <column> <newtype>; | ALTER COLUMN FAMILY <CF> ADD (<column> <newtype>,<column1> <newtype1>..... <column n> <newtype n>)
+ * ALTER COLUMN FAMILY <CF> DROP <column>; | ALTER COLUMN FAMILY <CF> DROP ( <column>,<column1>.....<column n>)
  * ALTER COLUMN FAMILY <CF> WITH <property> = <value>;
  * ALTER COLUMN FAMILY <CF> RENAME <column> TO <column>;
  */
@@ -858,19 +858,31 @@ alterTableStatement returns [AlterTableStatement expr]
         AlterTableStatement.Type type = null;
         TableAttributes attrs = new TableAttributes();
         Map<ColumnIdentifier.Raw, ColumnIdentifier.Raw> renames = new HashMap<ColumnIdentifier.Raw, ColumnIdentifier.Raw>();
-        boolean isStatic = false;
+        List<AlterTableStatementColumn> colNameList = new ArrayList<AlterTableStatementColumn>();
     }
     : K_ALTER K_COLUMNFAMILY cf=columnFamilyName
-          ( K_ALTER id=cident K_TYPE v=comparatorType { type = AlterTableStatement.Type.ALTER; }
-          | K_ADD   id=cident v=comparatorType ({ isStatic=true; } K_STATIC)? { type = AlterTableStatement.Type.ADD; }
-          | K_DROP  id=cident                         { type = AlterTableStatement.Type.DROP; }
+          ( K_ALTER id=cident  K_TYPE v=comparatorType  { type = AlterTableStatement.Type.ALTER; } { colNameList.add(new AlterTableStatementColumn(id,v)); }
+          | K_ADD  (        (id=cident   v=comparatorType   b1=cfisStatic { colNameList.add(new AlterTableStatementColumn(id,v,b1)); })
+                     | ('('  id1=cident  v1=comparatorType  b1=cfisStatic { colNameList.add(new AlterTableStatementColumn(id1,v1,b1)); }
+                       ( ',' idn=cident  vn=comparatorType  bn=cfisStatic { colNameList.add(new AlterTableStatementColumn(idn,vn,bn)); } )* ')' ) ) { type = AlterTableStatement.Type.ADD; }
+          | K_DROP (         id=cident  { colNameList.add(new AlterTableStatementColumn(id)); }
+                     | ('('  id1=cident { colNameList.add(new AlterTableStatementColumn(id1)); }
+                       ( ',' idn=cident { colNameList.add(new AlterTableStatementColumn(idn)); } )* ')') ) { type = AlterTableStatement.Type.DROP; }
           | K_WITH  properties[attrs]                 { type = AlterTableStatement.Type.OPTS; }
           | K_RENAME                                  { type = AlterTableStatement.Type.RENAME; }
                id1=cident K_TO toId1=cident { renames.put(id1, toId1); }
                ( K_AND idn=cident K_TO toIdn=cident { renames.put(idn, toIdn); } )*
           )
     {
-        $expr = new AlterTableStatement(cf, type, id, v, attrs, renames, isStatic);
+        $expr = new AlterTableStatement(cf, type, colNameList, attrs, renames);
+    }
+    ;
+
+cfisStatic returns [boolean isStaticColumn]
+    @init{
+        boolean isStatic = false;
+    }
+    : (K_STATIC { isStatic=true; })? { $isStaticColumn = isStatic;
     }
     ;
 
diff --git a/src/java/org/apache/cassandra/cql3/statements/AlterTableStatement.java b/src/java/org/apache/cassandra/cql3/statements/AlterTableStatement.java
index 3515c6bfdb..8ad3015b64 100644
--- a/src/java/org/apache/cassandra/cql3/statements/AlterTableStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/AlterTableStatement.java
@@ -52,27 +52,21 @@ public class AlterTableStatement extends SchemaAlteringStatement
     }
 
     public final Type oType;
-    public final CQL3Type.Raw validator;
-    public final ColumnIdentifier.Raw rawColumnName;
     private final TableAttributes attrs;
     private final Map<ColumnIdentifier.Raw, ColumnIdentifier.Raw> renames;
-    private final boolean isStatic; // Only for ALTER ADD
+    private final List<AlterTableStatementColumn> colNameList;
 
     public AlterTableStatement(CFName name,
                                Type type,
-                               ColumnIdentifier.Raw columnName,
-                               CQL3Type.Raw validator,
+                               List<AlterTableStatementColumn> colDataList,
                                TableAttributes attrs,
-                               Map<ColumnIdentifier.Raw, ColumnIdentifier.Raw> renames,
-                               boolean isStatic)
+                               Map<ColumnIdentifier.Raw, ColumnIdentifier.Raw> renames)
     {
         super(name);
         this.oType = type;
-        this.rawColumnName = columnName;
-        this.validator = validator; // used only for ADD/ALTER commands
+        this.colNameList = colDataList;
         this.attrs = attrs;
         this.renames = renames;
-        this.isStatic = isStatic;
     }
 
     public void checkAccess(ClientState state) throws UnauthorizedException, InvalidRequestException
@@ -92,15 +86,12 @@ public class AlterTableStatement extends SchemaAlteringStatement
             throw new InvalidRequestException("Cannot use ALTER TABLE on Materialized View");
 
         CFMetaData cfm = meta.copy();
-
-        CQL3Type validator = this.validator == null ? null : this.validator.prepare(keyspace());
         ColumnIdentifier columnName = null;
         ColumnDefinition def = null;
-        if (rawColumnName != null)
-        {
-            columnName = rawColumnName.prepare(cfm);
-            def = cfm.getColumnDefinition(columnName);
-        }
+        CQL3Type.Raw dataType = null;
+        boolean isStatic = false;
+        CQL3Type validator = null;
+        ColumnIdentifier.Raw rawColumnName = null;
 
         List<ViewDefinition> viewUpdates = null;
         Iterable<ViewDefinition> views = View.findAll(keyspace(), columnFamily());
@@ -108,81 +99,104 @@ public class AlterTableStatement extends SchemaAlteringStatement
         switch (oType)
         {
             case ADD:
-                assert columnName != null;
-                if (cfm.isDense())
-                    throw new InvalidRequestException("Cannot add new column to a COMPACT STORAGE table");
-
-                if (isStatic)
+                for (AlterTableStatementColumn colData : colNameList)
                 {
-                    if (!cfm.isCompound())
-                        throw new InvalidRequestException("Static columns are not allowed in COMPACT STORAGE tables");
-                    if (cfm.clusteringColumns().isEmpty())
-                        throw new InvalidRequestException("Static columns are only useful (and thus allowed) if the table has at least one clustering column");
-                }
-
-                if (def != null)
-                {
-                    switch (def.kind)
+                    columnName = null;
+                    rawColumnName = colData.getColumnName();
+                    if (rawColumnName != null)
                     {
-                        case PARTITION_KEY:
-                        case CLUSTERING:
-                            throw new InvalidRequestException(String.format("Invalid column name %s because it conflicts with a PRIMARY KEY part", columnName));
-                        default:
-                            throw new InvalidRequestException(String.format("Invalid column name %s because it conflicts with an existing column", columnName));
+                        columnName = rawColumnName.prepare(cfm);
+                        def =  cfm.getColumnDefinition(columnName);
+                        dataType = colData.getColumnType();
+                        isStatic = colData.getStaticType();
+                        validator = dataType == null ? null : dataType.prepare(keyspace());
                     }
-                }
 
-                // Cannot re-add a dropped counter column. See #7831.
-                if (meta.isCounter() && meta.getDroppedColumns().containsKey(columnName.bytes))
-                    throw new InvalidRequestException(String.format("Cannot re-add previously dropped counter column %s", columnName));
+                    assert columnName != null;
+                    if (cfm.isDense())
+                        throw new InvalidRequestException("Cannot add new column to a COMPACT STORAGE table");
 
-                AbstractType<?> type = validator.getType();
-                if (type.isCollection() && type.isMultiCell())
-                {
-                    if (!cfm.isCompound())
-                        throw new InvalidRequestException("Cannot use non-frozen collections in COMPACT STORAGE tables");
-                    if (cfm.isSuper())
-                        throw new InvalidRequestException("Cannot use non-frozen collections with super column families");
-
-                    // If there used to be a non-frozen collection column with the same name (that has been dropped),
-                    // we could still have some data using the old type, and so we can't allow adding a collection
-                    // with the same name unless the types are compatible (see #6276).
-                    CFMetaData.DroppedColumn dropped = cfm.getDroppedColumns().get(columnName.bytes);
-                    if (dropped != null && dropped.type instanceof CollectionType
-                        && dropped.type.isMultiCell() && !type.isCompatibleWith(dropped.type))
+                    if (isStatic)
                     {
-                        String message =
-                            String.format("Cannot add a collection with the name %s because a collection with the same name"
-                                          + " and a different type (%s) has already been used in the past",
-                                          columnName,
-                                          dropped.type.asCQL3Type());
-                        throw new InvalidRequestException(message);
+                        if (!cfm.isCompound())
+                            throw new InvalidRequestException("Static columns are not allowed in COMPACT STORAGE tables");
+                        if (cfm.clusteringColumns().isEmpty())
+                            throw new InvalidRequestException("Static columns are only useful (and thus allowed) if the table has at least one clustering column");
                     }
-                }
 
-                cfm.addColumnDefinition(isStatic
-                                        ? ColumnDefinition.staticDef(cfm, columnName.bytes, type)
-                                        : ColumnDefinition.regularDef(cfm, columnName.bytes, type));
-
-                // Adding a column to a table which has an include all view requires the column to be added to the view
-                // as well
-                if (!isStatic)
-                {
-                    for (ViewDefinition view : views)
+                    if (def != null)
                     {
-                        if (view.includeAllColumns)
+                        switch (def.kind)
                         {
-                            ViewDefinition viewCopy = view.copy();
-                            viewCopy.metadata.addColumnDefinition(ColumnDefinition.regularDef(viewCopy.metadata, columnName.bytes, type));
-                            if (viewUpdates == null)
-                                viewUpdates = new ArrayList<>();
-                            viewUpdates.add(viewCopy);
+                            case PARTITION_KEY:
+                            case CLUSTERING:
+                                throw new InvalidRequestException(String.format("Invalid column name %s because it conflicts with a PRIMARY KEY part", columnName));
+                            default:
+                                throw new InvalidRequestException(String.format("Invalid column name %s because it conflicts with an existing column", columnName));
+                        }
+                    }
+
+                    // Cannot re-add a dropped counter column. See #7831.
+                    if (meta.isCounter() && meta.getDroppedColumns().containsKey(columnName.bytes))
+                        throw new InvalidRequestException(String.format("Cannot re-add previously dropped counter column %s", columnName));
+
+                    AbstractType<?> type = validator.getType();
+                    if (type.isCollection() && type.isMultiCell())
+                    {
+                        if (!cfm.isCompound())
+                            throw new InvalidRequestException("Cannot use non-frozen collections in COMPACT STORAGE tables");
+                        if (cfm.isSuper())
+                            throw new InvalidRequestException("Cannot use non-frozen collections with super column families");
+
+                        // If there used to be a non-frozen collection column with the same name (that has been dropped),
+                        // we could still have some data using the old type, and so we can't allow adding a collection
+                        // with the same name unless the types are compatible (see #6276).
+                        CFMetaData.DroppedColumn dropped = cfm.getDroppedColumns().get(columnName.bytes);
+                        if (dropped != null && dropped.type instanceof CollectionType
+                            && dropped.type.isMultiCell() && !type.isCompatibleWith(dropped.type))
+                        {
+                            String message =
+                                String.format("Cannot add a collection with the name %s because a collection with the same name"
+                                              + " and a different type (%s) has already been used in the past",
+                                              columnName,
+                                              dropped.type.asCQL3Type());
+                            throw new InvalidRequestException(message);
+                        }
+                    }
+
+                    cfm.addColumnDefinition(isStatic
+                                            ? ColumnDefinition.staticDef(cfm, columnName.bytes, type)
+                                            : ColumnDefinition.regularDef(cfm, columnName.bytes, type));
+
+                    // Adding a column to a table which has an include all view requires the column to be added to the view
+                    // as well
+                    if (!isStatic)
+                    {
+                        for (ViewDefinition view : views)
+                        {
+                            if (view.includeAllColumns)
+                            {
+                                ViewDefinition viewCopy = view.copy();
+                                viewCopy.metadata.addColumnDefinition(ColumnDefinition.regularDef(viewCopy.metadata, columnName.bytes, type));
+                                if (viewUpdates == null)
+                                    viewUpdates = new ArrayList<>();
+                                viewUpdates.add(viewCopy);
+                            }
                         }
                     }
                 }
                 break;
 
             case ALTER:
+                rawColumnName = colNameList.get(0).getColumnName();
+                if (rawColumnName != null)
+                {
+                    columnName = rawColumnName.prepare(cfm);
+                    def = cfm.getColumnDefinition(columnName);
+                    dataType = colNameList.get(0).getColumnType();
+                    validator = dataType == null ? null : dataType.prepare(keyspace());
+                }
+
                 assert columnName != null;
                 if (def == null)
                     throw new InvalidRequestException(String.format("Column %s was not found in table %s", columnName, columnFamily()));
@@ -214,66 +228,76 @@ public class AlterTableStatement extends SchemaAlteringStatement
                 break;
 
             case DROP:
-                assert columnName != null;
-                if (!cfm.isCQLTable())
-                    throw new InvalidRequestException("Cannot drop columns from a non-CQL3 table");
-                if (def == null)
-                    throw new InvalidRequestException(String.format("Column %s was not found in table %s", columnName, columnFamily()));
-
-                switch (def.kind)
+                for (AlterTableStatementColumn colData : colNameList)
                 {
-                    case PARTITION_KEY:
-                    case CLUSTERING:
-                        throw new InvalidRequestException(String.format("Cannot drop PRIMARY KEY part %s", columnName));
-                    case REGULAR:
-                    case STATIC:
-                        ColumnDefinition toDelete = null;
-                        for (ColumnDefinition columnDef : cfm.partitionColumns())
-                        {
-                            if (columnDef.name.equals(columnName))
-                            {
-                                toDelete = columnDef;
-                                break;
-                            }
-                        }
+                    columnName = null;
+                    rawColumnName = colData.getColumnName();
+                    if (rawColumnName != null)
+                    {
+                        columnName = rawColumnName.prepare(cfm);
+                        def = cfm.getColumnDefinition(columnName);
+                    }
+                    assert columnName != null;
+                    if (!cfm.isCQLTable())
+                        throw new InvalidRequestException("Cannot drop columns from a non-CQL3 table");
+                    if (def == null)
+                        throw new InvalidRequestException(String.format("Column %s was not found in table %s", columnName, columnFamily()));
+
+                    switch (def.kind)
+                    {
+                         case PARTITION_KEY:
+                         case CLUSTERING:
+                              throw new InvalidRequestException(String.format("Cannot drop PRIMARY KEY part %s", columnName));
+                         case REGULAR:
+                         case STATIC:
+                              ColumnDefinition toDelete = null;
+                              for (ColumnDefinition columnDef : cfm.partitionColumns())
+                              {
+                                   if (columnDef.name.equals(columnName))
+                                      {
+                                        toDelete = columnDef;
+                                        break;
+                                      }
+                               }
                         assert toDelete != null;
                         cfm.removeColumnDefinition(toDelete);
                         cfm.recordColumnDrop(toDelete);
                         break;
-                }
+                    }
 
-                // If the dropped column is required by any secondary indexes
-                // we reject the operation, as the indexes must be dropped first
-                Indexes allIndexes = cfm.getIndexes();
-                if (!allIndexes.isEmpty())
-                {
-                    ColumnFamilyStore store = Keyspace.openAndGetStore(cfm);
-                    Set<IndexMetadata> dependentIndexes = store.indexManager.getDependentIndexes(def);
-                    if (!dependentIndexes.isEmpty())
-                        throw new InvalidRequestException(String.format("Cannot drop column %s because it has " +
-                                                                        "dependent secondary indexes (%s)",
-                                                                        def,
-                                                                        dependentIndexes.stream()
-                                                                                        .map(i -> i.name)
-                                                                                        .collect(Collectors.joining(","))));
-                }
+                    // If the dropped column is required by any secondary indexes
+                    // we reject the operation, as the indexes must be dropped first
+                    Indexes allIndexes = cfm.getIndexes();
+                    if (!allIndexes.isEmpty())
+                    {
+                        ColumnFamilyStore store = Keyspace.openAndGetStore(cfm);
+                        Set<IndexMetadata> dependentIndexes = store.indexManager.getDependentIndexes(def);
+                        if (!dependentIndexes.isEmpty())
+                            throw new InvalidRequestException(String.format("Cannot drop column %s because it has " +
+                                                                            "dependent secondary indexes (%s)",
+                                                                            def,
+                                                                            dependentIndexes.stream()
+                                                                                            .map(i -> i.name)
+                                                                                            .collect(Collectors.joining(","))));
+                    }
 
-                // If a column is dropped which is included in a view, we don't allow the drop to take place.
-                boolean rejectAlter = false;
-                StringBuilder builder = new StringBuilder();
-                for (ViewDefinition view : views)
-                {
-                    if (!view.includes(columnName)) continue;
+                    // If a column is dropped which is included in a view, we don't allow the drop to take place.
+                    boolean rejectAlter = false;
+                    StringBuilder builder = new StringBuilder();
+                    for (ViewDefinition view : views)
+                    {
+                        if (!view.includes(columnName)) continue;
+                        if (rejectAlter)
+                            builder.append(',');
+                        rejectAlter = true;
+                        builder.append(view.viewName);
+                    }
                     if (rejectAlter)
-                        builder.append(',');
-                    rejectAlter = true;
-                    builder.append(view.viewName);
+                        throw new InvalidRequestException(String.format("Cannot drop column %s, depended on by materialized views (%s.{%s})",
+                                                                        columnName.toString(),
+                                                                        keyspace(),
+                                                                        builder.toString()));
                 }
-                if (rejectAlter)
-                    throw new InvalidRequestException(String.format("Cannot drop column %s, depended on by materialized views (%s.{%s})",
-                                                                    columnName.toString(),
-                                                                    keyspace(),
-                                                                    builder.toString()));
                 break;
             case OPTS:
                 if (attrs == null)
@@ -379,12 +403,11 @@ public class AlterTableStatement extends SchemaAlteringStatement
         }
     }
 
+    @Override
     public String toString()
     {
         return String.format("AlterTableStatement(name=%s, type=%s, column=%s, validator=%s)",
                              cfName,
-                             oType,
-                             rawColumnName,
-                             validator);
+                             oType);
     }
 }
diff --git a/src/java/org/apache/cassandra/cql3/statements/AlterTableStatementColumn.java b/src/java/org/apache/cassandra/cql3/statements/AlterTableStatementColumn.java
new file mode 100644
index 0000000000..a5d7de7036
--- /dev/null
+++ b/src/java/org/apache/cassandra/cql3/statements/AlterTableStatementColumn.java
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.cassandra.cql3.statements;
+
+import org.apache.cassandra.cql3.CQL3Type;
+import org.apache.cassandra.cql3.ColumnIdentifier;
+
+public class AlterTableStatementColumn {
+    private final CQL3Type.Raw dataType;
+    private final ColumnIdentifier.Raw colName;
+    private final Boolean isStatic;
+
+    public AlterTableStatementColumn(ColumnIdentifier.Raw colName, CQL3Type.Raw dataType, boolean isStatic) {
+        this.dataType = dataType;
+        this.colName = colName;
+        this.isStatic = isStatic;
+    }
+
+    public AlterTableStatementColumn(ColumnIdentifier.Raw colName, CQL3Type.Raw dataType) {
+        this(colName, dataType,false );
+    }
+
+    public AlterTableStatementColumn(ColumnIdentifier.Raw colName) {
+        this(colName, null, false);
+    }
+
+    public CQL3Type.Raw getColumnType() {
+        return dataType;
+    }
+
+    public ColumnIdentifier.Raw getColumnName() {
+        return colName;
+    }
+
+    public Boolean getStaticType() {
+        return isStatic;
+    }
+}
diff --git a/test/unit/org/apache/cassandra/cql3/validation/operations/AlterTest.java b/test/unit/org/apache/cassandra/cql3/validation/operations/AlterTest.java
index 9f8bea2902..3eb55fd080 100644
--- a/test/unit/org/apache/cassandra/cql3/validation/operations/AlterTest.java
+++ b/test/unit/org/apache/cassandra/cql3/validation/operations/AlterTest.java
@@ -324,4 +324,85 @@ public class AlterTest extends CQLTester
         createTable("CREATE TABLE %s (key blob, column1 blob, value blob, PRIMARY KEY ((key), column1)) WITH COMPACT STORAGE");
         assertInvalidThrow(InvalidRequestException.class, "ALTER TABLE %s ALTER column1 TYPE ascii");
     }
+
+    /*
+     * Test case to check addition of one column
+    */
+    @Test
+    public void testAlterAddOneColumn() throws Throwable
+    {
+        createTable("CREATE TABLE IF NOT EXISTS %s (id int, name text, PRIMARY KEY (id))");
+        alterTable("ALTER TABLE %s add mail text;");
+
+        assertColumnNames(execute("SELECT * FROM %s"), "id", "mail", "name");
+    }
+
+    /*
+     * Test case to check addition of more than one column
+     */
+    @Test
+    public void testAlterAddMultiColumn() throws Throwable
+    {
+        createTable("CREATE TABLE IF NOT EXISTS %s (id int, yearofbirth int, PRIMARY KEY (id))");
+        alterTable("ALTER TABLE %s add (firstname text, password blob, lastname text, \"SOME escaped col\" bigint)");
+
+        assertColumnNames(execute("SELECT * FROM %s"), "id", "SOME escaped col", "firstname", "lastname", "password", "yearofbirth");
+    }
+
+    /*
+     *  Should throw SyntaxException if multiple columns are added using wrong syntax.
+     *  Expected Syntax : Alter table T1 add (C1 datatype,C2 datatype,C3 datatype)
+     */
+    @Test(expected = SyntaxException.class)
+    public void testAlterAddMultiColumnWithoutBraces() throws Throwable
+    {
+        execute("ALTER TABLE %s.users add lastname text, password blob, yearofbirth int;");
+    }
+
+    /*
+     *  Test case to check deletion of one column
+     */
+    @Test
+    public void testAlterDropOneColumn() throws Throwable
+    {
+        createTable("CREATE TABLE IF NOT EXISTS %s (id text, telephone int, yearofbirth int, PRIMARY KEY (id))");
+        alterTable("ALTER TABLE %s drop telephone");
+
+        assertColumnNames(execute("SELECT * FROM %s"), "id", "yearofbirth");
+    }
+
+    @Test
+    /*
+     * Test case to check deletion of more than one column
+     */
+    public void testAlterDropMultiColumn() throws Throwable
+    {
+        createTable("CREATE TABLE IF NOT EXISTS %s (id text, address text, telephone int, yearofbirth int, \"SOME escaped col\" bigint, PRIMARY KEY (id))");
+        alterTable("ALTER TABLE %s drop (address, telephone, \"SOME escaped col\");");
+
+        assertColumnNames(execute("SELECT * FROM %s"), "id", "yearofbirth");
+    }
+
+    /*
+     *  Should throw SyntaxException if multiple columns are dropped using wrong syntax.
+     */
+    @Test(expected = SyntaxException.class)
+    public void testAlterDeletionColumnWithoutBraces() throws Throwable
+    {
+        execute("ALTER TABLE %s.users drop name,address;");
+    }
+
+    @Test(expected = InvalidRequestException.class)
+    public void testAlterAddDuplicateColumn() throws Throwable
+    {
+        createTable("CREATE TABLE IF NOT EXISTS %s (id text, address text, telephone int, yearofbirth int, PRIMARY KEY (id))");
+        execute("ALTER TABLE %s add (salary int, salary int);");
+    }
+
+    @Test(expected = InvalidRequestException.class)
+    public void testAlterDropDuplicateColumn() throws Throwable
+    {
+        createTable("CREATE TABLE IF NOT EXISTS %s (id text, address text, telephone int, yearofbirth int, PRIMARY KEY (id))");
+        execute("ALTER TABLE %s drop (address, address);");
+    }
 }