diff --git a/core/src/integration-test/java/com/scalar/db/storage/cosmos/ConsensusCommitAdminIntegrationTestWithCosmos.java b/core/src/integration-test/java/com/scalar/db/storage/cosmos/ConsensusCommitAdminIntegrationTestWithCosmos.java
index b372b419fb..20d061d508 100644
--- a/core/src/integration-test/java/com/scalar/db/storage/cosmos/ConsensusCommitAdminIntegrationTestWithCosmos.java
+++ b/core/src/integration-test/java/com/scalar/db/storage/cosmos/ConsensusCommitAdminIntegrationTestWithCosmos.java
@@ -4,6 +4,7 @@
 import com.scalar.db.util.AdminTestUtils;
 import java.util.Map;
 import java.util.Properties;
+import org.junit.jupiter.api.Disabled;
 
 public class ConsensusCommitAdminIntegrationTestWithCosmos
     extends ConsensusCommitAdminIntegrationTestBase {
@@ -22,4 +23,28 @@ protected Map<String, String> getCreationOptions() {
   protected AdminTestUtils getAdminTestUtils(String testName) {
     return new CosmosAdminTestUtils(getProperties(testName));
   }
+
+  @Override
+  @Disabled("Cosmos DB does not support dropping columns")
+  public void dropColumnFromTable_DropColumnForEachExistingDataType_ShouldDropColumnsCorrectly() {}
+
+  @Override
+  @Disabled("Cosmos DB does not support dropping columns")
+  public void dropColumnFromTable_ForNonExistingTable_ShouldThrowIllegalArgumentException() {}
+
+  @Override
+  @Disabled("Cosmos DB does not support dropping columns")
+  public void dropColumnFromTable_ForNonExistingColumn_ShouldThrowIllegalArgumentException() {}
+
+  @Override
+  @Disabled("Cosmos DB does not support dropping columns")
+  public void dropColumnFromTable_ForPrimaryKeyColumn_ShouldThrowIllegalArgumentException() {}
+
+  @Disabled("Cosmos DB does not support dropping columns")
+  @Override
+  public void dropColumnFromTable_ForIndexedColumn_ShouldDropColumnAndIndexCorrectly() {}
+
+  @Override
+  @Disabled("Cosmos DB does not support dropping columns")
+  public void dropColumnFromTable_IfNotExists_ForNonExistingColumn_ShouldNotThrowAnyException() {}
 }
diff --git a/core/src/integration-test/java/com/scalar/db/storage/cosmos/CosmosAdminCaseSensitivityIntegrationTest.java b/core/src/integration-test/java/com/scalar/db/storage/cosmos/CosmosAdminCaseSensitivityIntegrationTest.java
index 4048c82c9c..e26cd4295e 100644
--- a/core/src/integration-test/java/com/scalar/db/storage/cosmos/CosmosAdminCaseSensitivityIntegrationTest.java
+++ b/core/src/integration-test/java/com/scalar/db/storage/cosmos/CosmosAdminCaseSensitivityIntegrationTest.java
@@ -4,6 +4,7 @@
 import com.scalar.db.util.AdminTestUtils;
 import java.util.Map;
 import java.util.Properties;
+import org.junit.jupiter.api.Disabled;
 
 public class CosmosAdminCaseSensitivityIntegrationTest
     extends DistributedStorageAdminCaseSensitivityIntegrationTestBase {
@@ -22,4 +23,28 @@ protected Map<String, String> getCreationOptions() {
   protected AdminTestUtils getAdminTestUtils(String testName) {
     return new CosmosAdminTestUtils(getProperties(testName));
   }
+
+  @Disabled("Cosmos DB does not support dropping columns")
+  @Override
+  public void dropColumnFromTable_DropColumnForEachExistingDataType_ShouldDropColumnsCorrectly() {}
+
+  @Disabled("Cosmos DB does not support dropping columns")
+  @Override
+  public void dropColumnFromTable_ForNonExistingTable_ShouldThrowIllegalArgumentException() {}
+
+  @Disabled("Cosmos DB does not support dropping columns")
+  @Override
+  public void dropColumnFromTable_ForNonExistingColumn_ShouldThrowIllegalArgumentException() {}
+
+  @Disabled("Cosmos DB does not support dropping columns")
+  @Override
+  public void dropColumnFromTable_ForPrimaryKeyColumn_ShouldThrowIllegalArgumentException() {}
+
+  @Disabled("Cosmos DB does not support dropping columns")
+  @Override
+  public void dropColumnFromTable_ForIndexedColumn_ShouldDropColumnAndIndexCorrectly() {}
+
+  @Disabled("Cosmos DB does not support dropping columns")
+  @Override
+  public void dropColumnFromTable_IfExists_ForNonExistingColumn_ShouldNotThrowAnyException() {}
 }
diff --git a/core/src/integration-test/java/com/scalar/db/storage/cosmos/CosmosAdminIntegrationTest.java b/core/src/integration-test/java/com/scalar/db/storage/cosmos/CosmosAdminIntegrationTest.java
index 03c2fbf59a..6c3aaf948c 100644
--- a/core/src/integration-test/java/com/scalar/db/storage/cosmos/CosmosAdminIntegrationTest.java
+++ b/core/src/integration-test/java/com/scalar/db/storage/cosmos/CosmosAdminIntegrationTest.java
@@ -4,6 +4,7 @@
 import com.scalar.db.util.AdminTestUtils;
 import java.util.Map;
 import java.util.Properties;
+import org.junit.jupiter.api.Disabled;
 
 public class CosmosAdminIntegrationTest extends DistributedStorageAdminIntegrationTestBase {
 
@@ -21,4 +22,28 @@ protected Map<String, String> getCreationOptions() {
   protected AdminTestUtils getAdminTestUtils(String testName) {
     return new CosmosAdminTestUtils(getProperties(testName));
   }
+
+  @Disabled("Cosmos DB does not support dropping columns")
+  @Override
+  public void dropColumnFromTable_DropColumnForEachExistingDataType_ShouldDropColumnsCorrectly() {}
+
+  @Disabled("Cosmos DB does not support dropping columns")
+  @Override
+  public void dropColumnFromTable_ForNonExistingTable_ShouldThrowIllegalArgumentException() {}
+
+  @Disabled("Cosmos DB does not support dropping columns")
+  @Override
+  public void dropColumnFromTable_ForNonExistingColumn_ShouldThrowIllegalArgumentException() {}
+
+  @Disabled("Cosmos DB does not support dropping columns")
+  @Override
+  public void dropColumnFromTable_ForPrimaryKeyColumn_ShouldThrowIllegalArgumentException() {}
+
+  @Disabled("Cosmos DB does not support dropping columns")
+  @Override
+  public void dropColumnFromTable_ForIndexedColumn_ShouldDropColumnAndIndexCorrectly() {}
+
+  @Disabled("Cosmos DB does not support dropping columns")
+  @Override
+  public void dropColumnFromTable_IfExists_ForNonExistingColumn_ShouldNotThrowAnyException() {}
 }
diff --git a/core/src/integration-test/java/com/scalar/db/storage/cosmos/SingleCrudOperationTransactionAdminIntegrationTestWithCosmos.java b/core/src/integration-test/java/com/scalar/db/storage/cosmos/SingleCrudOperationTransactionAdminIntegrationTestWithCosmos.java
index e9b37a9ea4..96deded7bf 100644
--- a/core/src/integration-test/java/com/scalar/db/storage/cosmos/SingleCrudOperationTransactionAdminIntegrationTestWithCosmos.java
+++ b/core/src/integration-test/java/com/scalar/db/storage/cosmos/SingleCrudOperationTransactionAdminIntegrationTestWithCosmos.java
@@ -3,6 +3,7 @@
 import com.scalar.db.transaction.singlecrudoperation.SingleCrudOperationTransactionAdminIntegrationTestBase;
 import java.util.Map;
 import java.util.Properties;
+import org.junit.jupiter.api.Disabled;
 
 public class SingleCrudOperationTransactionAdminIntegrationTestWithCosmos
     extends SingleCrudOperationTransactionAdminIntegrationTestBase {
@@ -16,4 +17,28 @@ protected Properties getProps(String testName) {
   protected Map<String, String> getCreationOptions() {
     return CosmosEnv.getCreationOptions();
   }
+
+  @Override
+  @Disabled("Cosmos DB does not support dropping columns")
+  public void dropColumnFromTable_DropColumnForEachExistingDataType_ShouldDropColumnsCorrectly() {}
+
+  @Override
+  @Disabled("Cosmos DB does not support dropping columns")
+  public void dropColumnFromTable_ForNonExistingTable_ShouldThrowIllegalArgumentException() {}
+
+  @Override
+  @Disabled("Cosmos DB does not support dropping columns")
+  public void dropColumnFromTable_ForNonExistingColumn_ShouldThrowIllegalArgumentException() {}
+
+  @Override
+  @Disabled("Cosmos DB does not support dropping columns")
+  public void dropColumnFromTable_ForPrimaryKeyColumn_ShouldThrowIllegalArgumentException() {}
+
+  @Disabled("Cosmos DB does not support dropping columns")
+  @Override
+  public void dropColumnFromTable_ForIndexedColumn_ShouldDropColumnAndIndexCorrectly() {}
+
+  @Override
+  @Disabled("Cosmos DB does not support dropping columns")
+  public void dropColumnFromTable_IfNotExists_ForNonExistingColumn_ShouldNotThrowAnyException() {}
 }
diff --git a/core/src/integration-test/java/com/scalar/db/storage/dynamo/ConsensusCommitAdminIntegrationTestWithDynamo.java b/core/src/integration-test/java/com/scalar/db/storage/dynamo/ConsensusCommitAdminIntegrationTestWithDynamo.java
index 0269fc8f11..b6163a00e1 100644
--- a/core/src/integration-test/java/com/scalar/db/storage/dynamo/ConsensusCommitAdminIntegrationTestWithDynamo.java
+++ b/core/src/integration-test/java/com/scalar/db/storage/dynamo/ConsensusCommitAdminIntegrationTestWithDynamo.java
@@ -4,6 +4,7 @@
 import com.scalar.db.util.AdminTestUtils;
 import java.util.Map;
 import java.util.Properties;
+import org.junit.jupiter.api.Disabled;
 
 public class ConsensusCommitAdminIntegrationTestWithDynamo
     extends ConsensusCommitAdminIntegrationTestBase {
@@ -27,4 +28,28 @@ protected boolean isIndexOnBooleanColumnSupported() {
   protected AdminTestUtils getAdminTestUtils(String testName) {
     return new DynamoAdminTestUtils(getProperties(testName));
   }
+
+  @Override
+  @Disabled("DynamoDB does not support dropping columns")
+  public void dropColumnFromTable_DropColumnForEachExistingDataType_ShouldDropColumnsCorrectly() {}
+
+  @Override
+  @Disabled("DynamoDB does not support dropping columns")
+  public void dropColumnFromTable_ForNonExistingTable_ShouldThrowIllegalArgumentException() {}
+
+  @Override
+  @Disabled("DynamoDB does not support dropping columns")
+  public void dropColumnFromTable_ForNonExistingColumn_ShouldThrowIllegalArgumentException() {}
+
+  @Override
+  @Disabled("DynamoDB does not support dropping columns")
+  public void dropColumnFromTable_ForPrimaryKeyColumn_ShouldThrowIllegalArgumentException() {}
+
+  @Disabled("DynamoDB does not support dropping columns")
+  @Override
+  public void dropColumnFromTable_ForIndexedColumn_ShouldDropColumnAndIndexCorrectly() {}
+
+  @Override
+  @Disabled("DynamoDB does not support dropping columns")
+  public void dropColumnFromTable_IfNotExists_ForNonExistingColumn_ShouldNotThrowAnyException() {}
 }
diff --git a/core/src/integration-test/java/com/scalar/db/storage/dynamo/DynamoAdminCaseSensitivityIntegrationTest.java b/core/src/integration-test/java/com/scalar/db/storage/dynamo/DynamoAdminCaseSensitivityIntegrationTest.java
index 94b7fa1d21..12d673135d 100644
--- a/core/src/integration-test/java/com/scalar/db/storage/dynamo/DynamoAdminCaseSensitivityIntegrationTest.java
+++ b/core/src/integration-test/java/com/scalar/db/storage/dynamo/DynamoAdminCaseSensitivityIntegrationTest.java
@@ -4,6 +4,7 @@
 import com.scalar.db.util.AdminTestUtils;
 import java.util.Map;
 import java.util.Properties;
+import org.junit.jupiter.api.Disabled;
 
 public class DynamoAdminCaseSensitivityIntegrationTest
     extends DistributedStorageAdminCaseSensitivityIntegrationTestBase {
@@ -27,4 +28,28 @@ protected boolean isIndexOnBooleanColumnSupported() {
   protected AdminTestUtils getAdminTestUtils(String testName) {
     return new DynamoAdminTestUtils(getProperties(testName));
   }
+
+  @Disabled("DynamoDB does not support dropping columns")
+  @Override
+  public void dropColumnFromTable_DropColumnForEachExistingDataType_ShouldDropColumnsCorrectly() {}
+
+  @Disabled("DynamoDB does not support dropping columns")
+  @Override
+  public void dropColumnFromTable_ForNonExistingTable_ShouldThrowIllegalArgumentException() {}
+
+  @Disabled("DynamoDB does not support dropping columns")
+  @Override
+  public void dropColumnFromTable_ForNonExistingColumn_ShouldThrowIllegalArgumentException() {}
+
+  @Disabled("DynamoDB does not support dropping columns")
+  @Override
+  public void dropColumnFromTable_ForPrimaryKeyColumn_ShouldThrowIllegalArgumentException() {}
+
+  @Disabled("DynamoDB does not support dropping columns")
+  @Override
+  public void dropColumnFromTable_ForIndexedColumn_ShouldDropColumnAndIndexCorrectly() {}
+
+  @Disabled("DynamoDB does not support dropping columns")
+  @Override
+  public void dropColumnFromTable_IfExists_ForNonExistingColumn_ShouldNotThrowAnyException() {}
 }
diff --git a/core/src/integration-test/java/com/scalar/db/storage/dynamo/DynamoAdminIntegrationTest.java b/core/src/integration-test/java/com/scalar/db/storage/dynamo/DynamoAdminIntegrationTest.java
index 938c9f0a20..c86fd33073 100644
--- a/core/src/integration-test/java/com/scalar/db/storage/dynamo/DynamoAdminIntegrationTest.java
+++ b/core/src/integration-test/java/com/scalar/db/storage/dynamo/DynamoAdminIntegrationTest.java
@@ -4,6 +4,7 @@
 import com.scalar.db.util.AdminTestUtils;
 import java.util.Map;
 import java.util.Properties;
+import org.junit.jupiter.api.Disabled;
 
 public class DynamoAdminIntegrationTest extends DistributedStorageAdminIntegrationTestBase {
 
@@ -26,4 +27,28 @@ protected boolean isIndexOnBooleanColumnSupported() {
   protected AdminTestUtils getAdminTestUtils(String testName) {
     return new DynamoAdminTestUtils(getProperties(testName));
   }
+
+  @Disabled("DynamoDB does not support dropping columns")
+  @Override
+  public void dropColumnFromTable_DropColumnForEachExistingDataType_ShouldDropColumnsCorrectly() {}
+
+  @Disabled("DynamoDB does not support dropping columns")
+  @Override
+  public void dropColumnFromTable_ForNonExistingTable_ShouldThrowIllegalArgumentException() {}
+
+  @Disabled("DynamoDB does not support dropping columns")
+  @Override
+  public void dropColumnFromTable_ForNonExistingColumn_ShouldThrowIllegalArgumentException() {}
+
+  @Disabled("DynamoDB does not support dropping columns")
+  @Override
+  public void dropColumnFromTable_ForPrimaryKeyColumn_ShouldThrowIllegalArgumentException() {}
+
+  @Disabled("DynamoDB does not support dropping columns")
+  @Override
+  public void dropColumnFromTable_ForIndexedColumn_ShouldDropColumnAndIndexCorrectly() {}
+
+  @Disabled("DynamoDB does not support dropping columns")
+  @Override
+  public void dropColumnFromTable_IfExists_ForNonExistingColumn_ShouldNotThrowAnyException() {}
 }
diff --git a/core/src/integration-test/java/com/scalar/db/storage/dynamo/DynamoAdminPermissionIntegrationTest.java b/core/src/integration-test/java/com/scalar/db/storage/dynamo/DynamoAdminPermissionIntegrationTest.java
index b2fcd6aa5e..2879b2e5ee 100644
--- a/core/src/integration-test/java/com/scalar/db/storage/dynamo/DynamoAdminPermissionIntegrationTest.java
+++ b/core/src/integration-test/java/com/scalar/db/storage/dynamo/DynamoAdminPermissionIntegrationTest.java
@@ -59,4 +59,9 @@ public void addRawColumnToTable_WithSufficientPermission_ShouldSucceed() {}
   @Override
   @Disabled("Import-related functionality is not supported in DynamoDB")
   public void importTable_WithSufficientPermission_ShouldSucceed() {}
+
+  @Test
+  @Override
+  @Disabled("DynamoDB does not support dropping columns")
+  public void dropColumnFromTable_WithSufficientPermission_ShouldSucceed() {}
 }
diff --git a/core/src/integration-test/java/com/scalar/db/storage/dynamo/SingleCrudOperationTransactionAdminIntegrationTestWithDynamo.java b/core/src/integration-test/java/com/scalar/db/storage/dynamo/SingleCrudOperationTransactionAdminIntegrationTestWithDynamo.java
index 7fa7d490e7..83dc86eef4 100644
--- a/core/src/integration-test/java/com/scalar/db/storage/dynamo/SingleCrudOperationTransactionAdminIntegrationTestWithDynamo.java
+++ b/core/src/integration-test/java/com/scalar/db/storage/dynamo/SingleCrudOperationTransactionAdminIntegrationTestWithDynamo.java
@@ -3,6 +3,7 @@
 import com.scalar.db.transaction.singlecrudoperation.SingleCrudOperationTransactionAdminIntegrationTestBase;
 import java.util.Map;
 import java.util.Properties;
+import org.junit.jupiter.api.Disabled;
 
 public class SingleCrudOperationTransactionAdminIntegrationTestWithDynamo
     extends SingleCrudOperationTransactionAdminIntegrationTestBase {
@@ -21,4 +22,28 @@ protected Map<String, String> getCreationOptions() {
   protected boolean isIndexOnBooleanColumnSupported() {
     return false;
   }
+
+  @Override
+  @Disabled("DynamoDB does not support dropping columns")
+  public void dropColumnFromTable_DropColumnForEachExistingDataType_ShouldDropColumnsCorrectly() {}
+
+  @Override
+  @Disabled("DynamoDB does not support dropping columns")
+  public void dropColumnFromTable_ForNonExistingTable_ShouldThrowIllegalArgumentException() {}
+
+  @Override
+  @Disabled("DynamoDB does not support dropping columns")
+  public void dropColumnFromTable_ForNonExistingColumn_ShouldThrowIllegalArgumentException() {}
+
+  @Override
+  @Disabled("DynamoDB does not support dropping columns")
+  public void dropColumnFromTable_ForPrimaryKeyColumn_ShouldThrowIllegalArgumentException() {}
+
+  @Disabled("DynamoDB does not support dropping columns")
+  @Override
+  public void dropColumnFromTable_ForIndexedColumn_ShouldDropColumnAndIndexCorrectly() {}
+
+  @Override
+  @Disabled("DynamoDB does not support dropping columns")
+  public void dropColumnFromTable_IfNotExists_ForNonExistingColumn_ShouldNotThrowAnyException() {}
 }
diff --git a/core/src/main/java/com/scalar/db/api/Admin.java b/core/src/main/java/com/scalar/db/api/Admin.java
index dc7fa43a0e..b309c2ec38 100644
--- a/core/src/main/java/com/scalar/db/api/Admin.java
+++ b/core/src/main/java/com/scalar/db/api/Admin.java
@@ -479,6 +479,47 @@ default void addNewColumnToTable(
     addNewColumnToTable(namespace, table, columnName, columnType);
   }
 
+  /**
+   * Drops a column from an existing table. The column cannot be a partition key or a clustering
+   * key.
+   *
+   * @param namespace the table namespace
+   * @param table the table name
+   * @param columnName the name of the column to drop
+   * @throws IllegalArgumentException if the table or column does not exist, or the column is a
+   *     partition key column or clustering key column
+   * @throws ExecutionException if the operation fails
+   */
+  void dropColumnFromTable(String namespace, String table, String columnName)
+      throws ExecutionException;
+
+  /**
+   * Drops a column from an existing table. The column cannot be a partition key or a clustering
+   * key.
+   *
+   * @param namespace the table namespace
+   * @param table the table name
+   * @param columnName the name of the column to drop
+   * @param IfExists if set to true, the column will be dropped only if it exists. If set to false,
+   *     it will throw an exception if it does not exist
+   * @throws IllegalArgumentException if the table does not exist, or the column is a partition key
+   *     column or clustering key column
+   * @throws ExecutionException if the operation fails
+   */
+  default void dropColumnFromTable(
+      String namespace, String table, String columnName, boolean IfExists)
+      throws ExecutionException {
+    TableMetadata tableMetadata = getTableMetadata(namespace, table);
+    if (tableMetadata == null) {
+      throw new IllegalArgumentException(
+          CoreError.TABLE_NOT_FOUND.buildMessage(ScalarDbUtils.getFullTableName(namespace, table)));
+    }
+    if (IfExists && !tableMetadata.getColumnNames().contains(columnName)) {
+      return;
+    }
+    dropColumnFromTable(namespace, table, columnName);
+  }
+
   /**
    * Imports an existing table that is not managed by ScalarDB.
    *
diff --git a/core/src/main/java/com/scalar/db/common/CheckedDistributedStorageAdmin.java b/core/src/main/java/com/scalar/db/common/CheckedDistributedStorageAdmin.java
index 0786f02456..afc4592b64 100644
--- a/core/src/main/java/com/scalar/db/common/CheckedDistributedStorageAdmin.java
+++ b/core/src/main/java/com/scalar/db/common/CheckedDistributedStorageAdmin.java
@@ -295,6 +295,42 @@ public void addNewColumnToTable(
     }
   }
 
+  @Override
+  public void dropColumnFromTable(String namespace, String table, String columnName)
+      throws ExecutionException {
+    TableMetadata tableMetadata = getTableMetadata(namespace, table);
+    if (tableMetadata == null) {
+      throw new IllegalArgumentException(
+          CoreError.TABLE_NOT_FOUND.buildMessage(ScalarDbUtils.getFullTableName(namespace, table)));
+    }
+
+    if (!tableMetadata.getColumnNames().contains(columnName)) {
+      throw new IllegalArgumentException(
+          CoreError.COLUMN_NOT_FOUND2.buildMessage(
+              ScalarDbUtils.getFullTableName(namespace, table), columnName));
+    }
+
+    if (tableMetadata.getPartitionKeyNames().contains(columnName)
+        || tableMetadata.getClusteringKeyNames().contains(columnName)) {
+      throw new IllegalArgumentException(
+          CoreError.DROP_PRIMARY_KEY_COLUMN_NOT_SUPPORTED.buildMessage(
+              ScalarDbUtils.getFullTableName(namespace, table), columnName));
+    }
+
+    if (tableMetadata.getSecondaryIndexNames().contains(columnName)) {
+      dropIndex(namespace, table, columnName);
+    }
+
+    try {
+      admin.dropColumnFromTable(namespace, table, columnName);
+    } catch (ExecutionException e) {
+      throw new ExecutionException(
+          CoreError.DROPPING_COLUMN_FROM_TABLE_FAILED.buildMessage(
+              ScalarDbUtils.getFullTableName(namespace, table), columnName),
+          e);
+    }
+  }
+
   @Override
   public Set<String> getNamespaceNames() throws ExecutionException {
     try {
diff --git a/core/src/main/java/com/scalar/db/common/CoreError.java b/core/src/main/java/com/scalar/db/common/CoreError.java
index b4e5fbe682..c187941fd7 100644
--- a/core/src/main/java/com/scalar/db/common/CoreError.java
+++ b/core/src/main/java/com/scalar/db/common/CoreError.java
@@ -678,6 +678,20 @@ public enum CoreError implements ScalarDbError {
       "Mutations across multiple storages are not allowed. Mutations: %s",
       "",
       ""),
+  DROP_PRIMARY_KEY_COLUMN_NOT_SUPPORTED(
+      Category.USER_ERROR,
+      "0216",
+      "Primary key columns cannot be dropped. Table: %s; Column: %s",
+      "",
+      ""),
+  COSMOS_DROP_COLUMN_NOT_SUPPORTED(
+      Category.USER_ERROR,
+      "0217",
+      "Cosmos DB does not support the dropping column feature",
+      "",
+      ""),
+  DYNAMO_DROP_COLUMN_NOT_SUPPORTED(
+      Category.USER_ERROR, "0218", "DynamoDB does not support the dropping column feature", "", ""),
 
   //
   // Errors for the concurrency error category
@@ -960,6 +974,12 @@ public enum CoreError implements ScalarDbError {
       Category.INTERNAL_ERROR, "0057", "Recovering records failed. Details: %s", "", ""),
   CONSENSUS_COMMIT_COMMITTING_RECORDS_FAILED(
       Category.INTERNAL_ERROR, "0058", "Committing records failed. Details: %s", "", ""),
+  DROPPING_COLUMN_FROM_TABLE_FAILED(
+      Category.INTERNAL_ERROR,
+      "0059",
+      "Dropping a column from the table failed. Table: %s; Column: %s",
+      "",
+      ""),
 
   //
   // Errors for the unknown transaction status error category
diff --git a/core/src/main/java/com/scalar/db/common/DecoratedDistributedTransactionAdmin.java b/core/src/main/java/com/scalar/db/common/DecoratedDistributedTransactionAdmin.java
index 9724fc937d..9b3a8e1053 100644
--- a/core/src/main/java/com/scalar/db/common/DecoratedDistributedTransactionAdmin.java
+++ b/core/src/main/java/com/scalar/db/common/DecoratedDistributedTransactionAdmin.java
@@ -208,6 +208,19 @@ public void addNewColumnToTable(
         namespace, table, columnName, columnType, encrypted, ifNotExists);
   }
 
+  @Override
+  public void dropColumnFromTable(String namespace, String table, String columnName)
+      throws ExecutionException {
+    distributedTransactionAdmin.dropColumnFromTable(namespace, table, columnName);
+  }
+
+  @Override
+  public void dropColumnFromTable(
+      String namespace, String table, String columnName, boolean ifExists)
+      throws ExecutionException {
+    distributedTransactionAdmin.dropColumnFromTable(namespace, table, columnName, ifExists);
+  }
+
   @Override
   public void importTable(String namespace, String table, Map<String, String> options)
       throws ExecutionException {
diff --git a/core/src/main/java/com/scalar/db/service/AdminService.java b/core/src/main/java/com/scalar/db/service/AdminService.java
index 18806b1ac5..f9edfbfa94 100644
--- a/core/src/main/java/com/scalar/db/service/AdminService.java
+++ b/core/src/main/java/com/scalar/db/service/AdminService.java
@@ -94,6 +94,12 @@ public void addNewColumnToTable(
     admin.addNewColumnToTable(namespace, table, columnName, columnType);
   }
 
+  @Override
+  public void dropColumnFromTable(String namespace, String table, String columnName)
+      throws ExecutionException {
+    admin.dropColumnFromTable(namespace, table, columnName);
+  }
+
   @Override
   public TableMetadata getImportTableMetadata(
       String namespace, String table, Map<String, DataType> overrideColumnsType)
diff --git a/core/src/main/java/com/scalar/db/storage/cassandra/CassandraAdmin.java b/core/src/main/java/com/scalar/db/storage/cassandra/CassandraAdmin.java
index 1d85327eea..1e9d90e86d 100644
--- a/core/src/main/java/com/scalar/db/storage/cassandra/CassandraAdmin.java
+++ b/core/src/main/java/com/scalar/db/storage/cassandra/CassandraAdmin.java
@@ -415,6 +415,27 @@ columnName, getFullTableName(namespace, table)),
     }
   }
 
+  @Override
+  public void dropColumnFromTable(String namespace, String table, String columnName)
+      throws ExecutionException {
+    try {
+      String alterTableQuery =
+          SchemaBuilder.alterTable(quoteIfNecessary(namespace), quoteIfNecessary(table))
+              .dropColumn(quoteIfNecessary(columnName))
+              .getQueryString();
+
+      clusterManager.getSession().execute(alterTableQuery);
+    } catch (IllegalArgumentException e) {
+      throw e;
+    } catch (RuntimeException e) {
+      throw new ExecutionException(
+          String.format(
+              "Dropping the %s column from the %s table failed",
+              columnName, getFullTableName(namespace, table)),
+          e);
+    }
+  }
+
   @Override
   public Set<String> getNamespaceNames() throws ExecutionException {
     try {
diff --git a/core/src/main/java/com/scalar/db/storage/cosmos/CosmosAdmin.java b/core/src/main/java/com/scalar/db/storage/cosmos/CosmosAdmin.java
index ff1c2f3cff..c7be4cd074 100644
--- a/core/src/main/java/com/scalar/db/storage/cosmos/CosmosAdmin.java
+++ b/core/src/main/java/com/scalar/db/storage/cosmos/CosmosAdmin.java
@@ -644,6 +644,13 @@ columnName, getFullTableName(namespace, table)),
     }
   }
 
+  @Override
+  public void dropColumnFromTable(String namespace, String table, String columnName)
+      throws ExecutionException {
+    throw new UnsupportedOperationException(
+        CoreError.COSMOS_DROP_COLUMN_NOT_SUPPORTED.buildMessage());
+  }
+
   @Override
   public TableMetadata getImportTableMetadata(
       String namespace, String table, Map<String, DataType> overrideColumnsType) {
diff --git a/core/src/main/java/com/scalar/db/storage/dynamo/DynamoAdmin.java b/core/src/main/java/com/scalar/db/storage/dynamo/DynamoAdmin.java
index 5a0b621170..74ab81da02 100644
--- a/core/src/main/java/com/scalar/db/storage/dynamo/DynamoAdmin.java
+++ b/core/src/main/java/com/scalar/db/storage/dynamo/DynamoAdmin.java
@@ -9,6 +9,7 @@
 import com.scalar.db.api.Scan.Ordering.Order;
 import com.scalar.db.api.StorageInfo;
 import com.scalar.db.api.TableMetadata;
+import com.scalar.db.common.CoreError;
 import com.scalar.db.common.StorageInfoImpl;
 import com.scalar.db.config.DatabaseConfig;
 import com.scalar.db.exception.storage.ExecutionException;
@@ -1440,6 +1441,13 @@ columnName, getFullTableName(namespace, table)),
     }
   }
 
+  @Override
+  public void dropColumnFromTable(String nonPrefixedNamespace, String table, String columnName)
+      throws ExecutionException {
+    throw new UnsupportedOperationException(
+        CoreError.DYNAMO_DROP_COLUMN_NOT_SUPPORTED.buildMessage());
+  }
+
   @Override
   public TableMetadata getImportTableMetadata(
       String namespace, String table, Map<String, DataType> overrideColumnsType) {
diff --git a/core/src/main/java/com/scalar/db/storage/jdbc/JdbcAdmin.java b/core/src/main/java/com/scalar/db/storage/jdbc/JdbcAdmin.java
index f578b3b5c9..0af4e6cbbd 100644
--- a/core/src/main/java/com/scalar/db/storage/jdbc/JdbcAdmin.java
+++ b/core/src/main/java/com/scalar/db/storage/jdbc/JdbcAdmin.java
@@ -861,6 +861,29 @@ columnName, getFullTableName(namespace, table)),
     }
   }
 
+  @Override
+  public void dropColumnFromTable(String namespace, String table, String columnName)
+      throws ExecutionException {
+    try {
+      TableMetadata currentTableMetadata = getTableMetadata(namespace, table);
+      TableMetadata updatedTableMetadata =
+          TableMetadata.newBuilder(currentTableMetadata).removeColumn(columnName).build();
+      String[] dropColumnStatements = rdbEngine.dropColumnSql(namespace, table, columnName);
+      try (Connection connection = dataSource.getConnection()) {
+        for (String dropColumnStatement : dropColumnStatements) {
+          execute(connection, dropColumnStatement);
+        }
+        addTableMetadata(connection, namespace, table, updatedTableMetadata, false, true);
+      }
+    } catch (SQLException e) {
+      throw new ExecutionException(
+          String.format(
+              "Dropping the %s column from the %s table failed",
+              columnName, getFullTableName(namespace, table)),
+          e);
+    }
+  }
+
   @Override
   public void addRawColumnToTable(
       String namespace, String table, String columnName, DataType columnType)
diff --git a/core/src/main/java/com/scalar/db/storage/jdbc/RdbEngineDb2.java b/core/src/main/java/com/scalar/db/storage/jdbc/RdbEngineDb2.java
index cc52d349b2..b68a3fad61 100644
--- a/core/src/main/java/com/scalar/db/storage/jdbc/RdbEngineDb2.java
+++ b/core/src/main/java/com/scalar/db/storage/jdbc/RdbEngineDb2.java
@@ -246,6 +246,17 @@ public void dropNamespaceTranslateSQLException(SQLException e, String namespace)
     throw new ExecutionException("Dropping the schema failed: " + namespace, e);
   }
 
+  @Override
+  public String[] dropColumnSql(String namespace, String table, String columnName) {
+    return new String[] {
+      "ALTER TABLE "
+          + encloseFullTableName(namespace, table)
+          + " DROP COLUMN "
+          + enclose(columnName),
+      "CALL SYSPROC.ADMIN_CMD('REORG TABLE " + encloseFullTableName(namespace, table) + "')"
+    };
+  }
+
   @Override
   public String alterColumnTypeSql(
       String namespace, String table, String columnName, String columnType) {
diff --git a/core/src/main/java/com/scalar/db/storage/jdbc/RdbEngineStrategy.java b/core/src/main/java/com/scalar/db/storage/jdbc/RdbEngineStrategy.java
index 39edb7ec60..d5992cfea4 100644
--- a/core/src/main/java/com/scalar/db/storage/jdbc/RdbEngineStrategy.java
+++ b/core/src/main/java/com/scalar/db/storage/jdbc/RdbEngineStrategy.java
@@ -98,6 +98,15 @@ default String truncateTableSql(String namespace, String table) {
   void dropNamespaceTranslateSQLException(SQLException e, String namespace)
       throws ExecutionException;
 
+  default String[] dropColumnSql(String namespace, String table, String columnName) {
+    return new String[] {
+      "ALTER TABLE "
+          + encloseFullTableName(namespace, table)
+          + " DROP COLUMN "
+          + enclose(columnName)
+    };
+  }
+
   String alterColumnTypeSql(String namespace, String table, String columnName, String columnType);
 
   String tableExistsInternalTableCheckSql(String fullTableName);
diff --git a/core/src/main/java/com/scalar/db/storage/multistorage/MultiStorageAdmin.java b/core/src/main/java/com/scalar/db/storage/multistorage/MultiStorageAdmin.java
index 5c7c4f9a79..b4d4c5de65 100644
--- a/core/src/main/java/com/scalar/db/storage/multistorage/MultiStorageAdmin.java
+++ b/core/src/main/java/com/scalar/db/storage/multistorage/MultiStorageAdmin.java
@@ -205,6 +205,12 @@ public void addNewColumnToTable(
     getAdmin(namespace, table).addNewColumnToTable(namespace, table, columnName, columnType);
   }
 
+  @Override
+  public void dropColumnFromTable(String namespace, String table, String columnName)
+      throws ExecutionException {
+    getAdmin(namespace, table).dropColumnFromTable(namespace, table, columnName);
+  }
+
   @Override
   public TableMetadata getImportTableMetadata(
       String namespace, String table, Map<String, DataType> overrideColumnsType)
diff --git a/core/src/main/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitAdmin.java b/core/src/main/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitAdmin.java
index d9e5d6c313..da479d6ce9 100644
--- a/core/src/main/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitAdmin.java
+++ b/core/src/main/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitAdmin.java
@@ -238,6 +238,22 @@ public void addNewColumnToTable(
     admin.addNewColumnToTable(namespace, table, beforeColumnName, columnType);
   }
 
+  @Override
+  public void dropColumnFromTable(String namespace, String table, String columnName)
+      throws ExecutionException {
+    checkNamespace(namespace);
+
+    TableMetadata tableMetadata = getTableMetadata(namespace, table);
+    if (tableMetadata == null) {
+      throw new IllegalArgumentException(
+          CoreError.TABLE_NOT_FOUND.buildMessage(ScalarDbUtils.getFullTableName(namespace, table)));
+    }
+    String beforeColumnName = getBeforeImageColumnName(columnName, tableMetadata);
+
+    admin.dropColumnFromTable(namespace, table, columnName);
+    admin.dropColumnFromTable(namespace, table, beforeColumnName);
+  }
+
   @Override
   public Set<String> getNamespaceNames() throws ExecutionException {
     return admin.getNamespaceNames().stream()
diff --git a/core/src/main/java/com/scalar/db/transaction/jdbc/JdbcTransactionAdmin.java b/core/src/main/java/com/scalar/db/transaction/jdbc/JdbcTransactionAdmin.java
index 76f060903e..2b99bc1094 100644
--- a/core/src/main/java/com/scalar/db/transaction/jdbc/JdbcTransactionAdmin.java
+++ b/core/src/main/java/com/scalar/db/transaction/jdbc/JdbcTransactionAdmin.java
@@ -160,6 +160,12 @@ public void addNewColumnToTable(
     jdbcAdmin.addNewColumnToTable(namespace, table, columnName, columnType);
   }
 
+  @Override
+  public void dropColumnFromTable(String namespace, String table, String columnName)
+      throws ExecutionException {
+    jdbcAdmin.dropColumnFromTable(namespace, table, columnName);
+  }
+
   @Override
   public Set<String> getNamespaceNames() throws ExecutionException {
     return jdbcAdmin.getNamespaceNames();
diff --git a/core/src/main/java/com/scalar/db/transaction/singlecrudoperation/SingleCrudOperationTransactionAdmin.java b/core/src/main/java/com/scalar/db/transaction/singlecrudoperation/SingleCrudOperationTransactionAdmin.java
index a941639f1f..161870d154 100644
--- a/core/src/main/java/com/scalar/db/transaction/singlecrudoperation/SingleCrudOperationTransactionAdmin.java
+++ b/core/src/main/java/com/scalar/db/transaction/singlecrudoperation/SingleCrudOperationTransactionAdmin.java
@@ -109,6 +109,12 @@ public void addNewColumnToTable(
     distributedStorageAdmin.addNewColumnToTable(namespace, table, columnName, columnType);
   }
 
+  @Override
+  public void dropColumnFromTable(String namespace, String table, String columnName)
+      throws ExecutionException {
+    distributedStorageAdmin.dropColumnFromTable(namespace, table, columnName);
+  }
+
   @Override
   public Set<String> getNamespaceNames() throws ExecutionException {
     return distributedStorageAdmin.getNamespaceNames();
diff --git a/core/src/test/java/com/scalar/db/storage/cassandra/CassandraAdminTest.java b/core/src/test/java/com/scalar/db/storage/cassandra/CassandraAdminTest.java
index c8b7717adf..b7446f7976 100644
--- a/core/src/test/java/com/scalar/db/storage/cassandra/CassandraAdminTest.java
+++ b/core/src/test/java/com/scalar/db/storage/cassandra/CassandraAdminTest.java
@@ -865,6 +865,33 @@ public void addNewColumnToTable_ShouldWorkProperly() throws ExecutionException {
     verify(cassandraSession).execute(alterTableQuery);
   }
 
+  @Test
+  public void dropColumnFromTable_ShouldWorkProperly() throws ExecutionException {
+    // Arrange
+    String namespace = "sample_ns";
+    String table = "tbl";
+    String column = "c2";
+    com.datastax.driver.core.TableMetadata tableMetadata =
+        mock(com.datastax.driver.core.TableMetadata.class);
+    ColumnMetadata c1 = mock(ColumnMetadata.class);
+    when(c1.getName()).thenReturn("c1");
+    when(c1.getType()).thenReturn(com.datastax.driver.core.DataType.text());
+    when(tableMetadata.getPartitionKey()).thenReturn(Collections.singletonList(c1));
+    when(tableMetadata.getClusteringColumns()).thenReturn(Collections.emptyList());
+    when(tableMetadata.getIndexes()).thenReturn(Collections.emptyList());
+    when(tableMetadata.getColumns()).thenReturn(Collections.singletonList(c1));
+    when(clusterManager.getMetadata(any(), any())).thenReturn(tableMetadata);
+    when(clusterManager.getSession()).thenReturn(cassandraSession);
+
+    // Act
+    cassandraAdmin.dropColumnFromTable(namespace, table, column);
+
+    // Assert
+    String alterTableQuery =
+        SchemaBuilder.alterTable(namespace, table).dropColumn(column).getQueryString();
+    verify(cassandraSession).execute(alterTableQuery);
+  }
+
   @Test
   public void getNamespacesNames_WithNonExistingKeyspaces_ShouldReturnEmptySet()
       throws ExecutionException {
diff --git a/core/src/test/java/com/scalar/db/storage/cosmos/CosmosAdminTest.java b/core/src/test/java/com/scalar/db/storage/cosmos/CosmosAdminTest.java
index 7506e98abf..8e8e39b52a 100644
--- a/core/src/test/java/com/scalar/db/storage/cosmos/CosmosAdminTest.java
+++ b/core/src/test/java/com/scalar/db/storage/cosmos/CosmosAdminTest.java
@@ -1149,11 +1149,13 @@ public void unsupportedOperations_ShouldThrowUnsupportedException() {
             () ->
                 admin.importTable(
                     namespace, table, Collections.emptyMap(), Collections.emptyMap()));
+    Throwable thrown4 = catchThrowable(() -> admin.dropColumnFromTable(namespace, table, column));
 
     // Assert
     assertThat(thrown1).isInstanceOf(UnsupportedOperationException.class);
     assertThat(thrown2).isInstanceOf(UnsupportedOperationException.class);
     assertThat(thrown3).isInstanceOf(UnsupportedOperationException.class);
+    assertThat(thrown4).isInstanceOf(UnsupportedOperationException.class);
   }
 
   @Test
diff --git a/core/src/test/java/com/scalar/db/storage/dynamo/DynamoAdminTestBase.java b/core/src/test/java/com/scalar/db/storage/dynamo/DynamoAdminTestBase.java
index c208d85c43..1f971cd82c 100644
--- a/core/src/test/java/com/scalar/db/storage/dynamo/DynamoAdminTestBase.java
+++ b/core/src/test/java/com/scalar/db/storage/dynamo/DynamoAdminTestBase.java
@@ -1717,11 +1717,13 @@ public void unsupportedOperations_ShouldThrowUnsupportedException() {
             () ->
                 admin.importTable(
                     NAMESPACE, TABLE, Collections.emptyMap(), Collections.emptyMap()));
+    Throwable thrown4 = catchThrowable(() -> admin.dropColumnFromTable(NAMESPACE, TABLE, "c1"));
 
     // Assert
     assertThat(thrown1).isInstanceOf(UnsupportedOperationException.class);
     assertThat(thrown2).isInstanceOf(UnsupportedOperationException.class);
     assertThat(thrown3).isInstanceOf(UnsupportedOperationException.class);
+    assertThat(thrown4).isInstanceOf(UnsupportedOperationException.class);
   }
 
   @Test
diff --git a/core/src/test/java/com/scalar/db/storage/jdbc/JdbcAdminTest.java b/core/src/test/java/com/scalar/db/storage/jdbc/JdbcAdminTest.java
index ee7480a99a..7cd83297fc 100644
--- a/core/src/test/java/com/scalar/db/storage/jdbc/JdbcAdminTest.java
+++ b/core/src/test/java/com/scalar/db/storage/jdbc/JdbcAdminTest.java
@@ -2939,6 +2939,146 @@ private void addNewColumnToTable_ForX_ShouldWorkProperly(
     }
   }
 
+  @Test
+  public void dropColumnFromTable_ForMysql_ShouldWorkProperly()
+      throws SQLException, ExecutionException {
+    dropColumnFromTable_ForX_ShouldWorkProperly(
+        RdbEngine.MYSQL,
+        "SELECT `column_name`,`data_type`,`key_type`,`clustering_order`,`indexed` FROM `"
+            + METADATA_SCHEMA
+            + "`.`metadata` WHERE `full_table_name`=? ORDER BY `ordinal_position` ASC",
+        "ALTER TABLE `ns`.`table` DROP COLUMN `c2`",
+        "DELETE FROM `" + METADATA_SCHEMA + "`.`metadata` WHERE `full_table_name` = 'ns.table'",
+        "INSERT INTO `"
+            + METADATA_SCHEMA
+            + "`.`metadata` VALUES ('ns.table','c1','TEXT','PARTITION',NULL,false,1)");
+  }
+
+  @Test
+  public void dropColumnFromTable_ForOracle_ShouldWorkProperly()
+      throws SQLException, ExecutionException {
+    dropColumnFromTable_ForX_ShouldWorkProperly(
+        RdbEngine.ORACLE,
+        "SELECT \"column_name\",\"data_type\",\"key_type\",\"clustering_order\",\"indexed\" FROM \""
+            + METADATA_SCHEMA
+            + "\".\"metadata\" WHERE \"full_table_name\"=? ORDER BY \"ordinal_position\" ASC",
+        "ALTER TABLE \"ns\".\"table\" DROP COLUMN \"c2\"",
+        "DELETE FROM \""
+            + METADATA_SCHEMA
+            + "\".\"metadata\" WHERE \"full_table_name\" = 'ns.table'",
+        "INSERT INTO \""
+            + METADATA_SCHEMA
+            + "\".\"metadata\" VALUES ('ns.table','c1','TEXT','PARTITION',NULL,0,1)");
+  }
+
+  @Test
+  public void dropColumnFromTable_ForPostgresql_ShouldWorkProperly()
+      throws SQLException, ExecutionException {
+    dropColumnFromTable_ForX_ShouldWorkProperly(
+        RdbEngine.POSTGRESQL,
+        "SELECT \"column_name\",\"data_type\",\"key_type\",\"clustering_order\",\"indexed\" FROM \""
+            + METADATA_SCHEMA
+            + "\".\"metadata\" WHERE \"full_table_name\"=? ORDER BY \"ordinal_position\" ASC",
+        "ALTER TABLE \"ns\".\"table\" DROP COLUMN \"c2\"",
+        "DELETE FROM \""
+            + METADATA_SCHEMA
+            + "\".\"metadata\" WHERE \"full_table_name\" = 'ns.table'",
+        "INSERT INTO \""
+            + METADATA_SCHEMA
+            + "\".\"metadata\" VALUES ('ns.table','c1','TEXT','PARTITION',NULL,false,1)");
+  }
+
+  @Test
+  public void dropColumnFromTable_ForSqlServer_ShouldWorkProperly()
+      throws SQLException, ExecutionException {
+    dropColumnFromTable_ForX_ShouldWorkProperly(
+        RdbEngine.SQL_SERVER,
+        "SELECT [column_name],[data_type],[key_type],[clustering_order],[indexed] FROM ["
+            + METADATA_SCHEMA
+            + "].[metadata] WHERE [full_table_name]=? ORDER BY [ordinal_position] ASC",
+        "ALTER TABLE [ns].[table] DROP COLUMN [c2]",
+        "DELETE FROM [" + METADATA_SCHEMA + "].[metadata] WHERE [full_table_name] = 'ns.table'",
+        "INSERT INTO ["
+            + METADATA_SCHEMA
+            + "].[metadata] VALUES ('ns.table','c1','TEXT','PARTITION',NULL,0,1)");
+  }
+
+  @Test
+  public void dropColumnFromTable_ForSqlite_ShouldWorkProperly()
+      throws SQLException, ExecutionException {
+    dropColumnFromTable_ForX_ShouldWorkProperly(
+        RdbEngine.SQLITE,
+        "SELECT \"column_name\",\"data_type\",\"key_type\",\"clustering_order\",\"indexed\" FROM \""
+            + METADATA_SCHEMA
+            + "$metadata\" WHERE \"full_table_name\"=? ORDER BY \"ordinal_position\" ASC",
+        "ALTER TABLE \"ns$table\" DROP COLUMN \"c2\"",
+        "DELETE FROM \"" + METADATA_SCHEMA + "$metadata\" WHERE \"full_table_name\" = 'ns.table'",
+        "INSERT INTO \""
+            + METADATA_SCHEMA
+            + "$metadata\" VALUES ('ns.table','c1','TEXT','PARTITION',NULL,FALSE,1)");
+  }
+
+  @Test
+  public void dropColumnFromTable_ForDb2_ShouldWorkProperly()
+      throws SQLException, ExecutionException {
+    dropColumnFromTable_ForX_ShouldWorkProperly(
+        RdbEngine.DB2,
+        "SELECT \"column_name\",\"data_type\",\"key_type\",\"clustering_order\",\"indexed\" FROM \""
+            + METADATA_SCHEMA
+            + "\".\"metadata\" WHERE \"full_table_name\"=? ORDER BY \"ordinal_position\" ASC",
+        "ALTER TABLE \"ns\".\"table\" DROP COLUMN \"c2\"",
+        "CALL SYSPROC.ADMIN_CMD('REORG TABLE \"ns\".\"table\"')",
+        "DELETE FROM \""
+            + METADATA_SCHEMA
+            + "\".\"metadata\" WHERE \"full_table_name\" = 'ns.table'",
+        "INSERT INTO \""
+            + METADATA_SCHEMA
+            + "\".\"metadata\" VALUES ('ns.table','c1','TEXT','PARTITION',NULL,false,1)");
+  }
+
+  private void dropColumnFromTable_ForX_ShouldWorkProperly(
+      RdbEngine rdbEngine, String expectedGetMetadataStatement, String... expectedSqlStatements)
+      throws SQLException, ExecutionException {
+    // Arrange
+    String namespace = "ns";
+    String table = "table";
+    String column1 = "c1";
+    String column2 = "c2";
+
+    PreparedStatement selectStatement = mock(PreparedStatement.class);
+    ResultSet resultSet =
+        mockResultSet(
+            new SelectAllFromMetadataTableResultSetMocker.Row(
+                column1, DataType.TEXT.toString(), "PARTITION", null, false),
+            new SelectAllFromMetadataTableResultSetMocker.Row(
+                column2, DataType.INT.toString(), null, null, false));
+    when(selectStatement.executeQuery()).thenReturn(resultSet);
+
+    when(connection.prepareStatement(any())).thenReturn(selectStatement);
+    List<Statement> expectedStatements = new ArrayList<>();
+    for (int i = 0; i < expectedSqlStatements.length; i++) {
+      Statement expectedStatement = mock(Statement.class);
+      expectedStatements.add(expectedStatement);
+    }
+    when(connection.createStatement())
+        .thenReturn(
+            expectedStatements.get(0),
+            expectedStatements.subList(1, expectedStatements.size()).toArray(new Statement[0]));
+
+    when(dataSource.getConnection()).thenReturn(connection);
+    JdbcAdmin admin = createJdbcAdminFor(rdbEngine);
+
+    // Act
+    admin.dropColumnFromTable(namespace, table, column2);
+
+    // Assert
+    verify(selectStatement).setString(1, getFullTableName(namespace, table));
+    verify(connection).prepareStatement(expectedGetMetadataStatement);
+    for (int i = 0; i < expectedSqlStatements.length; i++) {
+      verify(expectedStatements.get(i)).execute(expectedSqlStatements[i]);
+    }
+  }
+
   @Test
   public void getNamespaceNames_forMysql_ShouldReturnNamespaceNames() throws Exception {
     getNamespaceNames_forX_ShouldReturnNamespaceNames(
diff --git a/core/src/test/java/com/scalar/db/storage/multistorage/MultiStorageAdminTest.java b/core/src/test/java/com/scalar/db/storage/multistorage/MultiStorageAdminTest.java
index 2a74bf0a87..3e0cae38a5 100644
--- a/core/src/test/java/com/scalar/db/storage/multistorage/MultiStorageAdminTest.java
+++ b/core/src/test/java/com/scalar/db/storage/multistorage/MultiStorageAdminTest.java
@@ -499,6 +499,68 @@ public void addNewColumnToTable_ForTable1InNamespace2_ShouldCallAddNewColumnOfAd
     verify(admin2).addNewColumnToTable(namespace, table, column, dataType);
   }
 
+  @Test
+  public void dropColumnFromTable_ForTable1InNamespace1_ShouldCallDropColumnFromTableOfAdmin1()
+      throws ExecutionException {
+    // Arrange
+    String namespace = NAMESPACE1;
+    String table = TABLE1;
+    String column = "c1";
+
+    // Act
+    multiStorageAdmin.dropColumnFromTable(namespace, table, column);
+
+    // Assert
+    verify(admin1).dropColumnFromTable(namespace, table, column);
+  }
+
+  @Test
+  public void
+      dropColumnFromTable_ForTable2InNamespace1_ShouldShouldCallDropColumnFromTableOfAdmin2()
+          throws ExecutionException {
+    // Arrange
+    String namespace = NAMESPACE1;
+    String table = TABLE2;
+    String column = "c1";
+
+    // Act
+    multiStorageAdmin.dropColumnFromTable(namespace, table, column);
+
+    // Assert
+    verify(admin2).dropColumnFromTable(namespace, table, column);
+  }
+
+  @Test
+  public void
+      dropColumnFromTable_ForTable3InNamespace1_ShouldCallDropColumnFromTableOfDefaultAdmin()
+          throws ExecutionException {
+    // Arrange
+    String namespace = NAMESPACE1;
+    String table = TABLE3;
+    String column = "c1";
+
+    // Act
+    multiStorageAdmin.dropColumnFromTable(namespace, table, column);
+
+    // Assert
+    verify(admin3).dropColumnFromTable(namespace, table, column);
+  }
+
+  @Test
+  public void dropColumnFromTable_ForTable1InNamespace2_ShouldCallDropColumnFromTableOfAdmin2()
+      throws ExecutionException {
+    // Arrange
+    String namespace = NAMESPACE2;
+    String table = TABLE1;
+    String column = "c1";
+
+    // Act
+    multiStorageAdmin.dropColumnFromTable(namespace, table, column);
+
+    // Assert
+    verify(admin2).dropColumnFromTable(namespace, table, column);
+  }
+
   @Test
   public void
       getNamespaceNames_WithExistingNamespacesNotInMapping_ShouldReturnExistingNamespacesInMappingAndFromDefaultAdmin()
diff --git a/core/src/test/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitAdminTestBase.java b/core/src/test/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitAdminTestBase.java
index 696ea3c5d8..789124919f 100644
--- a/core/src/test/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitAdminTestBase.java
+++ b/core/src/test/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitAdminTestBase.java
@@ -602,6 +602,29 @@ public void addNewColumnToTable_WithEncrypted_ShouldThrowUnsupportedOperationExc
         .isInstanceOf(UnsupportedOperationException.class);
   }
 
+  @Test
+  public void dropColumnFromTable_ShouldCallJdbcAdminProperly() throws ExecutionException {
+    // Arrange
+    String targetColumn = "col2";
+    TableMetadata tableMetadata =
+        TableMetadata.newBuilder()
+            .addColumn("col1", DataType.INT)
+            .addColumn(targetColumn, DataType.INT)
+            .addPartitionKey("col1")
+            .build();
+    when(distributedStorageAdmin.getTableMetadata(any(), any()))
+        .thenReturn(ConsensusCommitUtils.buildTransactionTableMetadata(tableMetadata));
+
+    // Act
+    admin.dropColumnFromTable(NAMESPACE, TABLE, targetColumn);
+
+    // Assert
+    verify(distributedStorageAdmin).getTableMetadata(NAMESPACE, TABLE);
+    verify(distributedStorageAdmin).dropColumnFromTable(NAMESPACE, TABLE, targetColumn);
+    verify(distributedStorageAdmin)
+        .dropColumnFromTable(NAMESPACE, TABLE, Attribute.BEFORE_PREFIX + targetColumn);
+  }
+
   @Test
   public void importTable_ShouldCallStorageAdminProperly() throws ExecutionException {
     // Arrange
diff --git a/core/src/test/java/com/scalar/db/transaction/jdbc/JdbcTransactionAdminTest.java b/core/src/test/java/com/scalar/db/transaction/jdbc/JdbcTransactionAdminTest.java
index 8e0f9f077c..b88fdebe67 100644
--- a/core/src/test/java/com/scalar/db/transaction/jdbc/JdbcTransactionAdminTest.java
+++ b/core/src/test/java/com/scalar/db/transaction/jdbc/JdbcTransactionAdminTest.java
@@ -228,6 +228,20 @@ public void addNewColumnToTable_ShouldCallJdbcAdminProperly() throws ExecutionEx
     verify(jdbcAdmin).addNewColumnToTable(namespace, table, column, dataType);
   }
 
+  @Test
+  public void dropColumnFromTable_ShouldCallJdbcAdminProperly() throws ExecutionException {
+    // Arrange
+    String namespace = "ns";
+    String table = "tbl";
+    String column = "c1";
+
+    // Act
+    admin.dropColumnFromTable(namespace, table, column);
+
+    // Assert
+    verify(jdbcAdmin).dropColumnFromTable(namespace, table, column);
+  }
+
   @Test
   public void importTable_ShouldCallJdbcAdminProperly() throws ExecutionException {
     // Arrange
diff --git a/core/src/test/java/com/scalar/db/transaction/singlecrudoperation/SingleCrudOperationTransactionAdminTest.java b/core/src/test/java/com/scalar/db/transaction/singlecrudoperation/SingleCrudOperationTransactionAdminTest.java
index b108f5d801..f463ad7063 100644
--- a/core/src/test/java/com/scalar/db/transaction/singlecrudoperation/SingleCrudOperationTransactionAdminTest.java
+++ b/core/src/test/java/com/scalar/db/transaction/singlecrudoperation/SingleCrudOperationTransactionAdminTest.java
@@ -233,6 +233,21 @@ public void addNewColumnToTable_ShouldCallDistributedStorageAdminProperly()
     verify(distributedStorageAdmin).addNewColumnToTable(namespace, table, column, dataType);
   }
 
+  @Test
+  public void dropColumnFromTable_ShouldCallDistributedStorageAdminProperly()
+      throws ExecutionException {
+    // Arrange
+    String namespace = "ns";
+    String table = "tbl";
+    String column = "c1";
+
+    // Act
+    admin.dropColumnFromTable(namespace, table, column);
+
+    // Assert
+    verify(distributedStorageAdmin).dropColumnFromTable(namespace, table, column);
+  }
+
   @Test
   public void importTable_ShouldCallDistributedStorageAdminProperly() throws ExecutionException {
     // Arrange
diff --git a/integration-test/src/main/java/com/scalar/db/api/DistributedStorageAdminIntegrationTestBase.java b/integration-test/src/main/java/com/scalar/db/api/DistributedStorageAdminIntegrationTestBase.java
index 761555cc4b..952e70b9b3 100644
--- a/integration-test/src/main/java/com/scalar/db/api/DistributedStorageAdminIntegrationTestBase.java
+++ b/integration-test/src/main/java/com/scalar/db/api/DistributedStorageAdminIntegrationTestBase.java
@@ -977,6 +977,71 @@ public void addNewColumnToTable_AddColumnForEachExistingDataType_ShouldAddNewCol
     }
   }
 
+  @Test
+  public void dropColumnFromTable_DropColumnForEachExistingDataType_ShouldDropColumnsCorrectly()
+      throws ExecutionException {
+    try {
+      // Arrange
+      Map<String, String> options = getCreationOptions();
+      TableMetadata.Builder currentTableMetadataBuilder =
+          TableMetadata.newBuilder()
+              .addColumn(getColumnName1(), DataType.INT)
+              .addColumn(getColumnName2(), DataType.INT)
+              .addColumn(getColumnName3(), DataType.INT)
+              .addColumn(getColumnName4(), DataType.BIGINT)
+              .addColumn(getColumnName5(), DataType.FLOAT)
+              .addColumn(getColumnName6(), DataType.DOUBLE)
+              .addColumn(getColumnName7(), DataType.TEXT)
+              .addColumn(getColumnName8(), DataType.BLOB)
+              .addColumn(getColumnName9(), DataType.DATE)
+              .addColumn(getColumnName10(), DataType.TIME)
+              .addPartitionKey(getColumnName1())
+              .addClusteringKey(getColumnName2(), Scan.Ordering.Order.ASC);
+      if (isTimestampTypeSupported()) {
+        currentTableMetadataBuilder
+            .addColumn(getColumnName11(), DataType.TIMESTAMP)
+            .addColumn(getColumnName12(), DataType.TIMESTAMPTZ);
+      }
+      TableMetadata currentTableMetadata = currentTableMetadataBuilder.build();
+      admin.createTable(namespace1, getTable4(), currentTableMetadata, options);
+
+      // Act
+      admin.dropColumnFromTable(namespace1, getTable4(), getColumnName3());
+      admin.dropColumnFromTable(namespace1, getTable4(), getColumnName4());
+      admin.dropColumnFromTable(namespace1, getTable4(), getColumnName5());
+      admin.dropColumnFromTable(namespace1, getTable4(), getColumnName6());
+      admin.dropColumnFromTable(namespace1, getTable4(), getColumnName7());
+      admin.dropColumnFromTable(namespace1, getTable4(), getColumnName8());
+      admin.dropColumnFromTable(namespace1, getTable4(), getColumnName9());
+      admin.dropColumnFromTable(namespace1, getTable4(), getColumnName10());
+      if (isTimestampTypeSupported()) {
+        admin.dropColumnFromTable(namespace1, getTable4(), getColumnName11());
+        admin.dropColumnFromTable(namespace1, getTable4(), getColumnName12());
+      }
+
+      // Assert
+      TableMetadata.Builder expectedTableMetadataBuilder =
+          TableMetadata.newBuilder(currentTableMetadata)
+              .removeColumn(getColumnName3())
+              .removeColumn(getColumnName4())
+              .removeColumn(getColumnName5())
+              .removeColumn(getColumnName6())
+              .removeColumn(getColumnName7())
+              .removeColumn(getColumnName8())
+              .removeColumn(getColumnName9())
+              .removeColumn(getColumnName10());
+      if (isTimestampTypeSupported()) {
+        expectedTableMetadataBuilder
+            .removeColumn(getColumnName11())
+            .removeColumn(getColumnName12());
+      }
+      TableMetadata expectedTableMetadata = expectedTableMetadataBuilder.build();
+      assertThat(admin.getTableMetadata(namespace1, getTable4())).isEqualTo(expectedTableMetadata);
+    } finally {
+      admin.dropTable(namespace1, getTable4(), true);
+    }
+  }
+
   @Test
   public void getNamespaceNames_ShouldReturnCreatedNamespaces() throws ExecutionException {
     // Arrange
@@ -1010,6 +1075,79 @@ public void addNewColumnToTable_ForAlreadyExistingColumn_ShouldThrowIllegalArgum
         .isInstanceOf(IllegalArgumentException.class);
   }
 
+  @Test
+  public void dropColumnFromTable_ForNonExistingTable_ShouldThrowIllegalArgumentException() {
+    // Arrange
+
+    // Act Assert
+    assertThatThrownBy(() -> admin.dropColumnFromTable(namespace1, getTable4(), getColumnName2()))
+        .isInstanceOf(IllegalArgumentException.class);
+  }
+
+  @Test
+  public void dropColumnFromTable_ForNonExistingColumn_ShouldThrowIllegalArgumentException() {
+    // Arrange
+
+    // Act Assert
+    assertThatThrownBy(
+            () -> admin.dropColumnFromTable(namespace1, getTable1(), "nonExistingColumn"))
+        .isInstanceOf(IllegalArgumentException.class);
+  }
+
+  @Test
+  public void dropColumnFromTable_ForPrimaryKeyColumn_ShouldThrowIllegalArgumentException() {
+    // Arrange
+
+    // Act Assert
+    assertThatThrownBy(() -> admin.dropColumnFromTable(namespace1, getTable1(), getColumnName1()))
+        .isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> admin.dropColumnFromTable(namespace1, getTable1(), getColumnName3()))
+        .isInstanceOf(IllegalArgumentException.class);
+  }
+
+  @Test
+  public void dropColumnFromTable_ForIndexedColumn_ShouldDropColumnAndIndexCorrectly()
+      throws ExecutionException {
+    try {
+      // Arrange
+      Map<String, String> options = getCreationOptions();
+      TableMetadata currentTableMetadata =
+          TableMetadata.newBuilder()
+              .addColumn(getColumnName1(), DataType.INT)
+              .addColumn(getColumnName2(), DataType.INT)
+              .addColumn(getColumnName3(), DataType.TEXT)
+              .addPartitionKey(getColumnName1())
+              .addSecondaryIndex(getColumnName2())
+              .build();
+      admin.createTable(namespace1, getTable4(), currentTableMetadata, options);
+
+      // Act
+      admin.dropColumnFromTable(namespace1, getTable4(), getColumnName2());
+
+      // Assert
+      TableMetadata expectedTableMetadata =
+          TableMetadata.newBuilder()
+              .addColumn(getColumnName1(), DataType.INT)
+              .addColumn(getColumnName3(), DataType.TEXT)
+              .addPartitionKey(getColumnName1())
+              .build();
+      assertThat(admin.getTableMetadata(namespace1, getTable4())).isEqualTo(expectedTableMetadata);
+      assertThat(admin.indexExists(namespace1, getTable4(), getColumnName2())).isFalse();
+    } finally {
+      admin.dropTable(namespace1, getTable4(), true);
+    }
+  }
+
+  @Test
+  public void dropColumnFromTable_IfExists_ForNonExistingColumn_ShouldNotThrowAnyException() {
+    // Arrange
+
+    // Act Assert
+    assertThatCode(
+            () -> admin.dropColumnFromTable(namespace1, getTable1(), "nonExistingColumn", true))
+        .doesNotThrowAnyException();
+  }
+
   @Test
   public void
       addNewColumnToTable_IfNotExists_ForAlreadyExistingColumn_ShouldNotThrowAnyException() {
diff --git a/integration-test/src/main/java/com/scalar/db/api/DistributedStorageAdminPermissionIntegrationTestBase.java b/integration-test/src/main/java/com/scalar/db/api/DistributedStorageAdminPermissionIntegrationTestBase.java
index c6f3448557..7d4b2b8ad4 100644
--- a/integration-test/src/main/java/com/scalar/db/api/DistributedStorageAdminPermissionIntegrationTestBase.java
+++ b/integration-test/src/main/java/com/scalar/db/api/DistributedStorageAdminPermissionIntegrationTestBase.java
@@ -325,6 +325,18 @@ public void addNewColumnToTable_WithSufficientPermission_ShouldSucceed()
         .doesNotThrowAnyException();
   }
 
+  @Test
+  public void dropColumnFromTable_WithSufficientPermission_ShouldSucceed()
+      throws ExecutionException {
+    // Arrange
+    createNamespaceByRoot();
+    createTableByRoot();
+
+    // Act Assert
+    assertThatCode(() -> adminForNormalUser.dropColumnFromTable(NAMESPACE, TABLE, COL_NAME3))
+        .doesNotThrowAnyException();
+  }
+
   @Test
   public void importTable_WithSufficientPermission_ShouldSucceed() throws Exception {
     // Arrange
diff --git a/integration-test/src/main/java/com/scalar/db/api/DistributedTransactionAdminIntegrationTestBase.java b/integration-test/src/main/java/com/scalar/db/api/DistributedTransactionAdminIntegrationTestBase.java
index 8fb7037a66..3c39d3dfaf 100644
--- a/integration-test/src/main/java/com/scalar/db/api/DistributedTransactionAdminIntegrationTestBase.java
+++ b/integration-test/src/main/java/com/scalar/db/api/DistributedTransactionAdminIntegrationTestBase.java
@@ -899,6 +899,139 @@ public void addNewColumnToTable_ForAlreadyExistingColumn_ShouldThrowIllegalArgum
         .doesNotThrowAnyException();
   }
 
+  @Test
+  public void dropColumnFromTable_DropColumnForEachExistingDataType_ShouldDropColumnsCorrectly()
+      throws ExecutionException {
+    try {
+      // Arrange
+      Map<String, String> options = getCreationOptions();
+      TableMetadata.Builder currentTableMetadataBuilder =
+          TableMetadata.newBuilder()
+              .addColumn("c1", DataType.INT)
+              .addColumn("c2", DataType.INT)
+              .addColumn("c3", DataType.INT)
+              .addColumn("c4", DataType.BIGINT)
+              .addColumn("c5", DataType.FLOAT)
+              .addColumn("c6", DataType.DOUBLE)
+              .addColumn("c7", DataType.TEXT)
+              .addColumn("c8", DataType.BLOB)
+              .addColumn("c9", DataType.DATE)
+              .addColumn("c10", DataType.TIME)
+              .addPartitionKey("c1")
+              .addClusteringKey("c2", Scan.Ordering.Order.ASC);
+      if (isTimestampTypeSupported()) {
+        currentTableMetadataBuilder
+            .addColumn("c11", DataType.TIMESTAMP)
+            .addColumn("c12", DataType.TIMESTAMPTZ);
+      }
+      TableMetadata currentTableMetadata = currentTableMetadataBuilder.build();
+      admin.createTable(namespace1, TABLE4, currentTableMetadata, options);
+
+      // Act
+      admin.dropColumnFromTable(namespace1, TABLE4, "c3");
+      admin.dropColumnFromTable(namespace1, TABLE4, "c4");
+      admin.dropColumnFromTable(namespace1, TABLE4, "c5");
+      admin.dropColumnFromTable(namespace1, TABLE4, "c6");
+      admin.dropColumnFromTable(namespace1, TABLE4, "c7");
+      admin.dropColumnFromTable(namespace1, TABLE4, "c8");
+      admin.dropColumnFromTable(namespace1, TABLE4, "c9");
+      admin.dropColumnFromTable(namespace1, TABLE4, "c10");
+      if (isTimestampTypeSupported()) {
+        admin.dropColumnFromTable(namespace1, TABLE4, "c11");
+        admin.dropColumnFromTable(namespace1, TABLE4, "c12");
+      }
+
+      // Assert
+      TableMetadata.Builder expectedTableMetadataBuilder =
+          TableMetadata.newBuilder(currentTableMetadata)
+              .removeColumn("c3")
+              .removeColumn("c4")
+              .removeColumn("c5")
+              .removeColumn("c6")
+              .removeColumn("c7")
+              .removeColumn("c8")
+              .removeColumn("c9")
+              .removeColumn("c10")
+              .removeColumn("c11")
+              .removeColumn("c12");
+      TableMetadata expectedTableMetadata = expectedTableMetadataBuilder.build();
+      assertThat(admin.getTableMetadata(namespace1, TABLE4)).isEqualTo(expectedTableMetadata);
+    } finally {
+      admin.dropTable(namespace1, TABLE4, true);
+    }
+  }
+
+  @Test
+  public void dropColumnFromTable_ForNonExistingTable_ShouldThrowIllegalArgumentException() {
+    // Arrange
+
+    // Act Assert
+    assertThatThrownBy(() -> admin.dropColumnFromTable(namespace1, TABLE4, COL_NAME2))
+        .isInstanceOf(IllegalArgumentException.class);
+  }
+
+  @Test
+  public void dropColumnFromTable_ForNonExistingColumn_ShouldThrowIllegalArgumentException() {
+    // Arrange
+
+    // Act Assert
+    assertThatThrownBy(() -> admin.dropColumnFromTable(namespace1, TABLE1, "nonExistingColumn"))
+        .isInstanceOf(IllegalArgumentException.class);
+  }
+
+  @Test
+  public void dropColumnFromTable_ForPrimaryKeyColumn_ShouldThrowIllegalArgumentException() {
+    // Arrange
+
+    // Act Assert
+    assertThatThrownBy(() -> admin.dropColumnFromTable(namespace1, TABLE1, "c1"))
+        .isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> admin.dropColumnFromTable(namespace1, TABLE1, "c3"))
+        .isInstanceOf(IllegalArgumentException.class);
+  }
+
+  @Test
+  public void dropColumnFromTable_ForIndexedColumn_ShouldDropColumnAndIndexCorrectly()
+      throws ExecutionException {
+    try {
+      // Arrange
+      Map<String, String> options = getCreationOptions();
+      TableMetadata currentTableMetadata =
+          TableMetadata.newBuilder()
+              .addColumn("c1", DataType.INT)
+              .addColumn("c2", DataType.INT)
+              .addColumn("c3", DataType.TEXT)
+              .addPartitionKey("c1")
+              .addSecondaryIndex("c2")
+              .build();
+      admin.createTable(namespace1, TABLE4, currentTableMetadata, options);
+
+      // Act
+      admin.dropColumnFromTable(namespace1, TABLE4, "c2");
+
+      // Assert
+      TableMetadata expectedTableMetadata =
+          TableMetadata.newBuilder()
+              .addColumn("c1", DataType.INT)
+              .addColumn("c3", DataType.TEXT)
+              .addPartitionKey("c1")
+              .build();
+      assertThat(admin.getTableMetadata(namespace1, TABLE4)).isEqualTo(expectedTableMetadata);
+      assertThat(admin.indexExists(namespace1, TABLE4, "c2")).isFalse();
+    } finally {
+      admin.dropTable(namespace1, TABLE4, true);
+    }
+  }
+
+  @Test
+  public void dropColumnFromTable_IfNotExists_ForNonExistingColumn_ShouldNotThrowAnyException() {
+    // Arrange
+
+    // Act Assert
+    assertThatCode(() -> admin.dropColumnFromTable(namespace1, TABLE1, "nonExistingColumn", true))
+        .doesNotThrowAnyException();
+  }
+
   @Test
   public void createCoordinatorTables_ShouldCreateCoordinatorTablesCorrectly()
       throws ExecutionException {