[HUDI-3985] Refactor DLASyncTool to support read hoodie table as spark datasource table

hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DataSourceOptions.scala

@@ -26,11 +26,11 @@ import org.apache.hudi.common.table.HoodieTableConfig
 import org.apache.hudi.common.util.Option
 import org.apache.hudi.common.util.ValidationUtils.checkState
 import org.apache.hudi.config.{HoodieClusteringConfig, HoodieWriteConfig}
-import org.apache.hudi.hive.util.ConfigUtils
 import org.apache.hudi.hive.{HiveSyncConfig, HiveSyncTool}
 import org.apache.hudi.keygen.constant.KeyGeneratorOptions
 import org.apache.hudi.keygen.{ComplexKeyGenerator, CustomKeyGenerator, NonpartitionedKeyGenerator, SimpleKeyGenerator}
 import org.apache.hudi.sync.common.HoodieSyncConfig
+import org.apache.hudi.sync.common.util.ConfigUtils
 import org.apache.log4j.LogManager
 import org.apache.spark.sql.execution.datasources.{DataSourceUtils => SparkDataSourceUtils}
 

hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/DropHoodieTableCommand.scala

@@ -21,7 +21,7 @@ import org.apache.hadoop.fs.Path
 import org.apache.hudi.client.common.HoodieSparkEngineContext
 import org.apache.hudi.common.fs.FSUtils
 import org.apache.hudi.common.model.HoodieTableType
-import org.apache.hudi.hive.util.ConfigUtils
+import org.apache.hudi.sync.common.util.ConfigUtils
 import org.apache.spark.sql._
 import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.catalog._

hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/CreateHoodieTableAsSelectCommand.scala

@@ -21,10 +21,10 @@ import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.Path
 import org.apache.hudi.DataSourceWriteOptions
 import org.apache.hudi.hive.HiveSyncConfig
-import org.apache.hudi.hive.util.ConfigUtils
 import org.apache.hudi.sql.InsertMode
 import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, CatalogTableType, HoodieCatalogTable}
 import org.apache.spark.sql.catalyst.catalog.HoodieCatalogTable.needFilterProps
+import org.apache.hudi.sync.common.util.ConfigUtils
 import org.apache.spark.sql.catalyst.plans.QueryPlan
 import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Project}
 import org.apache.spark.sql.hudi.HoodieSqlCommonUtils

hudi-spark-datasource/hudi-spark3/src/main/scala/org/apache/spark/sql/hudi/catalog/HoodieCatalog.scala

@@ -20,8 +20,8 @@ package org.apache.spark.sql.hudi.catalog
 
 import org.apache.hadoop.fs.Path
 import org.apache.hudi.exception.HoodieException
-import org.apache.hudi.hive.util.ConfigUtils
 import org.apache.hudi.sql.InsertMode
+import org.apache.hudi.sync.common.util.ConfigUtils
 import org.apache.hudi.{DataSourceWriteOptions, SparkAdapterSupport}
 import org.apache.spark.sql.HoodieSpark3SqlUtils.convertTransforms
 import org.apache.spark.sql.catalyst.TableIdentifier

hudi-sync/hudi-dla-sync/pom.xml → hudi-sync/hudi-adb-sync/pom.xml

@@ -25,7 +25,7 @@
 
   <modelVersion>4.0.0</modelVersion>
 
-  <artifactId>hudi-dla-sync</artifactId>
+  <artifactId>hudi-adb-sync</artifactId>
   <packaging>jar</packaging>
 
   <properties>

hudi-sync/hudi-adb-sync/src/main/java/org/apache/hudi/sync/adb/AbstractAdbSyncHoodieClient.java

@@ -0,0 +1,128 @@
+/*
+ * 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.hudi.sync.adb;
+
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.util.StringUtils;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.hive.PartitionValueExtractor;
+import org.apache.hudi.hive.SchemaDifference;
+import org.apache.hudi.sync.common.AbstractSyncHoodieClient;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public abstract class AbstractAdbSyncHoodieClient extends AbstractSyncHoodieClient {
+  protected AdbSyncConfig adbSyncConfig;
+  protected PartitionValueExtractor partitionValueExtractor;
+  protected HoodieTimeline activeTimeline;
+
+  public AbstractAdbSyncHoodieClient(AdbSyncConfig syncConfig, FileSystem fs) {
+    super(syncConfig.basePath, syncConfig.assumeDatePartitioning,
+        syncConfig.useFileListingFromMetadata, false, fs);
+    this.adbSyncConfig = syncConfig;
+    final String clazz = adbSyncConfig.partitionValueExtractorClass;
+    try {
+      this.partitionValueExtractor = (PartitionValueExtractor) Class.forName(clazz).newInstance();
+    } catch (Exception e) {
+      throw new HoodieException("Fail to init PartitionValueExtractor class " + clazz, e);
+    }
+
+    activeTimeline = metaClient.getActiveTimeline().getCommitsTimeline().filterCompletedInstants();
+  }
+
+  public List<PartitionEvent> getPartitionEvents(Map<List<String>, String> tablePartitions,
+                                                 List<String> partitionStoragePartitions) {
+    Map<String, String> paths = new HashMap<>();
+
+    for (Map.Entry<List<String>, String> entry : tablePartitions.entrySet()) {
+      List<String> partitionValues = entry.getKey();
+      String fullTablePartitionPath = entry.getValue();
+      paths.put(String.join(", ", partitionValues), fullTablePartitionPath);
+    }
+    List<PartitionEvent> events = new ArrayList<>();
+    for (String storagePartition : partitionStoragePartitions) {
+      Path storagePartitionPath = FSUtils.getPartitionPath(adbSyncConfig.basePath, storagePartition);
+      String fullStoragePartitionPath = Path.getPathWithoutSchemeAndAuthority(storagePartitionPath).toUri().getPath();
+      // Check if the partition values or if hdfs path is the same
+      List<String> storagePartitionValues = partitionValueExtractor.extractPartitionValuesInPath(storagePartition);
+      if (adbSyncConfig.useHiveStylePartitioning) {
+        String partition = String.join("/", storagePartitionValues);
+        storagePartitionPath = FSUtils.getPartitionPath(adbSyncConfig.basePath, partition);
+        fullStoragePartitionPath = Path.getPathWithoutSchemeAndAuthority(storagePartitionPath).toUri().getPath();
+      }
+      if (!storagePartitionValues.isEmpty()) {
+        String storageValue = String.join(", ", storagePartitionValues);
+        if (!paths.containsKey(storageValue)) {
+          events.add(PartitionEvent.newPartitionAddEvent(storagePartition));
+        } else if (!paths.get(storageValue).equals(fullStoragePartitionPath)) {
+          events.add(PartitionEvent.newPartitionUpdateEvent(storagePartition));
+        }
+      }
+    }
+    return events;
+  }
+
+  public void close() {
+
+  }
+
+  public abstract Map<List<String>, String> scanTablePartitions(String tableName) throws Exception;
+
+  public abstract void updateTableDefinition(String tableName, SchemaDifference schemaDiff) throws Exception;
+
+  public abstract boolean databaseExists(String databaseName) throws Exception;
+
+  public abstract void createDatabase(String databaseName) throws Exception;
+
+  public abstract void dropTable(String tableName);
+
+  protected String getDatabasePath() {
+    String dbLocation = adbSyncConfig.dbLocation;
+    Path dbLocationPath;
+    if (StringUtils.isNullOrEmpty(dbLocation)) {
+      if (new Path(adbSyncConfig.basePath).isRoot()) {
+        dbLocationPath = new Path(adbSyncConfig.basePath);
+      } else {
+        dbLocationPath = new Path(adbSyncConfig.basePath).getParent();
+      }
+    } else {
+      dbLocationPath = new Path(dbLocation);
+    }
+    return generateAbsolutePathStr(dbLocationPath);
+  }
+
+  protected String generateAbsolutePathStr(Path path) {
+    String absolutePathStr = path.toString();
+    if (path.toUri().getScheme() == null) {
+      absolutePathStr = getDefaultFs() + absolutePathStr;
+    }
+    return absolutePathStr.endsWith("/") ? absolutePathStr : absolutePathStr + "/";
+  }
+
+  protected String getDefaultFs() {
+    return fs.getConf().get("fs.defaultFS");
+  }
+}