[590] Add Delta HMS Catalog Sync implementation

pom.xml

@@ -88,6 +88,7 @@
         <apache.rat.version>0.16.1</apache.rat.version>
         <google.java.format.version>1.8</google.java.format.version>
         <delta.standalone.version>0.5.0</delta.standalone.version>
+        <delta.hive.version>3.0.0</delta.hive.version>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <target.dir.pattern>**/target/**</target.dir.pattern>
         <delombok.output.dir>${project.build.directory}/delombok</delombok.output.dir>
@@ -280,6 +281,11 @@
                 <version>${iceberg.version}</version>
                 <scope>test</scope>
             </dependency>
+            <dependency>
+                <groupId>org.apache.iceberg</groupId>
+                <artifactId>iceberg-hive-runtime</artifactId>
+                <version>${iceberg.version}</version>
+            </dependency>
 
             <!-- Delta -->
             <dependency>
@@ -291,7 +297,11 @@
                 <groupId>io.delta</groupId>
                 <artifactId>delta-standalone_${scala.binary.version}</artifactId>
                 <version>${delta.standalone.version}</version>
-                <scope>test</scope>
+            </dependency>
+            <dependency>
+                <groupId>io.delta</groupId>
+                <artifactId>delta-hive_${scala.binary.version}</artifactId>
+                <version>${delta.hive.version}</version>
             </dependency>
 
             <!-- Spark -->

xtable-hive-metastore/pom.xml

@@ -46,7 +46,12 @@
         <dependency>
             <groupId>org.apache.iceberg</groupId>
             <artifactId>iceberg-hive-runtime</artifactId>
-            <version>${iceberg.version}</version>
+        </dependency>
+
+        <!-- Delta dependencies  -->
+        <dependency>
+            <groupId>io.delta</groupId>
+            <artifactId>delta-hive_${scala.binary.version}</artifactId>
         </dependency>
 
         <!-- HMS dependencies -->

xtable-hive-metastore/src/main/java/org/apache/xtable/hms/HMSCatalogSyncClient.java

@@ -194,8 +194,7 @@ private void _init(
     } catch (MetaException | HiveException e) {
       throw new CatalogSyncException("HiveMetastoreClient could not be created", e);
     }
-    this.tableBuilder =
-        HMSCatalogTableBuilderFactory.getTableBuilder(tableFormat, this.configuration);
+    this.tableBuilder = HMSCatalogTableBuilderFactory.getInstance(tableFormat, this.configuration);
   }
 
   /**

xtable-hive-metastore/src/main/java/org/apache/xtable/hms/HMSCatalogTableBuilderFactory.java

@@ -32,18 +32,21 @@
 
 import org.apache.xtable.catalog.CatalogTableBuilder;
 import org.apache.xtable.exception.NotSupportedException;
+import org.apache.xtable.hms.table.DeltaHMSCatalogTableBuilder;
 import org.apache.xtable.hms.table.IcebergHMSCatalogTableBuilder;
 import org.apache.xtable.model.catalog.CatalogTableIdentifier;
 import org.apache.xtable.model.catalog.HierarchicalTableIdentifier;
 import org.apache.xtable.model.storage.TableFormat;
 
 public class HMSCatalogTableBuilderFactory {
 
-  public static CatalogTableBuilder<Table, Table> getTableBuilder(
+  static CatalogTableBuilder<Table, Table> getInstance(
       String tableFormat, Configuration configuration) {
     switch (tableFormat) {
       case TableFormat.ICEBERG:
         return new IcebergHMSCatalogTableBuilder(configuration);
+      case TableFormat.DELTA:
+        return new DeltaHMSCatalogTableBuilder();
       default:
         throw new NotSupportedException("Unsupported table format: " + tableFormat);
     }

xtable-hive-metastore/src/main/java/org/apache/xtable/hms/table/DeltaHMSCatalogTableBuilder.java

@@ -0,0 +1,95 @@
+/*
+ * 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.hms.table;
+
+import static org.apache.iceberg.BaseMetastoreTableOperations.TABLE_TYPE_PROP;
+import static org.apache.xtable.catalog.Constants.PROP_EXTERNAL;
+import static org.apache.xtable.catalog.Constants.PROP_PATH;
+import static org.apache.xtable.catalog.Constants.PROP_SERIALIZATION_FORMAT;
+import static org.apache.xtable.hms.HMSCatalogTableBuilderFactory.newHmsTable;
+
+import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
+
+import org.apache.hadoop.hive.metastore.api.SerDeInfo;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import io.delta.hive.DeltaStorageHandler;
+
+import org.apache.xtable.catalog.CatalogTableBuilder;
+import org.apache.xtable.hms.HMSSchemaExtractor;
+import org.apache.xtable.model.InternalTable;
+import org.apache.xtable.model.catalog.CatalogTableIdentifier;
+import org.apache.xtable.model.storage.TableFormat;
+
+public class DeltaHMSCatalogTableBuilder implements CatalogTableBuilder<Table, Table> {
+
+  private final HMSSchemaExtractor schemaExtractor;
+  private static final String tableFormat = TableFormat.DELTA;
+
+  public DeltaHMSCatalogTableBuilder() {
+    this.schemaExtractor = HMSSchemaExtractor.getInstance();
+  }
+
+  @Override
+  public Table getCreateTableRequest(InternalTable table, CatalogTableIdentifier tableIdentifier) {
+    return newHmsTable(tableIdentifier, getStorageDescriptor(table), getTableParameters());
+  }
+
+  @Override
+  public Table getUpdateTableRequest(
+      InternalTable table, Table catalogTable, CatalogTableIdentifier tableIdentifier) {
+    Table copyTb = new Table(catalogTable);
+    copyTb.getSd().setCols(schemaExtractor.toColumns(tableFormat, table.getReadSchema()));
+    return copyTb;
+  }
+
+  @VisibleForTesting
+  StorageDescriptor getStorageDescriptor(InternalTable table) {
+    final StorageDescriptor storageDescriptor = new StorageDescriptor();
+    storageDescriptor.setCols(schemaExtractor.toColumns(tableFormat, table.getReadSchema()));
+    storageDescriptor.setLocation(table.getBasePath());
+    SerDeInfo serDeInfo = new SerDeInfo();
+    serDeInfo.setParameters(getSerDeParameters(table));
+    storageDescriptor.setSerdeInfo(serDeInfo);
+    return storageDescriptor;
+  }
+
+  @VisibleForTesting
+  Map<String, String> getTableParameters() {
+    Map<String, String> parameters = new HashMap<>();
+    parameters.put(PROP_EXTERNAL, "TRUE");
+    parameters.put(TABLE_TYPE_PROP, tableFormat.toUpperCase(Locale.ENGLISH));
+    parameters.put(
+        hive_metastoreConstants.META_TABLE_STORAGE, DeltaStorageHandler.class.getCanonicalName());
+    return parameters;
+  }
+
+  private Map<String, String> getSerDeParameters(InternalTable table) {
+    Map<String, String> parameters = new HashMap<>();
+    parameters.put(PROP_SERIALIZATION_FORMAT, "1");
+    parameters.put(PROP_PATH, table.getBasePath());
+    return parameters;
+  }
+}

xtable-hive-metastore/src/test/java/org/apache/xtable/hms/HMSCatalogSyncClientTestBase.java

@@ -18,21 +18,28 @@
 
 package org.apache.xtable.hms;
 
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hive.metastore.IMetaStoreClient;
 import org.apache.hadoop.hive.metastore.api.Database;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
 import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
 import org.apache.hadoop.hive.metastore.api.Table;
 import org.mockito.Mock;
 
 import org.apache.xtable.conversion.ExternalCatalogConfig;
 import org.apache.xtable.model.InternalTable;
 import org.apache.xtable.model.catalog.ThreePartHierarchicalTableIdentifier;
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalPartitionField;
 import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.InternalType;
+import org.apache.xtable.model.schema.PartitionTransformType;
 import org.apache.xtable.model.storage.CatalogType;
 import org.apache.xtable.model.storage.TableFormat;
 
@@ -57,17 +64,80 @@ public class HMSCatalogSyncClientTestBase {
 
   protected static final String ICEBERG_METADATA_FILE_LOCATION = "base-path/metadata";
   protected static final String ICEBERG_METADATA_FILE_LOCATION_V2 = "base-path/v2-metadata";
+  protected static final InternalPartitionField PARTITION_FIELD =
+      InternalPartitionField.builder()
+          .sourceField(
+              InternalField.builder()
+                  .name("partitionField")
+                  .schema(
+                      InternalSchema.builder().name("string").dataType(InternalType.STRING).build())
+                  .build())
+          .transformType(PartitionTransformType.VALUE)
+          .build();
+  protected static final InternalSchema INTERNAL_SCHEMA =
+      InternalSchema.builder()
+          .dataType(InternalType.RECORD)
+          .fields(
+              Arrays.asList(
+                  getInternalField("intField", "int", InternalType.INT),
+                  getInternalField("stringField", "string", InternalType.STRING),
+                  getInternalField("partitionField", "string", InternalType.STRING)))
+          .build();
+  protected static final InternalSchema UPDATED_INTERNAL_SCHEMA =
+      InternalSchema.builder()
+          .dataType(InternalType.RECORD)
+          .fields(
+              Arrays.asList(
+                  getInternalField("intField", "int", InternalType.INT),
+                  getInternalField("stringField", "string", InternalType.STRING),
+                  getInternalField("partitionField", "string", InternalType.STRING),
+                  getInternalField("booleanField", "boolean", InternalType.BOOLEAN)))
+          .build();
+  protected static final List<FieldSchema> FIELD_SCHEMA =
+      Arrays.asList(
+          getFieldSchema("intField", "int"),
+          getFieldSchema("stringField", "string"),
+          getFieldSchema("partitionField", "string"));
+  protected static final List<FieldSchema> UPDATED_FIELD_SCHEMA =
+      Arrays.asList(
+          getFieldSchema("intField", "int"),
+          getFieldSchema("stringField", "string"),
+          getFieldSchema("partitionField", "string"),
+          getFieldSchema("booleanField", "boolean"));
   protected static final InternalTable TEST_ICEBERG_INTERNAL_TABLE =
       InternalTable.builder()
           .basePath(TEST_BASE_PATH)
           .tableFormat(TableFormat.ICEBERG)
-          .readSchema(InternalSchema.builder().fields(Collections.emptyList()).build())
+          .readSchema(INTERNAL_SCHEMA)
+          .partitioningFields(Collections.singletonList(PARTITION_FIELD))
+          .build();
+  protected static final InternalTable TEST_UPDATED_ICEBERG_INTERNAL_TABLE =
+      InternalTable.builder()
+          .basePath(TEST_BASE_PATH)
+          .tableFormat(TableFormat.ICEBERG)
+          .readSchema(UPDATED_INTERNAL_SCHEMA)
+          .partitioningFields(Collections.singletonList(PARTITION_FIELD))
+          .build();
+  protected static final InternalTable TEST_DELTA_INTERNAL_TABLE =
+      InternalTable.builder()
+          .basePath(TEST_BASE_PATH)
+          .tableFormat(TableFormat.DELTA)
+          .readSchema(INTERNAL_SCHEMA)
+          .partitioningFields(Collections.singletonList(PARTITION_FIELD))
+          .build();
+  protected static final InternalTable TEST_UPDATED_DELTA_INTERNAL_TABLE =
+      InternalTable.builder()
+          .basePath(TEST_BASE_PATH)
+          .tableFormat(TableFormat.DELTA)
+          .readSchema(UPDATED_INTERNAL_SCHEMA)
+          .partitioningFields(Collections.singletonList(PARTITION_FIELD))
           .build();
   protected static final InternalTable TEST_HUDI_INTERNAL_TABLE =
       InternalTable.builder()
           .basePath(TEST_BASE_PATH)
           .tableFormat(TableFormat.HUDI)
-          .readSchema(InternalSchema.builder().fields(Collections.emptyList()).build())
+          .readSchema(INTERNAL_SCHEMA)
+          .partitioningFields(Collections.singletonList(PARTITION_FIELD))
           .build();
   protected static final ThreePartHierarchicalTableIdentifier TEST_CATALOG_TABLE_IDENTIFIER =
       new ThreePartHierarchicalTableIdentifier(TEST_HMS_DATABASE, TEST_HMS_TABLE);
@@ -95,4 +165,16 @@ protected Database newDatabase(String dbName) {
     return new Database(
         dbName, "Created by " + HMSCatalogSyncClient.class.getName(), null, Collections.emptyMap());
   }
+
+  protected static FieldSchema getFieldSchema(String name, String type) {
+    return new FieldSchema(name, type, null);
+  }
+
+  protected static InternalField getInternalField(
+      String fieldName, String schemaName, InternalType dataType) {
+    return InternalField.builder()
+        .name(fieldName)
+        .schema(InternalSchema.builder().name(schemaName).dataType(dataType).build())
+        .build();
+  }
 }

xtable-hive-metastore/src/test/java/org/apache/xtable/hms/TestHMSCatalogTableBuilderFactory.java

@@ -18,6 +18,7 @@
 
 package org.apache.xtable.hms;
 
+import static org.apache.xtable.hms.HMSCatalogSyncClientTestBase.FIELD_SCHEMA;
 import static org.apache.xtable.hms.HMSCatalogSyncClientTestBase.TEST_CATALOG_TABLE_IDENTIFIER;
 import static org.apache.xtable.hms.HMSCatalogSyncClientTestBase.TEST_HMS_DATABASE;
 import static org.apache.xtable.hms.HMSCatalogSyncClientTestBase.TEST_HMS_TABLE;
@@ -48,15 +49,15 @@ void testNewHmsTable() {
       expected.setTableName(TEST_HMS_TABLE);
       expected.setOwner(UserGroupInformation.getCurrentUser().getShortUserName());
       expected.setCreateTime((int) createdTime.getEpochSecond());
-      expected.setSd(getTestHmsTableStorageDescriptor());
+      expected.setSd(getTestHmsTableStorageDescriptor(FIELD_SCHEMA));
       expected.setTableType("EXTERNAL_TABLE");
       expected.setParameters(getTestHmsTableParameters());
 
       assertEquals(
           expected,
           HMSCatalogTableBuilderFactory.newHmsTable(
               TEST_CATALOG_TABLE_IDENTIFIER,
-              getTestHmsTableStorageDescriptor(),
+              getTestHmsTableStorageDescriptor(FIELD_SCHEMA),
               getTestHmsTableParameters()));
     }
   }