[590] Add Hudi HMS Catalog Sync Implementation

xtable-core/src/main/java/org/apache/xtable/hudi/catalog/HudiCatalogTableUtils.java → xtable-core/src/main/java/org/apache/xtable/hudi/catalog/HudiCatalogTablePropertiesExtractor.java

@@ -23,6 +23,9 @@
 import java.util.List;
 import java.util.Map;
 
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
 import org.apache.spark.sql.types.StructType;
 
 import org.apache.hudi.common.util.StringUtils;
@@ -33,15 +36,22 @@
 import org.apache.xtable.schema.SparkSchemaExtractor;
 
 /** Util class to fetch details about Hudi table */
-public class HudiCatalogTableUtils {
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class HudiCatalogTablePropertiesExtractor {
+
+  private static final HudiCatalogTablePropertiesExtractor INSTANCE =
+      new HudiCatalogTablePropertiesExtractor();
 
+  public static HudiCatalogTablePropertiesExtractor getInstance() {
+    return INSTANCE;
+  }
   /**
    * Get Spark Sql related table properties. This is used for spark datasource table.
    *
    * @param schema The schema to write to the table.
    * @return A new parameters added the spark's table properties.
    */
-  public static Map<String, String> getSparkTableProperties(
+  public Map<String, String> getSparkTableProperties(
       List<String> partitionNames,
       String sparkVersion,
       int schemaLengthThreshold,

xtable-core/src/test/java/org/apache/xtable/hudi/catalog/TestHudiCatalogTableUtils.java → xtable-core/src/test/java/org/apache/xtable/hudi/catalog/TestHudiCatalogTablePropertiesExtractor.java

@@ -32,7 +32,7 @@
 import org.apache.xtable.model.schema.InternalSchema;
 import org.apache.xtable.model.schema.InternalType;
 
-public class TestHudiCatalogTableUtils {
+public class TestHudiCatalogTablePropertiesExtractor {
 
   @Test
   void testGetSparkTableProperties() {
@@ -81,8 +81,8 @@ void testGetSparkTableProperties() {
             .build();
 
     Map<String, String> result =
-        HudiCatalogTableUtils.getSparkTableProperties(
-            partitionNames, sparkVersion, schemaLengthThreshold, schema);
+        HudiCatalogTablePropertiesExtractor.getInstance()
+            .getSparkTableProperties(partitionNames, sparkVersion, schemaLengthThreshold, schema);
 
     // Validate results
     assertEquals("hudi", result.get("spark.sql.sources.provider"));
@@ -127,8 +127,8 @@ void testGetSparkTablePropertiesEmptyPartitions() {
 
     // Call the method
     Map<String, String> result =
-        HudiCatalogTableUtils.getSparkTableProperties(
-            partitionNames, "", schemaLengthThreshold, schema);
+        HudiCatalogTablePropertiesExtractor.getInstance()
+            .getSparkTableProperties(partitionNames, "", schemaLengthThreshold, schema);
 
     assertEquals("hudi", result.get("spark.sql.sources.provider"));
     assertNull(result.get("spark.sql.create.version"));

xtable-hive-metastore/pom.xml

@@ -42,6 +42,14 @@
             <scope>provided</scope>
         </dependency>
 
+        <!-- Hudi dependencies -->
+        <dependency>
+            <groupId>org.apache.hudi</groupId>
+            <artifactId>hudi-hive-sync</artifactId>
+            <version>${hudi.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
         <!-- Iceberg dependencies  -->
         <dependency>
             <groupId>org.apache.iceberg</groupId>

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

@@ -26,6 +26,8 @@
 import lombok.RequiredArgsConstructor;
 import lombok.ToString;
 
+import org.apache.hudi.hive.MultiPartKeysValueExtractor;
+
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.DeserializationFeature;
@@ -43,6 +45,15 @@ public class HMSCatalogConfig {
   @JsonProperty("externalCatalog.hms.serverUrl")
   private final String serverUrl;
 
+  @JsonProperty("externalCatalog.hms.schema_string_length_thresh")
+  private final int schemaLengthThreshold = 4000;
+
+  @JsonProperty("externalCatalog.hms.partition_extractor_class")
+  private final String partitionExtractorClass = MultiPartKeysValueExtractor.class.getName();
+
+  @JsonProperty("externalCatalog.hms.max_partitions_per_request")
+  private final int maxPartitionsPerRequest = 1000;
+
   protected static HMSCatalogConfig of(Map<String, String> properties) {
     try {
       return OBJECT_MAPPER.readValue(

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

@@ -0,0 +1,221 @@
+/*
+ * 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;
+
+import static org.apache.xtable.catalog.CatalogUtils.toHierarchicalTableIdentifier;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.thrift.TException;
+
+import org.apache.hudi.common.util.CollectionUtils;
+import org.apache.hudi.exception.TableNotFoundException;
+
+import org.apache.xtable.catalog.CatalogPartition;
+import org.apache.xtable.catalog.CatalogPartitionSyncOperations;
+import org.apache.xtable.exception.CatalogSyncException;
+import org.apache.xtable.model.catalog.CatalogTableIdentifier;
+import org.apache.xtable.model.catalog.HierarchicalTableIdentifier;
+
+@Log4j2
+public class HMSCatalogPartitionSyncOperations implements CatalogPartitionSyncOperations {
+
+  private final IMetaStoreClient metaStoreClient;
+  private final HMSCatalogConfig catalogConfig;
+
+  public HMSCatalogPartitionSyncOperations(
+      IMetaStoreClient metaStoreClient, HMSCatalogConfig hmsCatalogConfig) {
+    this.metaStoreClient = metaStoreClient;
+    this.catalogConfig = hmsCatalogConfig;
+  }
+
+  @Override
+  public List<CatalogPartition> getAllPartitions(CatalogTableIdentifier catalogTableIdentifier) {
+    HierarchicalTableIdentifier tableIdentifier =
+        toHierarchicalTableIdentifier(catalogTableIdentifier);
+    try {
+      return metaStoreClient
+          .listPartitions(
+              tableIdentifier.getDatabaseName(), tableIdentifier.getTableName(), (short) -1)
+          .stream()
+          .map(p -> new CatalogPartition(p.getValues(), p.getSd().getLocation()))
+          .collect(Collectors.toList());
+    } catch (TException e) {
+      throw new CatalogSyncException(
+          "Failed to get all partitions for table " + tableIdentifier, e);
+    }
+  }
+
+  @Override
+  public void addPartitionsToTable(
+      CatalogTableIdentifier catalogTableIdentifier, List<CatalogPartition> partitionsToAdd) {
+    HierarchicalTableIdentifier tableIdentifier =
+        toHierarchicalTableIdentifier(catalogTableIdentifier);
+    if (partitionsToAdd.isEmpty()) {
+      log.info("No partitions to add for {}", tableIdentifier);
+      return;
+    }
+    log.info("Adding partitions {} to table {}", partitionsToAdd.size(), tableIdentifier);
+    try {
+      StorageDescriptor sd =
+          metaStoreClient
+              .getTable(tableIdentifier.getDatabaseName(), tableIdentifier.getTableName())
+              .getSd();
+
+      CollectionUtils.batches(partitionsToAdd, catalogConfig.getMaxPartitionsPerRequest())
+          .forEach(
+              batch -> {
+                List<Partition> partitionList = new ArrayList<>();
+                batch.forEach(
+                    partition -> {
+                      partitionList.add(createPartition(tableIdentifier, partition, sd));
+                    });
+                try {
+                  metaStoreClient.add_partitions(partitionList, true, false);
+                } catch (TException e) {
+                  log.error("{} add partition failed", tableIdentifier, e);
+                  throw new CatalogSyncException(tableIdentifier + " add partition failed", e);
+                }
+                log.info("Add batch partitions done: {}", partitionList.size());
+              });
+    } catch (TException e) {
+      log.error("Failed to add partitions to table {}", tableIdentifier, e);
+      throw new CatalogSyncException(tableIdentifier + " add partition failed", e);
+    }
+  }
+
+  @Override
+  public void updatePartitionsToTable(
+      CatalogTableIdentifier catalogTableIdentifier, List<CatalogPartition> changedPartitions) {
+    HierarchicalTableIdentifier tableIdentifier =
+        toHierarchicalTableIdentifier(catalogTableIdentifier);
+    try {
+      Table table =
+          metaStoreClient.getTable(
+              tableIdentifier.getDatabaseName(), tableIdentifier.getTableName());
+      StorageDescriptor tableSd = table.getSd();
+
+      List<Partition> updatedPartitions = new ArrayList<>();
+
+      changedPartitions.forEach(
+          partition -> {
+            updatedPartitions.add(createPartition(tableIdentifier, partition, tableSd));
+          });
+
+      // Update partitions
+      metaStoreClient.alter_partitions(
+          tableIdentifier.getDatabaseName(), tableIdentifier.getTableName(), updatedPartitions);
+    } catch (TException e) {
+      throw new CatalogSyncException(
+          "Failed to update partitions for the table " + tableIdentifier, e);
+    }
+  }
+
+  @Override
+  public void dropPartitions(
+      CatalogTableIdentifier catalogTableIdentifier, List<CatalogPartition> partitionsToDrop) {
+    HierarchicalTableIdentifier tableIdentifier =
+        toHierarchicalTableIdentifier(catalogTableIdentifier);
+    try {
+      for (CatalogPartition partition : partitionsToDrop) {
+        metaStoreClient.dropPartition(
+            tableIdentifier.getDatabaseName(),
+            tableIdentifier.getTableName(),
+            partition.getValues(),
+            false);
+      }
+    } catch (TException e) {
+      throw new CatalogSyncException("Failed to drop partitions for table " + tableIdentifier, e);
+    }
+  }
+
+  @Override
+  public Map<String, String> getTableProperties(
+      CatalogTableIdentifier catalogTableIdentifier, List<String> keysToRetrieve) {
+    HierarchicalTableIdentifier tableIdentifier =
+        toHierarchicalTableIdentifier(catalogTableIdentifier);
+    try {
+      Table table =
+          metaStoreClient.getTable(
+              tableIdentifier.getDatabaseName(), tableIdentifier.getTableName());
+      Map<String, String> tableParameters = table.getParameters();
+
+      return keysToRetrieve.stream()
+          .filter(tableParameters::containsKey)
+          .collect(Collectors.toMap(key -> key, tableParameters::get));
+    } catch (TableNotFoundException | TException e) {
+      throw new CatalogSyncException(
+          "failed to fetch last time synced properties for table" + tableIdentifier, e);
+    }
+  }
+
+  @Override
+  public void updateTableProperties(
+      CatalogTableIdentifier catalogTableIdentifier, Map<String, String> propertiesToUpdate) {
+    HierarchicalTableIdentifier tableIdentifier =
+        toHierarchicalTableIdentifier(catalogTableIdentifier);
+    try {
+      if (propertiesToUpdate == null || propertiesToUpdate.isEmpty()) {
+        return;
+      }
+
+      Table table =
+          metaStoreClient.getTable(
+              tableIdentifier.getDatabaseName(), tableIdentifier.getTableName());
+      Map<String, String> tableParameters = table.getParameters();
+      tableParameters.putAll(propertiesToUpdate);
+      table.setParameters(tableParameters);
+      metaStoreClient.alter_table(
+          tableIdentifier.getDatabaseName(), tableIdentifier.getTableName(), table);
+    } catch (TableNotFoundException | TException e) {
+      throw new CatalogSyncException(
+          "failed to update last time synced properties for table" + tableIdentifier, e);
+    }
+  }
+
+  private Partition createPartition(
+      HierarchicalTableIdentifier tableIdentifier,
+      CatalogPartition partition,
+      StorageDescriptor sd) {
+    StorageDescriptor partitionSd = new StorageDescriptor();
+    partitionSd.setCols(sd.getCols());
+    partitionSd.setInputFormat(sd.getInputFormat());
+    partitionSd.setOutputFormat(sd.getOutputFormat());
+    partitionSd.setSerdeInfo(sd.getSerdeInfo());
+    partitionSd.setLocation(partition.getStorageLocation());
+
+    return new Partition(
+        partition.getValues(),
+        tableIdentifier.getDatabaseName(),
+        tableIdentifier.getTableName(),
+        0,
+        0,
+        partitionSd,
+        null);
+  }
+}