Support --replication-tables option

Author: komamitsu

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);
+      }
     }
   }
 

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) {