[590] Add RunCatalogSync utility for synchronizing tables across catalogs

Author: vinishjail97

xtable-api/src/main/java/org/apache/xtable/conversion/ConversionConfig.java

@@ -18,7 +18,9 @@
 
 package org.apache.xtable.conversion;
 
+import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 
 import lombok.Builder;
 import lombok.NonNull;
@@ -29,22 +31,30 @@
 import org.apache.xtable.model.sync.SyncMode;
 
 @Value
+@Builder
 public class ConversionConfig {
   // The source of the sync
   @NonNull SourceTable sourceTable;
   // One or more targets to sync the table metadata to
   List<TargetTable> targetTables;
+  // Each target table can be synced to multiple target catalogs, this is map from
+  // targetTable to target catalogs.
+  Map<TargetTable, List<TargetCatalogConfig>> targetCatalogs;
   // The mode, incremental or snapshot
   SyncMode syncMode;
 
   @Builder
   ConversionConfig(
-      @NonNull SourceTable sourceTable, List<TargetTable> targetTables, SyncMode syncMode) {
+      @NonNull SourceTable sourceTable,
+      List<TargetTable> targetTables,
+      Map<TargetTable, List<TargetCatalogConfig>> targetCatalogs,
+      SyncMode syncMode) {
     this.sourceTable = sourceTable;
     this.targetTables = targetTables;
     Preconditions.checkArgument(
         targetTables != null && !targetTables.isEmpty(),
         "Please provide at-least one format to sync");
+    this.targetCatalogs = targetCatalogs == null ? Collections.emptyMap() : targetCatalogs;
     this.syncMode = syncMode == null ? SyncMode.INCREMENTAL : syncMode;
   }
 }

xtable-api/src/main/java/org/apache/xtable/conversion/ExternalCatalogConfig.java

@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+ 
+package org.apache.xtable.conversion;
+
+import java.util.Collections;
+import java.util.Map;
+
+import lombok.Builder;
+import lombok.NonNull;
+import lombok.Value;
+
+/**
+ * Defines the configuration for an external catalog, user needs to populate at-least one of {@link
+ * ExternalCatalogConfig#catalogType} or {@link ExternalCatalogConfig#catalogSyncClientImpl}
+ */
+@Value
+@Builder
+public class ExternalCatalogConfig {
+  /**
+   * A user-defined unique identifier for the catalog, allows user to sync table to multiple
+   * catalogs of the same name/type eg: HMS catalog with url1, HMS catalog with url2
+   */
+  @NonNull String catalogId;
+
+  /**
+   * The type of the catalog. If the catalogType implementation exists in XTable, the implementation
+   * class will be inferred.
+   */
+  String catalogType;
+
+  /**
+   * (Optional) A fully qualified class name that implements the interface for {@link
+   * org.apache.xtable.spi.sync.CatalogSyncClient}, it can be used if the implementation for
+   * catalogType doesn't exist in XTable.
+   */
+  String catalogSyncClientImpl;
+
+  /**
+   * (Optional) A fully qualified class name that implements the interface for {@link
+   * org.apache.xtable.spi.extractor.CatalogConversionSource} it can be used if the implementation
+   * for catalogType doesn't exist in XTable.
+   */
+  String catalogConversionSourceImpl;
+
+  /**
+   * The properties for this catalog, used for providing any custom behaviour during catalog sync
+   */
+  @NonNull @Builder.Default Map<String, String> catalogProperties = Collections.emptyMap();
+}

xtable-api/src/main/java/org/apache/xtable/conversion/TargetCatalogConfig.java

@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+ 
+package org.apache.xtable.conversion;
+
+import lombok.Builder;
+import lombok.NonNull;
+import lombok.Value;
+
+import org.apache.xtable.model.catalog.CatalogTableIdentifier;
+
+/**
+ * TargetCatalogConfig contains the parameters that are required when syncing {@link TargetTable} to
+ * a catalog.
+ */
+@Value
+@Builder
+public class TargetCatalogConfig {
+  /** The tableIdentifier that will be used when syncing {@link TargetTable} to the catalog. */
+  @NonNull CatalogTableIdentifier catalogTableIdentifier;
+
+  /** Configuration for the catalog. */
+  @NonNull ExternalCatalogConfig catalogConfig;
+}

xtable-api/src/main/java/org/apache/xtable/model/storage/CatalogType.java

@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+ 
+package org.apache.xtable.model.storage;
+
+/**
+ * Default constants for supported catalog types.
+ *
+ * @since 0.1
+ */
+public class CatalogType {
+  public static final String STORAGE = "STORAGE";
+}

xtable-api/src/main/java/org/apache/xtable/spi/extractor/CatalogConversionSource.java

@@ -29,4 +29,7 @@
 public interface CatalogConversionSource {
   /** Returns the source table object present in the catalog. */
   SourceTable getSourceTable(CatalogTableIdentifier tableIdentifier);
+
+  /** Returns the {@link org.apache.xtable.model.storage.CatalogType} for the catalog conversion */
+  String getCatalogType();
 }

xtable-api/src/main/java/org/apache/xtable/spi/sync/CatalogSyncClient.java

@@ -34,6 +34,9 @@ public interface CatalogSyncClient<TABLE> extends AutoCloseable {
    */
   String getCatalogId();
 
+  /** Returns the {@link org.apache.xtable.model.storage.CatalogType} the client syncs to */
+  String getCatalogType();
+
   /** Returns the storage location of the table synced to the catalog. */
   String getStorageLocation(TABLE table);
 

xtable-core/pom.xml

@@ -174,4 +174,24 @@
             <scope>test</scope>
         </dependency>
     </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-jar-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>test-jar</goal>
+                        </goals>
+                        <phase>test-compile</phase>
+                    </execution>
+                </executions>
+                <configuration>
+                    <skip>false</skip>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
 </project>

xtable-core/src/main/java/org/apache/xtable/catalog/CatalogConversionFactory.java

@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+ 
+package org.apache.xtable.catalog;
+
+import java.util.ServiceLoader;
+import java.util.function.Function;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+
+import org.apache.xtable.conversion.ExternalCatalogConfig;
+import org.apache.xtable.exception.NotSupportedException;
+import org.apache.xtable.reflection.ReflectionUtils;
+import org.apache.xtable.spi.extractor.CatalogConversionSource;
+import org.apache.xtable.spi.sync.CatalogSyncClient;
+
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class CatalogConversionFactory {
+  private static final CatalogConversionFactory INSTANCE = new CatalogConversionFactory();
+
+  public static CatalogConversionFactory getInstance() {
+    return INSTANCE;
+  }
+
+  /**
+   * Returns an implementation class for {@link CatalogConversionSource} that's used for converting
+   * table definitions in the catalog to {@link org.apache.xtable.conversion.SourceTable} object.
+   *
+   * @param sourceCatalogConfig configuration for the source catalog
+   * @param configuration hadoop configuration
+   */
+  public static CatalogConversionSource createCatalogConversionSource(
+      ExternalCatalogConfig sourceCatalogConfig, Configuration configuration) {
+    if (!StringUtils.isEmpty(sourceCatalogConfig.getCatalogType())) {
+      return findInstance(
+          CatalogConversionSource.class,
+          sourceCatalogConfig.getCatalogType(),
+          CatalogConversionSource::getCatalogType);
+    }
+    return ReflectionUtils.createInstanceOfClass(
+        sourceCatalogConfig.getCatalogConversionSourceImpl(), sourceCatalogConfig, configuration);
+  }
+
+  /**
+   * Returns an implementation class for {@link CatalogSyncClient} that's used for syncing {@link
+   * org.apache.xtable.conversion.TargetTable} to a catalog.
+   *
+   * @param targetCatalogConfig configuration for the target catalog
+   * @param configuration hadoop configuration
+   */
+  public <TABLE> CatalogSyncClient<TABLE> createCatalogSyncClient(
+      ExternalCatalogConfig targetCatalogConfig, String tableFormat, Configuration configuration) {
+    if (!StringUtils.isEmpty(targetCatalogConfig.getCatalogType())) {
+      return findInstance(
+          CatalogSyncClient.class,
+          targetCatalogConfig.getCatalogType(),
+          CatalogSyncClient::getCatalogType);
+    }
+    return ReflectionUtils.createInstanceOfClass(
+        targetCatalogConfig.getCatalogSyncClientImpl(),
+        targetCatalogConfig,
+        tableFormat,
+        configuration);
+  }
+
+  private static <T> T findInstance(
+      Class<T> serviceClass, String catalogType, Function<T, String> catalogTypeExtractor) {
+    ServiceLoader<T> loader = ServiceLoader.load(serviceClass);
+    for (T instance : loader) {
+      String instanceCatalogType = catalogTypeExtractor.apply(instance);
+      if (catalogType.equals(instanceCatalogType)) {
+        return instance;
+      }
+    }
+    throw new NotSupportedException("catalogType is not yet supported: " + catalogType);
+  }
+}