diff --git a/core/src/main/java/com/scalar/db/api/ReplicationAdmin.java b/core/src/main/java/com/scalar/db/api/ReplicationAdmin.java
index 63916fb898..ce595d8d64 100644
--- a/core/src/main/java/com/scalar/db/api/ReplicationAdmin.java
+++ b/core/src/main/java/com/scalar/db/api/ReplicationAdmin.java
@@ -2,7 +2,6 @@
 
 import com.scalar.db.common.error.CoreError;
 import com.scalar.db.exception.storage.ExecutionException;
-import java.util.Collections;
 import java.util.Map;
 
 /** An administrative interface for the replication feature. */
@@ -31,10 +30,7 @@ default void createReplicationTables(Map<String, String> options) throws Executi
    */
   default void createReplicationTables(boolean ifNotExist, Map<String, String> options)
       throws ExecutionException {
-    if (ifNotExist && replicationTablesExist()) {
-      return;
-    }
-    createReplicationTables(options);
+    throw new UnsupportedOperationException(CoreError.REPLICATION_NOT_ENABLED.buildMessage());
   }
 
   /**
@@ -48,10 +44,7 @@ default void createReplicationTables(boolean ifNotExist, Map<String, String> opt
    * @throws ExecutionException if the operation fails
    */
   default void createReplicationTables(boolean ifNotExist) throws ExecutionException {
-    if (ifNotExist && replicationTablesExist()) {
-      return;
-    }
-    createReplicationTables(Collections.emptyMap());
+    throw new UnsupportedOperationException(CoreError.REPLICATION_NOT_ENABLED.buildMessage());
   }
 
   /**
@@ -60,7 +53,7 @@ default void createReplicationTables(boolean ifNotExist) throws ExecutionExcepti
    * @throws ExecutionException if the operation fails
    */
   default void createReplicationTables() throws ExecutionException {
-    createReplicationTables(Collections.emptyMap());
+    throw new UnsupportedOperationException(CoreError.REPLICATION_NOT_ENABLED.buildMessage());
   }
 
   /**
@@ -83,10 +76,7 @@ default void dropReplicationTables() throws ExecutionException {
    * @throws ExecutionException if the operation fails
    */
   default void dropReplicationTables(boolean ifExist) throws ExecutionException {
-    if (ifExist && !replicationTablesExist()) {
-      return;
-    }
-    dropReplicationTables();
+    throw new UnsupportedOperationException(CoreError.REPLICATION_NOT_ENABLED.buildMessage());
   }
 
   /**
diff --git a/core/src/main/java/com/scalar/db/common/error/CoreError.java b/core/src/main/java/com/scalar/db/common/error/CoreError.java
index f6c02711b5..d58ff52f64 100644
--- a/core/src/main/java/com/scalar/db/common/error/CoreError.java
+++ b/core/src/main/java/com/scalar/db/common/error/CoreError.java
@@ -861,8 +861,7 @@ public enum CoreError implements ScalarDbError {
   REPLICATION_NOT_ENABLED(
       Category.USER_ERROR,
       "0188",
-      // TODO: Update the message once the licence type is determined.
-      "The replication feature is not enabled. To use this feature, you must enable it",
+      "The replication feature is not enabled. To use this feature, you must enable it. Note that this feature is supported only in the ScalarDB Enterprise edition",
       "",
       ""),
   DATA_LOADER_IMPORT_TARGET_MISSING(
diff --git a/schema-loader/build.gradle b/schema-loader/build.gradle
index 97523de479..26a2a358b4 100644
--- a/schema-loader/build.gradle
+++ b/schema-loader/build.gradle
@@ -19,6 +19,7 @@ dependencies {
     implementation "com.google.code.gson:gson:${gsonVersion}"
     implementation "info.picocli:picocli:${picocliVersion}"
     testImplementation "org.junit.jupiter:junit-jupiter-api:${junitVersion}"
+    testImplementation "org.junit.jupiter:junit-jupiter-params:${junitVersion}"
     testRuntimeOnly "org.junit.jupiter:junit-jupiter-engine:${junitVersion}"
     testImplementation "org.assertj:assertj-core:${assertjVersion}"
     testImplementation "org.mockito:mockito-core:${mockitoVersion}"
diff --git a/schema-loader/src/main/java/com/scalar/db/schemaloader/SchemaLoader.java b/schema-loader/src/main/java/com/scalar/db/schemaloader/SchemaLoader.java
index efeb9bd472..32d179ec1b 100644
--- a/schema-loader/src/main/java/com/scalar/db/schemaloader/SchemaLoader.java
+++ b/schema-loader/src/main/java/com/scalar/db/schemaloader/SchemaLoader.java
@@ -97,17 +97,19 @@ private static Object getCommand(String arg) {
    * @param schemaFilePath path to schema file.
    * @param options specific options for creating tables.
    * @param createCoordinatorTables create coordinator tables or not.
+   * @param createReplicationTables create replication tables or not.
    * @throws SchemaLoaderException thrown when creating tables fails.
    */
   public static void load(
       Properties configProperties,
       @Nullable Path schemaFilePath,
       Map<String, String> options,
-      boolean createCoordinatorTables)
+      boolean createCoordinatorTables,
+      boolean createReplicationTables)
       throws SchemaLoaderException {
     Either<Path, Properties> config = new Right<>(configProperties);
     Either<Path, String> schema = new Left<>(schemaFilePath);
-    load(config, schema, options, createCoordinatorTables);
+    load(config, schema, options, createCoordinatorTables, createReplicationTables);
   }
 
   /**
@@ -117,17 +119,19 @@ public static void load(
    * @param schemaFilePath path to schema file.
    * @param options specific options for creating tables.
    * @param createCoordinatorTables create coordinator tables or not.
+   * @param createReplicationTables create replication tables or not.
    * @throws SchemaLoaderException thrown when creating tables fails.
    */
   public static void load(
       Path configFilePath,
       @Nullable Path schemaFilePath,
       Map<String, String> options,
-      boolean createCoordinatorTables)
+      boolean createCoordinatorTables,
+      boolean createReplicationTables)
       throws SchemaLoaderException {
     Either<Path, Properties> config = new Left<>(configFilePath);
     Either<Path, String> schema = new Left<>(schemaFilePath);
-    load(config, schema, options, createCoordinatorTables);
+    load(config, schema, options, createCoordinatorTables, createReplicationTables);
   }
 
   /**
@@ -137,17 +141,19 @@ public static void load(
    * @param serializedSchemaJson serialized json string schema.
    * @param options specific options for creating tables.
    * @param createCoordinatorTables create coordinator tables or not.
+   * @param createReplicationTables create replication tables or not.
    * @throws SchemaLoaderException thrown when creating tables fails.
    */
   public static void load(
       Properties configProperties,
       @Nullable String serializedSchemaJson,
       Map<String, String> options,
-      boolean createCoordinatorTables)
+      boolean createCoordinatorTables,
+      boolean createReplicationTables)
       throws SchemaLoaderException {
     Either<Path, Properties> config = new Right<>(configProperties);
     Either<Path, String> schema = new Right<>(serializedSchemaJson);
-    load(config, schema, options, createCoordinatorTables);
+    load(config, schema, options, createCoordinatorTables, createReplicationTables);
   }
 
   /**
@@ -157,17 +163,19 @@ public static void load(
    * @param serializedSchemaJson serialized json string schema.
    * @param options specific options for creating tables.
    * @param createCoordinatorTables create coordinator tables or not.
+   * @param createReplicationTables create replication tables or not.
    * @throws SchemaLoaderException thrown when creating tables fails.
    */
   public static void load(
       Path configFilePath,
       @Nullable String serializedSchemaJson,
       Map<String, String> options,
-      boolean createCoordinatorTables)
+      boolean createCoordinatorTables,
+      boolean createReplicationTables)
       throws SchemaLoaderException {
     Either<Path, Properties> config = new Left<>(configFilePath);
     Either<Path, String> schema = new Right<>(serializedSchemaJson);
-    load(config, schema, options, createCoordinatorTables);
+    load(config, schema, options, createCoordinatorTables, createReplicationTables);
   }
 
   /**
@@ -177,13 +185,15 @@ public static void load(
    * @param schema schema definition.
    * @param options specific options for creating tables.
    * @param createCoordinatorTables create coordinator tables or not.
+   * @param createReplicationTables create replication tables or not.
    * @throws SchemaLoaderException thrown when creating tables fails.
    */
   private static void load(
       Either<Path, Properties> config,
       Either<Path, String> schema,
       Map<String, String> options,
-      boolean createCoordinatorTables)
+      boolean createCoordinatorTables,
+      boolean createReplicationTables)
       throws SchemaLoaderException {
     // Parse the schema
     List<TableSchema> tableSchemaList = getTableSchemaList(schema, options);
@@ -194,6 +204,9 @@ private static void load(
       if (createCoordinatorTables) {
         operator.createCoordinatorTables(options);
       }
+      if (createReplicationTables) {
+        operator.createReplicationTables(options);
+      }
     }
   }
 
@@ -203,14 +216,18 @@ private static void load(
    * @param configProperties ScalarDB config properties.
    * @param schemaFilePath path to schema file.
    * @param deleteCoordinatorTables delete coordinator tables or not.
+   * @param deleteReplicationTables delete replication tables or not.
    * @throws SchemaLoaderException thrown when deleting tables fails.
    */
   public static void unload(
-      Properties configProperties, @Nullable Path schemaFilePath, boolean deleteCoordinatorTables)
+      Properties configProperties,
+      @Nullable Path schemaFilePath,
+      boolean deleteCoordinatorTables,
+      boolean deleteReplicationTables)
       throws SchemaLoaderException {
     Either<Path, Properties> config = new Right<>(configProperties);
     Either<Path, String> schema = new Left<>(schemaFilePath);
-    unload(config, schema, deleteCoordinatorTables);
+    unload(config, schema, deleteCoordinatorTables, deleteReplicationTables);
   }
 
   /**
@@ -219,14 +236,18 @@ public static void unload(
    * @param configFilePath path to ScalarDB config file.
    * @param schemaFilePath path to schema file.
    * @param deleteCoordinatorTables delete coordinator tables or not.
+   * @param deleteReplicationTables delete replication tables or not.
    * @throws SchemaLoaderException thrown when deleting tables fails.
    */
   public static void unload(
-      Path configFilePath, @Nullable Path schemaFilePath, boolean deleteCoordinatorTables)
+      Path configFilePath,
+      @Nullable Path schemaFilePath,
+      boolean deleteCoordinatorTables,
+      boolean deleteReplicationTables)
       throws SchemaLoaderException {
     Either<Path, Properties> config = new Left<>(configFilePath);
     Either<Path, String> schema = new Left<>(schemaFilePath);
-    unload(config, schema, deleteCoordinatorTables);
+    unload(config, schema, deleteCoordinatorTables, deleteReplicationTables);
   }
 
   /**
@@ -235,16 +256,18 @@ public static void unload(
    * @param configProperties ScalarDB config properties.
    * @param serializedSchemaJson serialized json string schema.
    * @param deleteCoordinatorTables delete coordinator tables or not.
+   * @param deleteReplicationTables delete replication tables or not.
    * @throws SchemaLoaderException thrown when deleting tables fails.
    */
   public static void unload(
       Properties configProperties,
       @Nullable String serializedSchemaJson,
-      boolean deleteCoordinatorTables)
+      boolean deleteCoordinatorTables,
+      boolean deleteReplicationTables)
       throws SchemaLoaderException {
     Either<Path, Properties> config = new Right<>(configProperties);
     Either<Path, String> schema = new Right<>(serializedSchemaJson);
-    unload(config, schema, deleteCoordinatorTables);
+    unload(config, schema, deleteCoordinatorTables, deleteReplicationTables);
   }
 
   /**
@@ -253,14 +276,18 @@ public static void unload(
    * @param configFilePath path to ScalarDB config file.
    * @param serializedSchemaJson serialized json string schema.
    * @param deleteCoordinatorTables delete coordinator tables or not.
+   * @param deleteReplicationTables delete replication tables or not.
    * @throws SchemaLoaderException thrown when deleting tables fails.
    */
   public static void unload(
-      Path configFilePath, @Nullable String serializedSchemaJson, boolean deleteCoordinatorTables)
+      Path configFilePath,
+      @Nullable String serializedSchemaJson,
+      boolean deleteCoordinatorTables,
+      boolean deleteReplicationTables)
       throws SchemaLoaderException {
     Either<Path, Properties> config = new Left<>(configFilePath);
     Either<Path, String> schema = new Right<>(serializedSchemaJson);
-    unload(config, schema, deleteCoordinatorTables);
+    unload(config, schema, deleteCoordinatorTables, deleteReplicationTables);
   }
 
   /**
@@ -269,10 +296,14 @@ public static void unload(
    * @param config ScalarDB config.
    * @param schema schema definition.
    * @param deleteCoordinatorTables delete coordinator tables or not.
+   * @param deleteReplicationTables delete replication tables or not.
    * @throws SchemaLoaderException thrown when deleting tables fails.
    */
   private static void unload(
-      Either<Path, Properties> config, Either<Path, String> schema, boolean deleteCoordinatorTables)
+      Either<Path, Properties> config,
+      Either<Path, String> schema,
+      boolean deleteCoordinatorTables,
+      boolean deleteReplicationTables)
       throws SchemaLoaderException {
     // Parse the schema
     List<TableSchema> tableSchemaList = getTableSchemaList(schema, Collections.emptyMap());
@@ -283,6 +314,9 @@ private static void unload(
       if (deleteCoordinatorTables) {
         operator.dropCoordinatorTables();
       }
+      if (deleteReplicationTables) {
+        operator.dropReplicationTables();
+      }
     }
   }
 
@@ -296,17 +330,19 @@ private static void unload(
    * @param serializedSchemaJson serialized json string schema.
    * @param options specific options for repairing.
    * @param repairCoordinatorTable repair coordinator tables or not.
+   * @param repairReplicationTable repair replication tables or not.
    * @throws SchemaLoaderException thrown when repairing fails.
    */
   public static void repairAll(
       Properties configProperties,
       String serializedSchemaJson,
       Map<String, String> options,
-      boolean repairCoordinatorTable)
+      boolean repairCoordinatorTable,
+      boolean repairReplicationTable)
       throws SchemaLoaderException {
     Either<Path, Properties> config = new Right<>(configProperties);
     Either<Path, String> schema = new Right<>(serializedSchemaJson);
-    repairAll(config, schema, options, repairCoordinatorTable);
+    repairAll(config, schema, options, repairCoordinatorTable, repairReplicationTable);
   }
 
   /**
@@ -319,17 +355,19 @@ public static void repairAll(
    * @param schemaPath path to the schema file.
    * @param options specific options for repairing.
    * @param repairCoordinatorTable repair coordinator tables or not.
+   * @param repairReplicationTable repair replication tables or not.
    * @throws SchemaLoaderException thrown when repairing fails.
    */
   public static void repairAll(
       Properties configProperties,
       Path schemaPath,
       Map<String, String> options,
-      boolean repairCoordinatorTable)
+      boolean repairCoordinatorTable,
+      boolean repairReplicationTable)
       throws SchemaLoaderException {
     Either<Path, Properties> config = new Right<>(configProperties);
     Either<Path, String> schema = new Left<>(schemaPath);
-    repairAll(config, schema, options, repairCoordinatorTable);
+    repairAll(config, schema, options, repairCoordinatorTable, repairReplicationTable);
   }
 
   /**
@@ -342,17 +380,19 @@ public static void repairAll(
    * @param serializedSchemaJson serialized json string schema.
    * @param options specific options for repairing.
    * @param repairCoordinatorTable repair coordinator tables or not.
+   * @param repairReplicationTable repair replication tables or not.
    * @throws SchemaLoaderException thrown when repairing fails.
    */
   public static void repairAll(
       Path configPath,
       String serializedSchemaJson,
       Map<String, String> options,
-      boolean repairCoordinatorTable)
+      boolean repairCoordinatorTable,
+      boolean repairReplicationTable)
       throws SchemaLoaderException {
     Either<Path, Properties> config = new Left<>(configPath);
     Either<Path, String> schema = new Right<>(serializedSchemaJson);
-    repairAll(config, schema, options, repairCoordinatorTable);
+    repairAll(config, schema, options, repairCoordinatorTable, repairReplicationTable);
   }
 
   /**
@@ -365,14 +405,19 @@ public static void repairAll(
    * @param schemaPath path to the schema file.
    * @param options specific options for repairing.
    * @param repairCoordinatorTable repair coordinator tables or not.
+   * @param repairReplicationTable repair replication tables or not.
    * @throws SchemaLoaderException thrown when repairing fails.
    */
   public static void repairAll(
-      Path configPath, Path schemaPath, Map<String, String> options, boolean repairCoordinatorTable)
+      Path configPath,
+      Path schemaPath,
+      Map<String, String> options,
+      boolean repairCoordinatorTable,
+      boolean repairReplicationTable)
       throws SchemaLoaderException {
     Either<Path, Properties> config = new Left<>(configPath);
     Either<Path, String> schema = new Left<>(schemaPath);
-    repairAll(config, schema, options, repairCoordinatorTable);
+    repairAll(config, schema, options, repairCoordinatorTable, repairReplicationTable);
   }
 
   /**
@@ -382,13 +427,15 @@ public static void repairAll(
    * @param schema schema.
    * @param options specific options for repairing.
    * @param repairCoordinatorTable repair coordinator tables or not.
+   * @param repairReplicationTable repair replication tables or not.
    * @throws SchemaLoaderException thrown when repairing fails.
    */
   private static void repairAll(
       Either<Path, Properties> config,
       Either<Path, String> schema,
       Map<String, String> options,
-      boolean repairCoordinatorTable)
+      boolean repairCoordinatorTable,
+      boolean repairReplicationTable)
       throws SchemaLoaderException {
     // Parse the schema
     List<TableSchema> tableSchemaList = getTableSchemaList(schema, options);
@@ -400,6 +447,9 @@ private static void repairAll(
       if (repairCoordinatorTable) {
         operator.repairCoordinatorTables(options);
       }
+      if (repairReplicationTable) {
+        operator.repairReplicationTables(options);
+      }
     }
   }
 
diff --git a/schema-loader/src/main/java/com/scalar/db/schemaloader/SchemaOperator.java b/schema-loader/src/main/java/com/scalar/db/schemaloader/SchemaOperator.java
index efe7c01ed0..b586b2b6fa 100644
--- a/schema-loader/src/main/java/com/scalar/db/schemaloader/SchemaOperator.java
+++ b/schema-loader/src/main/java/com/scalar/db/schemaloader/SchemaOperator.java
@@ -286,6 +286,49 @@ public void repairCoordinatorTables(Map<String, String> options) throws SchemaLo
     }
   }
 
+  public void createReplicationTables(Map<String, String> options) throws SchemaLoaderException {
+    if (replicationTablesExist()) {
+      logger.warn("The replication tables already exist");
+      return;
+    }
+    try {
+      transactionAdmin.get().createReplicationTables(options);
+      logger.info("Creating the replication tables succeeded");
+    } catch (ExecutionException e) {
+      throw new SchemaLoaderException(e.getMessage(), e);
+    }
+  }
+
+  public void dropReplicationTables() throws SchemaLoaderException {
+    if (!replicationTablesExist()) {
+      logger.warn("The replication tables don't exist");
+      return;
+    }
+    try {
+      transactionAdmin.get().dropReplicationTables();
+      logger.info("Deleting the replication tables succeeded");
+    } catch (ExecutionException e) {
+      throw new SchemaLoaderException(e.getMessage(), e);
+    }
+  }
+
+  private boolean replicationTablesExist() throws SchemaLoaderException {
+    try {
+      return transactionAdmin.get().replicationTablesExist();
+    } catch (ExecutionException e) {
+      throw new SchemaLoaderException(e.getMessage(), e);
+    }
+  }
+
+  public void repairReplicationTables(Map<String, String> options) throws SchemaLoaderException {
+    try {
+      transactionAdmin.get().repairReplicationTables(options);
+      logger.info("Repairing the replication tables succeeded");
+    } catch (ExecutionException e) {
+      throw new SchemaLoaderException(e.getMessage(), e);
+    }
+  }
+
   public void alterTables(List<TableSchema> tableSchemaList, Map<String, String> options)
       throws SchemaLoaderException {
     for (TableSchema tableSchema : tableSchemaList) {
diff --git a/schema-loader/src/main/java/com/scalar/db/schemaloader/command/SchemaLoaderCommand.java b/schema-loader/src/main/java/com/scalar/db/schemaloader/command/SchemaLoaderCommand.java
index a7da2921d1..e8293fb35d 100644
--- a/schema-loader/src/main/java/com/scalar/db/schemaloader/command/SchemaLoaderCommand.java
+++ b/schema-loader/src/main/java/com/scalar/db/schemaloader/command/SchemaLoaderCommand.java
@@ -62,6 +62,12 @@ public class SchemaLoaderCommand implements Callable<Integer> {
       defaultValue = "false")
   private boolean coordinator;
 
+  @Option(
+      names = "--replication-tables",
+      description = "Create/delete/repair replication tables",
+      defaultValue = "false")
+  private boolean replicationTables;
+
   @Option(
       names = {"-f", "--schema-file"},
       description = "Path to the schema json file")
@@ -115,7 +121,7 @@ public Integer call() throws Exception {
     if (mode == null) {
       createTables();
     } else if (mode.deleteTables) {
-      SchemaLoader.unload(configPath, schemaFile, coordinator);
+      SchemaLoader.unload(configPath, schemaFile, coordinator, replicationTables);
     } else if (mode.repairAll) {
       repairAll();
     } else if (mode.alterTables) {
@@ -130,7 +136,7 @@ public Integer call() throws Exception {
 
   private void createTables() throws SchemaLoaderException {
     Map<String, String> options = prepareAllOptions();
-    SchemaLoader.load(configPath, schemaFile, options, coordinator);
+    SchemaLoader.load(configPath, schemaFile, options, coordinator, replicationTables);
   }
 
   private void repairAll() throws SchemaLoaderException {
@@ -140,7 +146,7 @@ private void repairAll() throws SchemaLoaderException {
               .buildMessage());
     }
     Map<String, String> options = prepareAllOptions();
-    SchemaLoader.repairAll(configPath, schemaFile, options, coordinator);
+    SchemaLoader.repairAll(configPath, schemaFile, options, coordinator, replicationTables);
   }
 
   private void alterTables() throws SchemaLoaderException {
diff --git a/schema-loader/src/test/java/com/scalar/db/schemaloader/SchemaLoaderTest.java b/schema-loader/src/test/java/com/scalar/db/schemaloader/SchemaLoaderTest.java
index 666c54955e..4a320ba0c1 100644
--- a/schema-loader/src/test/java/com/scalar/db/schemaloader/SchemaLoaderTest.java
+++ b/schema-loader/src/test/java/com/scalar/db/schemaloader/SchemaLoaderTest.java
@@ -17,6 +17,8 @@
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 import org.mockito.Mock;
 import org.mockito.MockedStatic;
 import org.mockito.MockitoAnnotations;
@@ -59,612 +61,859 @@ public void tearDown() throws Exception {
     closeable.close();
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      load_WithConfigFileAndSchemaFileWithCreateCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      load_WithConfigFileAndSchemaFileWithCreateCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.load(configFilePath, schemaFilePath, options, true);
+    SchemaLoader.load(configFilePath, schemaFilePath, options, true, withReplicationTables);
 
     // Assert
     verify(parser).parse();
     verify(operator).createTables(anyList());
     verify(operator).createCoordinatorTables(options);
+    if (withReplicationTables) {
+      verify(operator).createReplicationTables(options);
+    } else {
+      verify(operator, never()).createReplicationTables(options);
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      load_WithConfigFileAndSchemaFileWithoutCreateCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      load_WithConfigFileAndSchemaFileWithoutCreateCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.load(configFilePath, schemaFilePath, options, false);
+    SchemaLoader.load(configFilePath, schemaFilePath, options, false, withReplicationTables);
 
     // Assert
     verify(parser).parse();
     verify(operator).createTables(anyList());
     verify(operator, never()).createCoordinatorTables(options);
+    if (withReplicationTables) {
+      verify(operator).createReplicationTables(options);
+    } else {
+      verify(operator, never()).createReplicationTables(options);
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      load_WithConfigFileAndNullSchemaFileWithCreateCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      load_WithConfigFileAndNullSchemaFileWithCreateCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.load(configFilePath, (Path) null, options, true);
+    SchemaLoader.load(configFilePath, (Path) null, options, true, withReplicationTables);
 
     // Assert
     verify(parser, never()).parse();
     verify(operator).createTables(anyList());
     verify(operator).createCoordinatorTables(options);
+    if (withReplicationTables) {
+      verify(operator).createReplicationTables(options);
+    } else {
+      verify(operator, never()).createReplicationTables(options);
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      load_WithConfigFileAndNullSchemaFileWithoutCreateCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      load_WithConfigFileAndNullSchemaFileWithoutCreateCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.load(configFilePath, (Path) null, options, false);
+    SchemaLoader.load(configFilePath, (Path) null, options, false, withReplicationTables);
 
     // Assert
     verify(parser, never()).parse();
     verify(operator).createTables(anyList());
     verify(operator, never()).createCoordinatorTables(options);
+    if (withReplicationTables) {
+      verify(operator).createReplicationTables(options);
+    } else {
+      verify(operator, never()).createReplicationTables(options);
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      load_WithConfigFileAndSerializedSchemaJsonWithCreateCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      load_WithConfigFileAndSerializedSchemaJsonWithCreateCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.load(configFilePath, SERIALIZED_SCHEMA_JSON, options, true);
+    SchemaLoader.load(configFilePath, SERIALIZED_SCHEMA_JSON, options, true, withReplicationTables);
 
     // Assert
     verify(parser).parse();
     verify(operator).createTables(anyList());
     verify(operator).createCoordinatorTables(options);
+    if (withReplicationTables) {
+      verify(operator).createReplicationTables(options);
+    } else {
+      verify(operator, never()).createReplicationTables(options);
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      load_WithConfigFileAndSerializedSchemaJsonWithoutCreateCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      load_WithConfigFileAndSerializedSchemaJsonWithoutCreateCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.load(configFilePath, SERIALIZED_SCHEMA_JSON, options, false);
+    SchemaLoader.load(
+        configFilePath, SERIALIZED_SCHEMA_JSON, options, false, withReplicationTables);
 
     // Assert
     verify(parser).parse();
     verify(operator).createTables(anyList());
     verify(operator, never()).createCoordinatorTables(options);
+    if (withReplicationTables) {
+      verify(operator).createReplicationTables(options);
+    } else {
+      verify(operator, never()).createReplicationTables(options);
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      load_WithConfigFileAndNullSerializedSchemaJsonWithCreateCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      load_WithConfigFileAndNullSerializedSchemaJsonWithCreateCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.load(configFilePath, (String) null, options, true);
+    SchemaLoader.load(configFilePath, (String) null, options, true, withReplicationTables);
 
     // Assert
     verify(parser, never()).parse();
     verify(operator).createTables(anyList());
     verify(operator).createCoordinatorTables(options);
+    if (withReplicationTables) {
+      verify(operator).createReplicationTables(options);
+    } else {
+      verify(operator, never()).createReplicationTables(options);
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      load_WithConfigFileAndNullSerializedSchemaJsonWithoutCreateCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      load_WithConfigFileAndNullSerializedSchemaJsonWithoutCreateCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.load(configFilePath, (String) null, options, false);
+    SchemaLoader.load(configFilePath, (String) null, options, false, withReplicationTables);
 
     // Assert
     verify(parser, never()).parse();
     verify(operator).createTables(anyList());
     verify(operator, never()).createCoordinatorTables(options);
+    if (withReplicationTables) {
+      verify(operator).createReplicationTables(options);
+    } else {
+      verify(operator, never()).createReplicationTables(options);
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      load_WithConfigPropertiesAndSchemaFileWithCreateCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      load_WithConfigPropertiesAndSchemaFileWithCreateCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.load(configProperties, schemaFilePath, options, true);
+    SchemaLoader.load(configProperties, schemaFilePath, options, true, withReplicationTables);
 
     // Assert
     verify(parser).parse();
     verify(operator).createTables(anyList());
     verify(operator).createCoordinatorTables(options);
+    if (withReplicationTables) {
+      verify(operator).createReplicationTables(options);
+    } else {
+      verify(operator, never()).createReplicationTables(options);
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      load_WithConfigPropertiesAndSchemaFileWithoutCreateCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      load_WithConfigPropertiesAndSchemaFileWithoutCreateCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.load(configProperties, schemaFilePath, options, false);
+    SchemaLoader.load(configProperties, schemaFilePath, options, false, withReplicationTables);
 
     // Assert
     verify(parser).parse();
     verify(operator).createTables(anyList());
     verify(operator, never()).createCoordinatorTables(options);
+    if (withReplicationTables) {
+      verify(operator).createReplicationTables(options);
+    } else {
+      verify(operator, never()).createReplicationTables(options);
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      load_WithConfigPropertiesAndNullSchemaFileWithCreateCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      load_WithConfigPropertiesAndNullSchemaFileWithCreateCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.load(configProperties, (Path) null, options, true);
+    SchemaLoader.load(configProperties, (Path) null, options, true, withReplicationTables);
 
     // Assert
     verify(parser, never()).parse();
     verify(operator).createTables(anyList());
     verify(operator).createCoordinatorTables(options);
+    if (withReplicationTables) {
+      verify(operator).createReplicationTables(options);
+    } else {
+      verify(operator, never()).createReplicationTables(options);
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      load_WithConfigPropertiesAndNullSchemaFileWithoutCreateCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      load_WithConfigPropertiesAndNullSchemaFileWithoutCreateCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.load(configProperties, (Path) null, options, false);
+    SchemaLoader.load(configProperties, (Path) null, options, false, withReplicationTables);
 
     // Assert
     verify(parser, never()).parse();
     verify(operator).createTables(anyList());
     verify(operator, never()).createCoordinatorTables(options);
+    if (withReplicationTables) {
+      verify(operator).createReplicationTables(options);
+    } else {
+      verify(operator, never()).createReplicationTables(options);
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      load_WithConfigPropertiesAndSerializedSchemaJsonWithCreateCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      load_WithConfigPropertiesAndSerializedSchemaJsonWithCreateCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.load(configProperties, SERIALIZED_SCHEMA_JSON, options, true);
+    SchemaLoader.load(
+        configProperties, SERIALIZED_SCHEMA_JSON, options, true, withReplicationTables);
 
     // Assert
     verify(parser).parse();
     verify(operator).createTables(anyList());
     verify(operator).createCoordinatorTables(options);
+    if (withReplicationTables) {
+      verify(operator).createReplicationTables(options);
+    } else {
+      verify(operator, never()).createReplicationTables(options);
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      load_WithConfigPropertiesAndSerializedSchemaJsonWithoutCreateCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      load_WithConfigPropertiesAndSerializedSchemaJsonWithoutCreateCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.load(configProperties, SERIALIZED_SCHEMA_JSON, options, false);
+    SchemaLoader.load(
+        configProperties, SERIALIZED_SCHEMA_JSON, options, false, withReplicationTables);
 
     // Assert
     verify(parser).parse();
     verify(operator).createTables(anyList());
     verify(operator, never()).createCoordinatorTables(options);
+    if (withReplicationTables) {
+      verify(operator).createReplicationTables(options);
+    } else {
+      verify(operator, never()).createReplicationTables(options);
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      load_WithConfigPropertiesAndNullSerializedSchemaJsonWithCreateCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      load_WithConfigPropertiesAndNullSerializedSchemaJsonWithCreateCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.load(configProperties, (String) null, options, true);
+    SchemaLoader.load(configProperties, (String) null, options, true, withReplicationTables);
 
     // Assert
     verify(parser, never()).parse();
     verify(operator).createTables(anyList());
     verify(operator).createCoordinatorTables(options);
+    if (withReplicationTables) {
+      verify(operator).createReplicationTables(options);
+    } else {
+      verify(operator, never()).createReplicationTables(options);
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      load_WithConfigPropertiesAndNullSerializedSchemaJsonWithoutCreateCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      load_WithConfigPropertiesAndNullSerializedSchemaJsonWithoutCreateCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.load(configProperties, (String) null, options, false);
+    SchemaLoader.load(configProperties, (String) null, options, false, withReplicationTables);
 
     // Assert
     verify(parser, never()).parse();
     verify(operator).createTables(anyList());
     verify(operator, never()).createCoordinatorTables(options);
+    if (withReplicationTables) {
+      verify(operator).createReplicationTables(options);
+    } else {
+      verify(operator, never()).createReplicationTables(options);
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      unload_WithConfigFileAndSchemaFileWithDeleteCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      unload_WithConfigFileAndSchemaFileWithDeleteCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.unload(configFilePath, schemaFilePath, true);
+    SchemaLoader.unload(configFilePath, schemaFilePath, true, withReplicationTables);
 
     // Assert
     verify(parser).parse();
     verify(operator).deleteTables(anyList());
     verify(operator).dropCoordinatorTables();
+    if (withReplicationTables) {
+      verify(operator).dropReplicationTables();
+    } else {
+      verify(operator, never()).dropReplicationTables();
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      unload_WithConfigFileAndSchemaFileWithoutDeleteCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      unload_WithConfigFileAndSchemaFileWithoutDeleteCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.unload(configFilePath, schemaFilePath, false);
+    SchemaLoader.unload(configFilePath, schemaFilePath, false, withReplicationTables);
 
     // Assert
     verify(parser).parse();
     verify(operator).deleteTables(anyList());
     verify(operator, never()).dropCoordinatorTables();
+    if (withReplicationTables) {
+      verify(operator).dropReplicationTables();
+    } else {
+      verify(operator, never()).dropReplicationTables();
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      unload_WithConfigFileAndNullSchemaFileWithDeleteCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      unload_WithConfigFileAndNullSchemaFileWithDeleteCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.unload(configFilePath, (Path) null, true);
+    SchemaLoader.unload(configFilePath, (Path) null, true, withReplicationTables);
 
     // Assert
     verify(parser, never()).parse();
     verify(operator).deleteTables(anyList());
     verify(operator).dropCoordinatorTables();
+    if (withReplicationTables) {
+      verify(operator).dropReplicationTables();
+    } else {
+      verify(operator, never()).dropReplicationTables();
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      unload_WithConfigFileAndNullSchemaFileWithoutDeleteCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      unload_WithConfigFileAndNullSchemaFileWithoutDeleteCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.unload(configFilePath, (Path) null, false);
+    SchemaLoader.unload(configFilePath, (Path) null, false, withReplicationTables);
 
     // Assert
     verify(parser, never()).parse();
     verify(operator).deleteTables(anyList());
     verify(operator, never()).dropCoordinatorTables();
+    if (withReplicationTables) {
+      verify(operator).dropReplicationTables();
+    } else {
+      verify(operator, never()).dropReplicationTables();
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      unload_WithConfigFileAndSerializedSchemaJsonWithDeleteCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      unload_WithConfigFileAndSerializedSchemaJsonWithDeleteCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.unload(configFilePath, SERIALIZED_SCHEMA_JSON, true);
+    SchemaLoader.unload(configFilePath, SERIALIZED_SCHEMA_JSON, true, withReplicationTables);
 
     // Assert
     verify(parser).parse();
     verify(operator).deleteTables(anyList());
     verify(operator).dropCoordinatorTables();
+    if (withReplicationTables) {
+      verify(operator).dropReplicationTables();
+    } else {
+      verify(operator, never()).dropReplicationTables();
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      unload_WithConfigFileAndSerializedSchemaJsonWithoutDeleteCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      unload_WithConfigFileAndSerializedSchemaJsonWithoutDeleteCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.unload(configFilePath, SERIALIZED_SCHEMA_JSON, false);
+    SchemaLoader.unload(configFilePath, SERIALIZED_SCHEMA_JSON, false, withReplicationTables);
 
     // Assert
     verify(parser).parse();
     verify(operator).deleteTables(anyList());
     verify(operator, never()).dropCoordinatorTables();
+    if (withReplicationTables) {
+      verify(operator).dropReplicationTables();
+    } else {
+      verify(operator, never()).dropReplicationTables();
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      unload_WithConfigFileAndNullSerializedSchemaJsonWithDeleteCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      unload_WithConfigFileAndNullSerializedSchemaJsonWithDeleteCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.unload(configFilePath, (String) null, true);
+    SchemaLoader.unload(configFilePath, (String) null, true, withReplicationTables);
 
     // Assert
     verify(parser, never()).parse();
     verify(operator).deleteTables(anyList());
     verify(operator).dropCoordinatorTables();
+    if (withReplicationTables) {
+      verify(operator).dropReplicationTables();
+    } else {
+      verify(operator, never()).dropReplicationTables();
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      unload_WithConfigFileAndNullSerializedSchemaJsonWithoutDeleteCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      unload_WithConfigFileAndNullSerializedSchemaJsonWithoutDeleteCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.unload(configFilePath, (String) null, false);
+    SchemaLoader.unload(configFilePath, (String) null, false, withReplicationTables);
 
     // Assert
     verify(parser, never()).parse();
     verify(operator).deleteTables(anyList());
     verify(operator, never()).dropCoordinatorTables();
+    if (withReplicationTables) {
+      verify(operator).dropReplicationTables();
+    } else {
+      verify(operator, never()).dropReplicationTables();
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      unload_WithConfigPropertiesAndSchemaFileWithDeleteCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      unload_WithConfigPropertiesAndSchemaFileWithDeleteCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.unload(configProperties, schemaFilePath, true);
+    SchemaLoader.unload(configProperties, schemaFilePath, true, withReplicationTables);
 
     // Assert
     verify(parser).parse();
     verify(operator).deleteTables(anyList());
     verify(operator).dropCoordinatorTables();
+    if (withReplicationTables) {
+      verify(operator).dropReplicationTables();
+    } else {
+      verify(operator, never()).dropReplicationTables();
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      unload_WithConfigPropertiesAndSchemaFileWithoutDeleteCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      unload_WithConfigPropertiesAndSchemaFileWithoutDeleteCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.unload(configProperties, schemaFilePath, false);
+    SchemaLoader.unload(configProperties, schemaFilePath, false, withReplicationTables);
 
     // Assert
     verify(parser).parse();
     verify(operator).deleteTables(anyList());
     verify(operator, never()).dropCoordinatorTables();
+    if (withReplicationTables) {
+      verify(operator).dropReplicationTables();
+    } else {
+      verify(operator, never()).dropReplicationTables();
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      unload_WithConfigPropertiesAndNullSchemaFileWithDeleteCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      unload_WithConfigPropertiesAndNullSchemaFileWithDeleteCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.unload(configProperties, (Path) null, true);
+    SchemaLoader.unload(configProperties, (Path) null, true, withReplicationTables);
 
     // Assert
     verify(parser, never()).parse();
     verify(operator).deleteTables(anyList());
     verify(operator).dropCoordinatorTables();
+    if (withReplicationTables) {
+      verify(operator).dropReplicationTables();
+    } else {
+      verify(operator, never()).dropReplicationTables();
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      unload_WithConfigPropertiesAndNullSchemaFileWithoutDeleteCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      unload_WithConfigPropertiesAndNullSchemaFileWithoutDeleteCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.unload(configProperties, (Path) null, false);
+    SchemaLoader.unload(configProperties, (Path) null, false, withReplicationTables);
 
     // Assert
     verify(parser, never()).parse();
     verify(operator).deleteTables(anyList());
     verify(operator, never()).dropCoordinatorTables();
+    if (withReplicationTables) {
+      verify(operator).dropReplicationTables();
+    } else {
+      verify(operator, never()).dropReplicationTables();
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      unload_WithConfigPropertiesAndSerializedSchemaJsonWithDeleteCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      unload_WithConfigPropertiesAndSerializedSchemaJsonWithDeleteCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.unload(configProperties, SERIALIZED_SCHEMA_JSON, true);
+    SchemaLoader.unload(configProperties, SERIALIZED_SCHEMA_JSON, true, withReplicationTables);
 
     // Assert
     verify(parser).parse();
     verify(operator).deleteTables(anyList());
     verify(operator).dropCoordinatorTables();
+    if (withReplicationTables) {
+      verify(operator).dropReplicationTables();
+    } else {
+      verify(operator, never()).dropReplicationTables();
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      unload_WithConfigPropertiesAndSerializedSchemaJsonWithoutDeleteCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      unload_WithConfigPropertiesAndSerializedSchemaJsonWithoutDeleteCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.unload(configProperties, SERIALIZED_SCHEMA_JSON, false);
+    SchemaLoader.unload(configProperties, SERIALIZED_SCHEMA_JSON, false, withReplicationTables);
 
     // Assert
     verify(parser).parse();
     verify(operator).deleteTables(anyList());
     verify(operator, never()).dropCoordinatorTables();
+    if (withReplicationTables) {
+      verify(operator).dropReplicationTables();
+    } else {
+      verify(operator, never()).dropReplicationTables();
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      unload_WithConfigPropertiesAndNullSerializedSchemaJsonWithDeleteCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      unload_WithConfigPropertiesAndNullSerializedSchemaJsonWithDeleteCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.unload(configProperties, (String) null, true);
+    SchemaLoader.unload(configProperties, (String) null, true, withReplicationTables);
 
     // Assert
     verify(parser, never()).parse();
     verify(operator).deleteTables(anyList());
     verify(operator).dropCoordinatorTables();
+    if (withReplicationTables) {
+      verify(operator).dropReplicationTables();
+    } else {
+      verify(operator, never()).dropReplicationTables();
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      unload_WithConfigPropertiesAndNullSerializedSchemaJsonWithoutDeleteCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      unload_WithConfigPropertiesAndNullSerializedSchemaJsonWithoutDeleteCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.unload(configProperties, (String) null, false);
+    SchemaLoader.unload(configProperties, (String) null, false, withReplicationTables);
 
     // Assert
     verify(parser, never()).parse();
     verify(operator).deleteTables(anyList());
     verify(operator, never()).dropCoordinatorTables();
+    if (withReplicationTables) {
+      verify(operator).dropReplicationTables();
+    } else {
+      verify(operator, never()).dropReplicationTables();
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      repairAll_WithConfigFilePathAndSerializedSchemaAndDoRepairCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      repairAll_WithConfigFilePathAndSerializedSchemaAndDoRepairCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.repairAll(configFilePath, SERIALIZED_SCHEMA_JSON, options, true);
+    SchemaLoader.repairAll(
+        configFilePath, SERIALIZED_SCHEMA_JSON, options, true, withReplicationTables);
 
     // Assert
     verify(parser).parse();
     verify(operator).repairNamespaces(anyList());
     verify(operator).repairTables(anyList());
     verify(operator).repairCoordinatorTables(options);
+    if (withReplicationTables) {
+      verify(operator).repairReplicationTables(options);
+    } else {
+      verify(operator, never()).repairReplicationTables(options);
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      repairAll_WithConfigFilePathAndSerializedSchemaAndDoNotRepairCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      repairAll_WithConfigFilePathAndSerializedSchemaAndDoNotRepairCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.repairAll(configFilePath, SERIALIZED_SCHEMA_JSON, options, false);
+    SchemaLoader.repairAll(
+        configFilePath, SERIALIZED_SCHEMA_JSON, options, false, withReplicationTables);
 
     // Assert
     verify(parser).parse();
     verify(operator).repairNamespaces(anyList());
     verify(operator).repairTables(anyList());
     verify(operator, never()).repairCoordinatorTables(anyMap());
+    if (withReplicationTables) {
+      verify(operator).repairReplicationTables(options);
+    } else {
+      verify(operator, never()).repairReplicationTables(options);
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      repairAll_WithConfigPropertiesAndSerializedSchemaAndDoRepairCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      repairAll_WithConfigPropertiesAndSerializedSchemaAndDoRepairCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.repairAll(configProperties, SERIALIZED_SCHEMA_JSON, options, true);
+    SchemaLoader.repairAll(
+        configProperties, SERIALIZED_SCHEMA_JSON, options, true, withReplicationTables);
 
     // Assert
     verify(parser).parse();
     verify(operator).repairNamespaces(anyList());
     verify(operator).repairTables(anyList());
     verify(operator).repairCoordinatorTables(options);
+    if (withReplicationTables) {
+      verify(operator).repairReplicationTables(options);
+    } else {
+      verify(operator, never()).repairReplicationTables(options);
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      repairAll_WithConfigPropertiesAndSerializedSchemaAndDoNotRepairCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      repairAll_WithConfigPropertiesAndSerializedSchemaAndDoNotRepairCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.repairAll(configProperties, SERIALIZED_SCHEMA_JSON, options, false);
+    SchemaLoader.repairAll(
+        configProperties, SERIALIZED_SCHEMA_JSON, options, false, withReplicationTables);
 
     // Assert
     verify(parser).parse();
     verify(operator).repairNamespaces(anyList());
     verify(operator).repairTables(anyList());
     verify(operator, never()).repairCoordinatorTables(anyMap());
+    if (withReplicationTables) {
+      verify(operator).repairReplicationTables(options);
+    } else {
+      verify(operator, never()).repairReplicationTables(options);
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      repairAll_WithConfigPropertiesAndSchemaFilePathAndDoRepairCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      repairAll_WithConfigPropertiesAndSchemaFilePathAndDoRepairCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.repairAll(configProperties, schemaFilePath, options, true);
+    SchemaLoader.repairAll(configProperties, schemaFilePath, options, true, withReplicationTables);
 
     // Assert
     verify(parser).parse();
     verify(operator).repairNamespaces(anyList());
     verify(operator).repairTables(anyList());
     verify(operator).repairCoordinatorTables(options);
+    if (withReplicationTables) {
+      verify(operator).repairReplicationTables(options);
+    } else {
+      verify(operator, never()).repairReplicationTables(options);
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      repairAll_WithConfigPropertiesAndSchemaFilePathAndDoNotRepairCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      repairAll_WithConfigPropertiesAndSchemaFilePathAndDoNotRepairCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.repairAll(configProperties, schemaFilePath, options, false);
+    SchemaLoader.repairAll(configProperties, schemaFilePath, options, false, withReplicationTables);
 
     // Assert
     verify(parser).parse();
     verify(operator).repairNamespaces(anyList());
     verify(operator).repairTables(anyList());
     verify(operator, never()).repairCoordinatorTables(anyMap());
+    if (withReplicationTables) {
+      verify(operator).repairReplicationTables(options);
+    } else {
+      verify(operator, never()).repairReplicationTables(options);
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      repairAll_WithConfigFilePathAndSchemaFilePathAndDoRepairCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      repairAll_WithConfigFilePathAndSchemaFilePathAndDoRepairCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.repairAll(configFilePath, schemaFilePath, options, true);
+    SchemaLoader.repairAll(configFilePath, schemaFilePath, options, true, withReplicationTables);
 
     // Assert
     verify(parser).parse();
     verify(operator).repairNamespaces(anyList());
     verify(operator).repairTables(anyList());
     verify(operator).repairCoordinatorTables(options);
+    if (withReplicationTables) {
+      verify(operator).repairReplicationTables(options);
+    } else {
+      verify(operator, never()).repairReplicationTables(options);
+    }
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
   public void
-      repairAll_WithConfigFilePathAndSchemaFilePathAndDoNotRepairCoordinatorTables_ShouldCallParserAndOperatorProperly()
-          throws Exception {
+      repairAll_WithConfigFilePathAndSchemaFilePathAndDoNotRepairCoordinatorTables_ShouldCallParserAndOperatorProperly(
+          boolean withReplicationTables) throws Exception {
     // Arrange
 
     // Act
-    SchemaLoader.repairAll(configFilePath, schemaFilePath, options, false);
+    SchemaLoader.repairAll(configFilePath, schemaFilePath, options, false, withReplicationTables);
 
     // Assert
     verify(parser).parse();
     verify(operator).repairNamespaces(anyList());
     verify(operator).repairTables(anyList());
     verify(operator, never()).repairCoordinatorTables(anyMap());
+    if (withReplicationTables) {
+      verify(operator).repairReplicationTables(options);
+    } else {
+      verify(operator, never()).repairReplicationTables(options);
+    }
   }
 
   @Test
diff --git a/schema-loader/src/test/java/com/scalar/db/schemaloader/SchemaOperatorTest.java b/schema-loader/src/test/java/com/scalar/db/schemaloader/SchemaOperatorTest.java
index 9bc58c13fa..e45aec7e56 100644
--- a/schema-loader/src/test/java/com/scalar/db/schemaloader/SchemaOperatorTest.java
+++ b/schema-loader/src/test/java/com/scalar/db/schemaloader/SchemaOperatorTest.java
@@ -123,6 +123,36 @@ public void createTables_WithNonTransactionTables_ShouldCallProperMethods() thro
     verify(transactionAdmin, never()).createCoordinatorTables(options);
   }
 
+  @Test
+  public void
+      createReplicationTables_IfReplicationTablesNotExist_ShouldCallCreateReplicationTables()
+          throws Exception {
+    // Arrange
+    when(transactionAdmin.replicationTablesExist()).thenReturn(false);
+
+    // Act
+    operator.createReplicationTables(options);
+
+    // Assert
+    verify(transactionAdmin).replicationTablesExist();
+    verify(transactionAdmin).createReplicationTables(options);
+  }
+
+  @Test
+  public void
+      createReplicationTables_IfReplicationTablesExist_ShouldNotCallCreateReplicationTables()
+          throws Exception {
+    // Arrange
+    when(transactionAdmin.replicationTablesExist()).thenReturn(true);
+
+    // Act
+    operator.createReplicationTables(options);
+
+    // Assert
+    verify(transactionAdmin).replicationTablesExist();
+    verify(transactionAdmin, never()).createReplicationTables(options);
+  }
+
   @Test
   public void deleteTables_WithTableList_ShouldCallProperMethods() throws Exception {
     // Arrange
@@ -197,6 +227,34 @@ public void dropCoordinatorTables_IfCoordinatorTablesNotExist_ShouldCallNotDropC
     verify(transactionAdmin, never()).dropCoordinatorTables();
   }
 
+  @Test
+  public void dropReplicationTables_IfReplicationTablesExist_ShouldCallDropReplicationTables()
+      throws Exception {
+    // Arrange
+    when(transactionAdmin.replicationTablesExist()).thenReturn(true);
+
+    // Act
+    operator.dropReplicationTables();
+
+    // Assert
+    verify(transactionAdmin).replicationTablesExist();
+    verify(transactionAdmin).dropReplicationTables();
+  }
+
+  @Test
+  public void dropReplicationTables_IfReplicationTablesNotExist_ShouldCallNotDropReplicationTables()
+      throws Exception {
+    // Arrange
+    when(transactionAdmin.replicationTablesExist()).thenReturn(false);
+
+    // Act
+    operator.dropReplicationTables();
+
+    // Assert
+    verify(transactionAdmin).replicationTablesExist();
+    verify(transactionAdmin, never()).dropReplicationTables();
+  }
+
   @Test
   public void repairTables_WithTransactionTables_ShouldCallProperMethods() throws Exception {
     // Arrange
@@ -247,6 +305,18 @@ public void repairCoordinatorTables_ShouldCallProperMethods() throws Exception {
     verifyNoInteractions(storageAdmin);
   }
 
+  @Test
+  public void repairReplicationTables_ShouldCallProperMethods() throws Exception {
+    // Arrange
+
+    // Act
+    operator.repairReplicationTables(options);
+
+    // Assert
+    verify(transactionAdmin).repairReplicationTables(options);
+    verifyNoInteractions(storageAdmin);
+  }
+
   @Test
   public void alterTables_WithoutChanges_ShouldDoNothing()
       throws SchemaLoaderException, ExecutionException {
diff --git a/schema-loader/src/test/java/com/scalar/db/schemaloader/command/SchemaLoaderCommandTest.java b/schema-loader/src/test/java/com/scalar/db/schemaloader/command/SchemaLoaderCommandTest.java
index aa108b559e..d9912e6900 100644
--- a/schema-loader/src/test/java/com/scalar/db/schemaloader/command/SchemaLoaderCommandTest.java
+++ b/schema-loader/src/test/java/com/scalar/db/schemaloader/command/SchemaLoaderCommandTest.java
@@ -85,11 +85,12 @@ public void call_WithProperArgumentsForCreatingTables_ShouldCallLoadProperly() {
         "--no-backup",
         "-f",
         schemaFile,
-        "--coordinator");
+        "--coordinator",
+        "--replication-tables");
 
     // Assert
     schemaLoaderMockedStatic.verify(
-        () -> SchemaLoader.load(Paths.get(configFile), Paths.get(schemaFile), options, true));
+        () -> SchemaLoader.load(Paths.get(configFile), Paths.get(schemaFile), options, true, true));
   }
 
   @Test
@@ -121,11 +122,51 @@ public void call_WithProperArgumentsForCreatingTables_ShouldCallLoadProperly() {
         "--no-scaling",
         "--no-backup",
         "-f",
-        schemaFile);
+        schemaFile,
+        "--replication-tables");
+
+    // Assert
+    schemaLoaderMockedStatic.verify(
+        () ->
+            SchemaLoader.load(Paths.get(configFile), Paths.get(schemaFile), options, false, true));
+  }
+
+  @Test
+  public void
+      call_WithProperArgumentsForCreatingTablesWithoutReplicationTablesArgument_ShouldCallLoadProperly() {
+    // Arrange
+    Map<String, String> options =
+        ImmutableMap.<String, String>builder()
+            .put(CassandraAdmin.REPLICATION_STRATEGY, replicationStrategy)
+            .put(CassandraAdmin.COMPACTION_STRATEGY, compactionStrategy)
+            .put(CassandraAdmin.REPLICATION_FACTOR, replicationFactor)
+            .put(DynamoAdmin.REQUEST_UNIT, ru)
+            .put(DynamoAdmin.NO_SCALING, noScaling.toString())
+            .put(DynamoAdmin.NO_BACKUP, noBackup.toString())
+            .build();
+
+    // Act
+    commandLine.execute(
+        "--config",
+        configFile,
+        "--replication-strategy",
+        replicationStrategy,
+        "--compaction-strategy",
+        compactionStrategy,
+        "--replication-factor",
+        replicationFactor,
+        "--ru",
+        ru,
+        "--no-scaling",
+        "--no-backup",
+        "-f",
+        schemaFile,
+        "--coordinator");
 
     // Assert
     schemaLoaderMockedStatic.verify(
-        () -> SchemaLoader.load(Paths.get(configFile), Paths.get(schemaFile), options, false));
+        () ->
+            SchemaLoader.load(Paths.get(configFile), Paths.get(schemaFile), options, true, false));
   }
 
   @Test
@@ -160,7 +201,7 @@ public void call_WithProperArgumentsForCreatingTables_ShouldCallLoadProperly() {
 
     // Assert
     schemaLoaderMockedStatic.verify(
-        () -> SchemaLoader.load(Paths.get(configFile), (Path) null, options, true));
+        () -> SchemaLoader.load(Paths.get(configFile), (Path) null, options, true, false));
   }
 
   @Test
@@ -170,11 +211,12 @@ public void call_WithProperArgumentsForDeletingTables_ShouldCallUnloadProperly()
     String configFile = "path_to_config_file";
 
     // Act
-    commandLine.execute("-f", schemaFile, "-D", "--config", configFile, "--coordinator");
+    commandLine.execute(
+        "-f", schemaFile, "-D", "--config", configFile, "--coordinator", "--replication-tables");
 
     // Assert
     schemaLoaderMockedStatic.verify(
-        () -> SchemaLoader.unload(Paths.get(configFile), Paths.get(schemaFile), true));
+        () -> SchemaLoader.unload(Paths.get(configFile), Paths.get(schemaFile), true, true));
   }
 
   @Test
@@ -185,11 +227,26 @@ public void call_WithProperArgumentsForDeletingTables_ShouldCallUnloadProperly()
     String configFile = "path_to_config_file";
 
     // Act
-    commandLine.execute("-f", schemaFile, "-D", "--config", configFile);
+    commandLine.execute("-f", schemaFile, "-D", "--config", configFile, "--replication-tables");
 
     // Assert
     schemaLoaderMockedStatic.verify(
-        () -> SchemaLoader.unload(Paths.get(configFile), Paths.get(schemaFile), false));
+        () -> SchemaLoader.unload(Paths.get(configFile), Paths.get(schemaFile), false, true));
+  }
+
+  @Test
+  public void
+      call_WithProperArgumentsForDeletingTablesWithoutReplicationTablesArgument_ShouldCallUnloadProperly() {
+    // Arrange
+    String schemaFile = "path_to_file";
+    String configFile = "path_to_config_file";
+
+    // Act
+    commandLine.execute("-f", schemaFile, "-D", "--config", configFile, "--coordinator");
+
+    // Assert
+    schemaLoaderMockedStatic.verify(
+        () -> SchemaLoader.unload(Paths.get(configFile), Paths.get(schemaFile), true, false));
   }
 
   @Test
@@ -205,7 +262,9 @@ public void call_WithProperArgumentsForDeletingTables_ShouldCallUnloadProperly()
 
     // Assert
     schemaLoaderMockedStatic.verify(
-        () -> SchemaLoader.repairAll(Paths.get(configFile), Paths.get(schemaFile), options, false));
+        () ->
+            SchemaLoader.repairAll(
+                Paths.get(configFile), Paths.get(schemaFile), options, false, false));
   }
 
   @Test
@@ -222,7 +281,34 @@ public void call_WithProperArgumentsForDeletingTables_ShouldCallUnloadProperly()
 
     // Assert
     schemaLoaderMockedStatic.verify(
-        () -> SchemaLoader.repairAll(Paths.get(configFile), Paths.get(schemaFile), options, true));
+        () ->
+            SchemaLoader.repairAll(
+                Paths.get(configFile), Paths.get(schemaFile), options, true, false));
+  }
+
+  @Test
+  public void
+      call_WithProperArgumentsForRepairingAllWithReplicationTablesArgument_ShouldCallRepairAllProperly() {
+    // Arrange
+    String schemaFile = "path_to_file";
+    String configFile = "path_to_config_file";
+    Map<String, String> options = ImmutableMap.of(DynamoAdmin.NO_BACKUP, noBackup.toString());
+
+    // Act
+    commandLine.execute(
+        "-f",
+        schemaFile,
+        "--repair-all",
+        "--config",
+        configFile,
+        "--replication-tables",
+        "--no-backup");
+
+    // Assert
+    schemaLoaderMockedStatic.verify(
+        () ->
+            SchemaLoader.repairAll(
+                Paths.get(configFile), Paths.get(schemaFile), options, false, true));
   }
 
   @Test
@@ -250,7 +336,7 @@ public void call_forRepairingAllWithoutSchemaFile_ShouldThrowIllegalArgumentExce
 
     // Assert
     schemaLoaderMockedStatic.verify(
-        () -> SchemaLoader.unload(Paths.get(configFile), (Path) null, true));
+        () -> SchemaLoader.unload(Paths.get(configFile), (Path) null, true, false));
   }
 
   @Test