[590] Add RunCatalogSync utility for synchronizing tables across cata…

…logs

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

@@ -19,6 +19,7 @@
 package org.apache.xtable.conversion;
 
 import java.util.List;
+import java.util.Map;
 
 import lombok.Builder;
 import lombok.NonNull;
@@ -34,14 +35,21 @@ public class ConversionConfig {
   @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
+  // targetTableIdentifier to target catalogs.
+  Map<String, 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<String, List<TargetCatalogConfig>> targetCatalogs,
+      SyncMode syncMode) {
     this.sourceTable = sourceTable;
     this.targetTables = targetTables;
+    this.targetCatalogs = targetCatalogs;
     Preconditions.checkArgument(
         targetTables != null && !targetTables.isEmpty(),
         "Please provide at-least one format to sync");

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

@@ -0,0 +1,42 @@
+/*
+ * 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. */
+@Value
+@Builder
+public class ExternalCatalogConfig implements CatalogConfig {
+  /** The name of the catalog, it also acts as a unique identifier for each catalog */
+  @NonNull String catalogName;
+
+  /** The implementation class path for the catalog */
+  @NonNull String catalogImpl;
+
+  /**
+   * The properties for each catalog, used for providing any custom behaviour during catalog sync
+   */
+  @NonNull @Builder.Default Map<String, String> catalogOptions = Collections.emptyMap();
+}

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

@@ -0,0 +1,42 @@
+/*
+ * 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 tableIdentifiers(databaseName, tableName) 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/conversion/TargetTable.java

@@ -44,4 +44,8 @@ public TargetTable(
     this.metadataRetention =
         metadataRetention == null ? Duration.of(7, ChronoUnit.DAYS) : metadataRetention;
   }
+
+  public String getId() {
+    return String.format("%s#%s", sanitizeBasePath(this.basePath), formatName);
+  }
 }

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

@@ -0,0 +1,60 @@
+/*
+ * 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 org.apache.hadoop.conf.Configuration;
+
+import org.apache.xtable.conversion.ExternalCatalogConfig;
+import org.apache.xtable.reflection.ReflectionUtils;
+import org.apache.xtable.spi.extractor.CatalogConversionSource;
+import org.apache.xtable.spi.sync.CatalogSyncClient;
+
+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) {
+    return ReflectionUtils.createInstanceOfClass(
+        sourceCatalogConfig.getCatalogImpl(), 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 CatalogSyncClient createCatalogSyncClient(
+      ExternalCatalogConfig targetCatalogConfig, Configuration configuration) {
+    return ReflectionUtils.createInstanceOfClass(
+        targetCatalogConfig.getCatalogImpl(), targetCatalogConfig, configuration);
+  }
+}

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

@@ -0,0 +1,38 @@
+/*
+ * 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.Map;
+
+import org.apache.xtable.conversion.ExternalCatalogConfig;
+
+/** A factory class which returns {@link ExternalCatalogConfig} based on catalogType. */
+public class ExternalCatalogConfigFactory {
+
+  public static ExternalCatalogConfig fromCatalogType(
+      String catalogType, String catalogName, Map<String, String> properties) {
+    // TODO: Choose existing implementation based on catalogType.
+    String catalogImpl = "";
+    return ExternalCatalogConfig.builder()
+        .catalogImpl(catalogImpl)
+        .catalogName(catalogName)
+        .catalogOptions(properties)
+        .build();
+  }
+}