[590] Add Iceberg HMS Catalog Sync implementation

pom.xml

@@ -51,6 +51,7 @@
         <module>xtable-hudi-support</module>
         <module>xtable-core</module>
         <module>xtable-utilities</module>
+        <module>xtable-hive-metastore</module>
     </modules>
 
     <properties>
@@ -61,8 +62,10 @@
         <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.4.1</hadoop.version>
         <hudi.version>0.14.0</hudi.version>
+        <aws.version>2.29.40</aws.version>
+        <hive.version>2.3.9</hive.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>
@@ -371,19 +374,21 @@
                 <version>${hadoop.version}</version>
                 <scope>runtime</scope>
             </dependency>
-            <dependency>
-                <groupId>com.amazonaws</groupId>
-                <artifactId>aws-java-sdk-bundle</artifactId>
-                <version>1.12.328</version>
-                <scope>runtime</scope>
-            </dependency>
             <dependency>
                 <groupId>com.google.cloud.bigdataoss</groupId>
                 <artifactId>gcs-connector</artifactId>
                 <version>hadoop3-2.2.22</version>
                 <scope>runtime</scope>
             </dependency>
 
+            <!-- AWS -->
+            <dependency>
+                <groupId>software.amazon.awssdk</groupId>
+                <artifactId>bundle</artifactId>
+                <version>${aws.version}</version>
+                <scope>runtime</scope>
+            </dependency>
+
             <!-- Protobuf -->
             <dependency>
                 <groupId>com.google.protobuf</groupId>

rfc/template.md

@@ -0,0 +1,55 @@
+<!--
+  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.
+-->
+# RFC-[number]: [Title]
+
+## Proposers
+
+- @<proposer1 github username>
+- @<proposer2 github username>
+
+## Approvers
+- @<approver1 github username>
+- @<approver2 github username>
+
+## Status
+
+GH Feature Request: <link to umbrella JIRA>
+
+> Please keep the status updated in `rfc/README.md`.
+
+## Abstract
+
+Describe the problem you are trying to solve and a brief description of why it’s needed.
+
+## Background
+Introduce any background context which is relevant or necessary to understand the feature and design choices.
+
+## Implementation
+Describe the new thing you want to do in appropriate detail, how it fits into the project architecture.<br>
+Provide a detailed description of how you intend to implement this feature, this may be fairly extensive and have large subsections of its own or it may be a few sentences.<br>
+Use judgement to decide on how detailed the description needs to be based on the scope of the change. If unclear, you can ask questions in [email protected].
+
+## Rollout/Adoption Plan
+
+- Are there any breaking changes as part of this new feature/functionality?
+- What impact (if any) will there be on existing users?
+- If we are changing behavior how will we phase out the older behavior? When will we remove the existing behavior? 
+- If we need special migration tools, describe them here.
+
+## Test Plan
+
+Describe in few sentences how the RFC will be tested. How will we know that the implementation works as expected? How will we know nothing breaks?

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

@@ -61,5 +61,5 @@ public class ExternalCatalogConfig {
   /**
    * The properties for this catalog, used for providing any custom behaviour during catalog sync
    */
-  @NonNull @Builder.Default Map<String, String> catalogProperties = Collections.emptyMap();
+  @Builder.Default Map<String, String> catalogProperties = Collections.emptyMap();
 }

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

@@ -25,4 +25,5 @@
  */
 public class CatalogType {
   public static final String STORAGE = "STORAGE";
+  public static final String HMS = "HMS";
 }

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/TestThreePartHierarchicalTableIdentifier.java

@@ -23,11 +23,11 @@
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
-class TestThreePartHierarchicalTableIdentifier {
+class TestHierarchicalTableIdentifier {
 
   @Test
-  void testGetId() {
-    ThreePartHierarchicalTableIdentifier catalogTableIdentifier =
+  void testToString() {
+    HierarchicalTableIdentifier catalogTableIdentifier =
         ThreePartHierarchicalTableIdentifier.fromDotSeparatedIdentifier(
             "catalogName.databaseName.tableName");
     assertEquals("catalogName.databaseName.tableName", catalogTableIdentifier.getId());

xtable-core/pom.xml

@@ -144,6 +144,10 @@
             <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/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/TableFormatUtils.java

@@ -0,0 +1,89 @@
+/*
+ * 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.Locale;
+import java.util.Map;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.iceberg.TableProperties;
+
+import com.google.common.base.Strings;
+
+import org.apache.xtable.exception.NotSupportedException;
+import org.apache.xtable.model.storage.TableFormat;
+
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+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 from given table properties
+   *
+   * @param properties catalog table properties
+   * @return table format name
+   *     <li>In case of ICEBERG, table_type param will give the table format
+   *     <li>In case of DELTA, table_type or spark.sql.sources.provider param will give the table
+   *         format
+   *     <li>In case of HUDI, spark.sql.sources.provider param will give the table format
+   */
+  public static String getTableFormat(Map<String, String> properties) {
+    String tableFormat = properties.get(TABLE_TYPE_PROP);
+    if (Strings.isNullOrEmpty(tableFormat)) {
+      tableFormat = properties.get(PROP_SPARK_SQL_SOURCES_PROVIDER);
+    }
+    if (Strings.isNullOrEmpty(tableFormat)) {
+      throw new IllegalArgumentException("Invalid TableFormat: null or empty");
+    }
+    return tableFormat.toUpperCase(Locale.ENGLISH);
+  }
+}

xtable-core/src/main/java/org/apache/xtable/exception/CatalogSyncException.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.exception;
+
+import org.apache.xtable.model.exception.ErrorCode;
+import org.apache.xtable.model.exception.InternalException;
+
+public class CatalogSyncException extends InternalException {
+
+  public CatalogSyncException(ErrorCode errorCode, String message, Throwable e) {
+    super(errorCode, message, e);
+  }
+
+  public CatalogSyncException(String message, Throwable e) {
+    super(ErrorCode.CATALOG_SYNC_GENERIC_EXCEPTION, message, e);
+  }
+}