[590] Add Iceberg Glue Catalog Sync implementation

pom.xml

@@ -61,8 +61,9 @@
         <junit.version>5.9.0</junit.version>
         <lombok.version>1.18.30</lombok.version>
         <lombok-maven-plugin.version>1.18.20.0</lombok-maven-plugin.version>
-        <hadoop.version>3.4.0</hadoop.version>
+        <hadoop.version>3.3.6</hadoop.version>
         <hudi.version>0.14.0</hudi.version>
+        <aws.version>2.28.22</aws.version>
         <maven-source-plugin.version>3.3.1</maven-source-plugin.version>
         <maven-javadoc-plugin.version>3.8.0</maven-javadoc-plugin.version>
         <maven-gpg-plugin.version>3.2.4</maven-gpg-plugin.version>
@@ -372,9 +373,9 @@
                 <scope>runtime</scope>
             </dependency>
             <dependency>
-                <groupId>com.amazonaws</groupId>
-                <artifactId>aws-java-sdk-bundle</artifactId>
-                <version>1.12.328</version>
+                <groupId>software.amazon.awssdk</groupId>
+                <artifactId>bundle</artifactId>
+                <version>${aws.version}</version>
                 <scope>runtime</scope>
             </dependency>
             <dependency>

xtable-api/src/main/java/org/apache/xtable/model/exception/ErrorCode.java

@@ -31,7 +31,8 @@ public enum ErrorCode {
   UNSUPPORTED_SCHEMA_TYPE(10007),
   UNSUPPORTED_FEATURE(10008),
   PARSE_EXCEPTION(10009),
-  CATALOG_REFRESH_EXCEPTION(10010);
+  CATALOG_REFRESH_EXCEPTION(10010),
+  CATALOG_SYNC_GENERIC_EXCEPTION(10011);
 
   private final int errorCode;
 

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

@@ -0,0 +1,23 @@
+/*
+ * 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;
+
+public class CatalogType {
+  public static final String GLUE = "GLUE";
+}

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

@@ -83,6 +83,11 @@ private <TABLE> CatalogSyncStatus syncCatalog(
       CatalogSyncClient<TABLE> catalogSyncClient,
       CatalogTableIdentifier tableIdentifier,
       InternalTable table) {
+    log.info(
+        "Running catalog sync for table {} with format {} using catalogSync {}",
+        table.getBasePath(),
+        table.getTableFormat(),
+        catalogSyncClient.getClass().getName());
     if (!catalogSyncClient.hasDatabase(tableIdentifier)) {
       catalogSyncClient.createDatabase(tableIdentifier);
     }

xtable-api/src/test/java/org/apache/xtable/model/catalog/TestHierarchicalTableIdentifier.java

@@ -0,0 +1,50 @@
+/*
+ * 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.catalog;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class TestHierarchicalTableIdentifier {
+
+  @Test
+  void testToString() {
+    HierarchicalTableIdentifier catalogTableIdentifier =
+        ThreePartHierarchicalTableIdentifier.fromDotSeparatedIdentifier(
+            "catalogName.databaseName.tableName");
+    assertEquals("catalogName.databaseName.tableName", catalogTableIdentifier.getId());
+  }
+
+  @Test
+  void testConstructorForHierarchicalTableIdentifier() {
+    Assertions.assertDoesNotThrow(
+        () ->
+            ThreePartHierarchicalTableIdentifier.fromDotSeparatedIdentifier(
+                "catalogName.databaseName.tableName"));
+    Assertions.assertDoesNotThrow(
+        () ->
+            ThreePartHierarchicalTableIdentifier.fromDotSeparatedIdentifier(
+                "databaseName.tableName"));
+    Assertions.assertThrows(
+        IllegalArgumentException.class,
+        () -> ThreePartHierarchicalTableIdentifier.fromDotSeparatedIdentifier("tableName"));
+  }
+}

xtable-core/pom.xml

@@ -138,12 +138,23 @@
             <scope>provided</scope>
         </dependency>
 
+        <!-- AWS Glue dependencies -->
+        <dependency>
+            <groupId>software.amazon.awssdk</groupId>
+            <artifactId>glue</artifactId>
+            <version>${aws.version}</version>
+        </dependency>
+
         <!-- Mockito -->
         <dependency>
             <groupId>org.mockito</groupId>
             <artifactId>mockito-core</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-junit-jupiter</artifactId>
+        </dependency>
 
         <!-- Junit -->
         <dependency>

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

@@ -57,7 +57,7 @@ public CatalogSyncClient createCatalogSyncClient(
     return ReflectionUtils.createInstanceOfClass(
         targetCatalogConfig.getCatalogSyncClientImpl(),
         targetCatalogConfig,
-        tableFormat,
-        configuration);
+        configuration,
+        tableFormat);
   }
 }

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

@@ -0,0 +1,33 @@
+/*
+ * 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.xtable.model.InternalTable;
+import org.apache.xtable.model.catalog.CatalogTableIdentifier;
+
+/**
+ * The interface for creating/updating catalog table object, each catalog can have its own
+ * implementation that can be plugged in.
+ */
+public interface CatalogTableBuilder<REQUEST, TABLE> {
+  public REQUEST getCreateTableRequest(InternalTable table, CatalogTableIdentifier tableIdentifier);
+
+  public REQUEST getUpdateTableRequest(
+      InternalTable table, TABLE catalogTable, CatalogTableIdentifier tableIdentifier);
+}

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

@@ -0,0 +1,33 @@
+/*
+ * 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.xtable.model.catalog.CatalogTableIdentifier;
+import org.apache.xtable.model.catalog.HierarchicalTableIdentifier;
+
+public class CatalogUtils {
+
+  public static HierarchicalTableIdentifier castToHierarchicalTableIdentifier(
+      CatalogTableIdentifier tableIdentifier) {
+    if (tableIdentifier instanceof HierarchicalTableIdentifier) {
+      return (HierarchicalTableIdentifier) tableIdentifier;
+    }
+    throw new IllegalArgumentException("Invalid tableIdentifier implementation");
+  }
+}

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

@@ -0,0 +1,27 @@
+/*
+ * 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;
+
+public class Constants {
+
+  public static final String PROP_SPARK_SQL_SOURCES_PROVIDER = "spark.sql.sources.provider";
+  public static final String PROP_PATH = "path";
+  public static final String PROP_SERIALIZATION_FORMAT = "serialization.format";
+  public static final String PROP_EXTERNAL = "EXTERNAL";
+}

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

@@ -20,16 +20,27 @@
 
 import java.util.Map;
 
+import org.apache.xtable.catalog.glue.GlueCatalogConversionSource;
+import org.apache.xtable.catalog.glue.GlueCatalogSyncClient;
 import org.apache.xtable.conversion.ExternalCatalogConfig;
+import org.apache.xtable.exception.NotSupportedException;
+import org.apache.xtable.model.storage.CatalogType;
 
 /** A factory class which returns {@link ExternalCatalogConfig} based on catalogType. */
 public class ExternalCatalogConfigFactory {
 
   public static ExternalCatalogConfig fromCatalogType(
       String catalogType, String catalogId, Map<String, String> properties) {
-    // TODO: Choose existing implementation based on catalogType.
-    String catalogSyncClientImpl = "";
-    String catalogConversionSourceImpl = "";
+    String catalogSyncClientImpl;
+    String catalogConversionSourceImpl;
+    switch (catalogType) {
+      case CatalogType.GLUE:
+        catalogSyncClientImpl = GlueCatalogSyncClient.class.getName();
+        catalogConversionSourceImpl = GlueCatalogConversionSource.class.getName();
+        break;
+      default:
+        throw new NotSupportedException("Unsupported catalogType: " + catalogType);
+    }
     return ExternalCatalogConfig.builder()
         .catalogType(catalogType)
         .catalogSyncClientImpl(catalogSyncClientImpl)

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

@@ -0,0 +1,76 @@
+/*
+ * 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 static org.apache.iceberg.BaseMetastoreTableOperations.TABLE_TYPE_PROP;
+import static org.apache.xtable.catalog.Constants.PROP_SPARK_SQL_SOURCES_PROVIDER;
+
+import java.util.Map;
+
+import org.apache.iceberg.TableProperties;
+
+import com.google.common.base.Strings;
+
+import org.apache.xtable.exception.NotSupportedException;
+import org.apache.xtable.model.storage.TableFormat;
+
+public class TableFormatUtils {
+
+  public static String getTableDataLocation(
+      String tableFormat, String tableLocation, Map<String, String> properties) {
+    switch (tableFormat) {
+      case TableFormat.ICEBERG:
+        return getIcebergDataLocation(tableLocation, properties);
+      case TableFormat.DELTA:
+      case TableFormat.HUDI:
+        return tableLocation;
+      default:
+        throw new NotSupportedException("Unsupported table format: " + tableFormat);
+    }
+  }
+
+  /** Get iceberg table data files location */
+  private static String getIcebergDataLocation(
+      String tableLocation, Map<String, String> properties) {
+    String dataLocation = properties.get(TableProperties.WRITE_DATA_LOCATION);
+    if (dataLocation == null) {
+      dataLocation = properties.get(TableProperties.WRITE_FOLDER_STORAGE_LOCATION);
+      if (dataLocation == null) {
+        dataLocation = properties.get(TableProperties.OBJECT_STORE_PATH);
+        if (dataLocation == null) {
+          dataLocation = String.format("%s/data", tableLocation);
+        }
+      }
+    }
+    return dataLocation;
+  }
+
+  // Get table format name from table properties
+  public static String getTableFormat(Map<String, String> properties) {
+    //  - In case of ICEBERG, table_type param will give the table format
+    //  - In case of DELTA, table_type or spark.sql.sources.provider param will give the table
+    // format
+    //  - In case of HUDI, spark.sql.sources.provider param will give the table format
+    String tableFormat = properties.get(TABLE_TYPE_PROP);
+    if (Strings.isNullOrEmpty(tableFormat)) {
+      tableFormat = properties.get(PROP_SPARK_SQL_SOURCES_PROVIDER);
+    }
+    return tableFormat;
+  }
+}