[834] Add Hudi table version 9 (Hudi 1.x) support to the Hudi target

pom.xml

@@ -70,7 +70,7 @@
         <lombok.version>1.18.46</lombok.version>
         <lombok-maven-plugin.version>1.18.20.0</lombok-maven-plugin.version>
         <hadoop.version>3.4.1</hadoop.version>
-        <hudi.version>1.2.0</hudi.version>
+        <hudi.version>1.3.0-SNAPSHOT</hudi.version>
         <aws.version>2.29.40</aws.version>
         <hive.version>3.1.3</hive.version>
         <maven-source-plugin.version>3.3.1</maven-source-plugin.version>

xtable-core/src/main/java/org/apache/hudi/stats/XTableValueMetadata.java → xtable-core/src/main/java/org/apache/hudi/metadata/stats/XTableValueMetadata.java

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.hudi.stats;
+package org.apache.hudi.metadata.stats;
 
 import static org.apache.xtable.model.schema.InternalSchema.MetadataKey.TIMESTAMP_PRECISION;
 import static org.apache.xtable.model.schema.InternalSchema.MetadataValue.MICROS;

xtable-core/src/main/java/org/apache/xtable/hudi/BaseFileUpdatesExtractor.java

@@ -57,9 +57,9 @@
 import org.apache.hudi.hadoop.fs.CachingPath;
 import org.apache.hudi.metadata.HoodieIndexVersion;
 import org.apache.hudi.metadata.HoodieTableMetadata;
-import org.apache.hudi.stats.HoodieColumnRangeMetadata;
-import org.apache.hudi.stats.ValueMetadata;
-import org.apache.hudi.stats.XTableValueMetadata;
+import org.apache.hudi.metadata.stats.HoodieColumnRangeMetadata;
+import org.apache.hudi.metadata.stats.ValueMetadata;
+import org.apache.hudi.metadata.stats.XTableValueMetadata;
 
 import org.apache.xtable.collectors.CustomCollectors;
 import org.apache.xtable.exception.ReadException;
@@ -75,6 +75,13 @@ public class BaseFileUpdatesExtractor {
   private static final Pattern HUDI_BASE_FILE_PATTERN =
       Pattern.compile(
           "^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}-[0-9]_[0-9a-fA-F-]+_[0-9]+\\.");
+  // External bucketed sources (e.g. Paimon) lay files out as {@code <partition>/bucket-N/<file>}.
+  // Hudi treats the trailing {@code bucket-N} directory as a file-group prefix within the partition
+  // rather than as part of the partition path (see Hudi PR #17788). Detecting such a directory lets
+  // us register the file under its true partition with the prefix encoded into the external-file
+  // marker, so an unpartitioned external table is not mistakenly read as partitioned by "bucket-N".
+  private static final Pattern EXTERNAL_FILE_GROUP_PREFIX_PATTERN =
+      Pattern.compile("bucket-[0-9]+");
   private final HoodieEngineContext engineContext;
   private final Path tableBasePath;
 
@@ -234,7 +241,7 @@ ReplaceMetadata convertDiff(
             .map(file -> new CachingPath(file.getPhysicalPath()))
             .collect(
                 Collectors.groupingBy(
-                    path -> HudiPathUtils.getPartitionPath(tableBasePath, path),
+                    path -> truePartitionPath(tableBasePath, path),
                     Collectors.mapping(this::getFileId, Collectors.toList())));
     // For all added files, group by partition and extract the file id
     List<WriteStatus> writeStatuses =
@@ -250,7 +257,42 @@ private String getFileId(Path filePath) {
     if (isFileCreatedByHudiWriter(fileName)) {
       return FSUtils.getFileId(fileName);
     }
-    return fileName;
+    // External bucketed files keep their file-group prefix as part of the fileId so the prefix can
+    // be recovered when Hudi resolves the physical path of the externally created file.
+    return externalFileGroupPrefix(filePath)
+        .map(prefix -> prefix + "/" + fileName)
+        .orElse(fileName);
+  }
+
+  /**
+   * Returns the external file-group prefix (e.g. Paimon's {@code bucket-N}) when the file's
+   * immediate parent directory denotes a file group within the partition rather than a partition
+   * segment, otherwise empty.
+   */
+  private Optional<String> externalFileGroupPrefix(Path filePath) {
+    Path parent = filePath.getParent();
+    if (parent == null) {
+      return Optional.empty();
+    }
+    String parentName = parent.getName();
+    return EXTERNAL_FILE_GROUP_PREFIX_PATTERN.matcher(parentName).matches()
+        ? Optional.of(parentName)
+        : Optional.empty();
+  }
+
+  /**
+   * Resolves the true Hudi partition path for a file, stripping any trailing external file-group
+   * prefix directory (e.g. {@code bucket-N}) so it is not treated as part of the partition.
+   */
+  private String truePartitionPath(Path tableBasePath, Path filePath) {
+    String partitionPath = HudiPathUtils.getPartitionPath(tableBasePath, filePath);
+    Optional<String> prefix = externalFileGroupPrefix(filePath);
+    if (!prefix.isPresent()) {
+      return partitionPath;
+    }
+    return partitionPath.equals(prefix.get())
+        ? ""
+        : partitionPath.substring(0, partitionPath.length() - prefix.get().length() - 1);
   }
 
   /**
@@ -273,17 +315,31 @@ private WriteStatus toWriteStatus(
     WriteStatus writeStatus = new WriteStatus();
     Path path = new CachingPath(file.getPhysicalPath());
     String partitionPath =
-        partitionPathOptional.orElseGet(() -> HudiPathUtils.getPartitionPath(tableBasePath, path));
+        partitionPathOptional.orElseGet(() -> truePartitionPath(tableBasePath, path));
     String fileId = getFileId(path);
     String filePath =
         path.toUri().getPath().substring(tableBasePath.toUri().getPath().length() + 1);
     String fileName = path.getName();
+    Optional<String> fileGroupPrefix = externalFileGroupPrefix(path);
+    // For external bucketed files encode the file-group prefix in the marker and keep the file name
+    // (not the bucket-relative path) as the marked name; otherwise fall back to the plain marker on
+    // the full relative path. In both cases the directory portion is preserved as-is.
+    String markedPath =
+        fileGroupPrefix
+            .map(
+                prefix ->
+                    filePath.substring(0, filePath.length() - fileName.length())
+                        + ExternalFilePathUtil.appendCommitTimeAndExternalFileMarker(
+                            fileName, commitTime, prefix))
+            .orElseGet(
+                () ->
+                    ExternalFilePathUtil.appendCommitTimeAndExternalFileMarker(
+                        filePath, commitTime));
     writeStatus.setFileId(fileId);
     writeStatus.setPartitionPath(partitionPath);
     HoodieDeltaWriteStat writeStat = new HoodieDeltaWriteStat();
     writeStat.setFileId(fileId);
-    writeStat.setPath(
-        ExternalFilePathUtil.appendCommitTimeAndExternalFileMarker(filePath, commitTime));
+    writeStat.setPath(markedPath);
     writeStat.setPartitionPath(partitionPath);
     writeStat.setNumWrites(file.getRecordCount());
     writeStat.setTotalWriteBytes(file.getFileSizeBytes());
@@ -338,7 +394,6 @@ private ReplaceMetadata combine(ReplaceMetadata other) {
   }
 
   private String getPartitionPath(Path tableBasePath, List<InternalDataFile> files) {
-    return HudiPathUtils.getPartitionPath(
-        tableBasePath, new CachingPath(files.get(0).getPhysicalPath()));
+    return truePartitionPath(tableBasePath, new CachingPath(files.get(0).getPhysicalPath()));
   }
 }

xtable-core/src/main/java/org/apache/xtable/hudi/HudiConversionSource.java

@@ -23,10 +23,13 @@
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.Comparator;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import lombok.Builder;
 import lombok.NonNull;
@@ -35,6 +38,7 @@
 
 import org.apache.hudi.avro.model.HoodieCleanMetadata;
 import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.HoodieTableVersion;
 import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
 import org.apache.hudi.common.table.timeline.HoodieInstant;
 import org.apache.hudi.common.table.timeline.HoodieTimeline;
@@ -83,30 +87,26 @@ public InternalTable getTable(HoodieInstant commit) {
   public InternalTable getCurrentTable() {
     HoodieActiveTimeline activeTimeline = metaClient.getActiveTimeline();
     HoodieTimeline completedTimeline = activeTimeline.filterCompletedInstants();
-    // get latest commit
-    HoodieInstant latestCommit =
-        completedTimeline
-            .lastInstant()
-            .orElseThrow(
-                () -> new ReadException("Unable to read latest commit from Hudi source table"));
-    return getTable(latestCommit);
+    return getTable(getLatestCompletedInstant(completedTimeline));
   }
 
   @Override
   public InternalSnapshot getCurrentSnapshot() {
     HoodieActiveTimeline activeTimeline = metaClient.getActiveTimeline();
     HoodieTimeline completedTimeline = activeTimeline.filterCompletedInstants();
     // get latest commit
-    HoodieInstant latestCommit =
-        completedTimeline
-            .lastInstant()
-            .orElseThrow(
-                () -> new ReadException("Unable to read latest commit from Hudi source table"));
+    HoodieInstant latestCommit = getLatestCompletedInstant(completedTimeline);
+    // On table version 9 (timeline layout V2) a commit becomes visible at its completion time, so
+    // an instant with an earlier requested time may complete after the latest commit. Capture all
+    // currently inflight/requested instants as pending so none are missed; on version 6 keep the
+    // historical requested-time window.
     List<HoodieInstant> pendingInstants =
-        activeTimeline
-            .filterInflightsAndRequested()
-            .findInstantsBefore(latestCommit.requestedTime())
-            .getInstants();
+        usesCompletionTimeOrdering()
+            ? activeTimeline.filterInflightsAndRequested().getInstants()
+            : activeTimeline
+                .filterInflightsAndRequested()
+                .findInstantsBefore(latestCommit.requestedTime())
+                .getInstants();
     InternalTable table = getTable(latestCommit);
     return InternalSnapshot.builder()
         .table(table)
@@ -124,10 +124,17 @@ public InternalSnapshot getCurrentSnapshot() {
   @Override
   public TableChange getTableChangeForCommit(HoodieInstant hoodieInstantForDiff) {
     HoodieActiveTimeline activeTimeline = metaClient.getActiveTimeline();
+    // The set of commits visible as-of the diff commit is ordered by completion time on table
+    // version 9 (timeline layout V2) and by requested time on version 6.
     HoodieTimeline visibleTimeline =
-        activeTimeline
-            .filterCompletedInstants()
-            .findInstantsBeforeOrEquals(hoodieInstantForDiff.requestedTime());
+        usesCompletionTimeOrdering()
+            ? activeTimeline
+                .filterCompletedInstants()
+                .findInstantsModifiedBeforeOrEqualsByCompletionTime(
+                    hoodieInstantForDiff.getCompletionTime())
+            : activeTimeline
+                .filterCompletedInstants()
+                .findInstantsBeforeOrEquals(hoodieInstantForDiff.requestedTime());
     InternalTable table = getTable(hoodieInstantForDiff);
     return TableChange.builder()
         .tableAsOfChange(table)
@@ -148,9 +155,13 @@ public CommitsBacklog<HoodieInstant> getCommitsBacklog(
     CommitsPair lastPendingHoodieInstantsCommitsPair =
         getCompletedAndPendingCommitsForInstants(lastPendingInstants);
     List<HoodieInstant> commitsToProcessNext =
-        mergeAndDedupLists(
-            lastPendingHoodieInstantsCommitsPair.getCompletedCommits(),
-            commitsPair.getCompletedCommits());
+        usesCompletionTimeOrdering()
+            ? orderByCompletionTimeAndDedup(
+                lastPendingHoodieInstantsCommitsPair.getCompletedCommits(),
+                commitsPair.getCompletedCommits())
+            : mergeAndDedupLists(
+                lastPendingHoodieInstantsCommitsPair.getCompletedCommits(),
+                commitsPair.getCompletedCommits());
     List<Instant> pendingInstantsToProcessNext =
         mergeAndDedupLists(
             lastPendingHoodieInstantsCommitsPair.getPendingCommits(),
@@ -237,10 +248,86 @@ private HoodieTimeline getCompletedCommits() {
     return metaClient.getActiveTimeline().filterCompletedInstants();
   }
 
+  /**
+   * Table version 8+ (Hudi 1.x, timeline layout V2) makes a commit visible at its completion time
+   * rather than its requested (instant) time, so incremental selection and ordering must be based
+   * on completion time. Table version 6 keeps the legacy requested-time ordering.
+   */
+  private boolean usesCompletionTimeOrdering() {
+    return metaClient
+        .getTableConfig()
+        .getTableVersion()
+        .greaterThanOrEquals(HoodieTableVersion.EIGHT);
+  }
+
+  private HoodieInstant getLatestCompletedInstant(HoodieTimeline completedTimeline) {
+    Option<HoodieInstant> latestCommit =
+        usesCompletionTimeOrdering()
+            ? Option.fromJavaOptional(
+                completedTimeline
+                    .getInstantsOrderedByCompletionTime()
+                    .reduce((first, second) -> second))
+            : completedTimeline.lastInstant();
+    return latestCommit.orElseThrow(
+        () -> new ReadException("Unable to read latest commit from Hudi source table"));
+  }
+
+  /**
+   * Selects the commits that completed after the last synced commit's completion time, ordered by
+   * completion time. Unlike the requested-time path this also surfaces commits whose requested time
+   * is older than the last synced commit but whose completion is newer (out-of-order completion).
+   */
+  private CommitsPair getCompletedAndPendingCommitsAfterCompletionTime(
+      HoodieInstant commitInstant) {
+    List<HoodieInstant> modifiedAfter =
+        metaClient
+            .getActiveTimeline()
+            .findInstantsModifiedAfterByCompletionTime(commitInstant.getCompletionTime())
+            .getInstants();
+    List<HoodieInstant> completedInstants =
+        modifiedAfter.stream()
+            .filter(HoodieInstant::isCompleted)
+            .sorted(Comparator.comparing(HoodieInstant::getCompletionTime))
+            .collect(Collectors.toList());
+    List<Instant> pendingInstants =
+        modifiedAfter.stream()
+            .filter(hoodieInstant -> hoodieInstant.isInflight() || hoodieInstant.isRequested())
+            .map(
+                hoodieInstant ->
+                    HudiInstantUtils.parseFromInstantTime(hoodieInstant.requestedTime()))
+            .collect(Collectors.toList());
+    return CommitsPair.builder()
+        .completedCommits(completedInstants)
+        .pendingCommits(pendingInstants)
+        .build();
+  }
+
+  /**
+   * Merges two completed-commit lists, dedupes by requested time and action, and orders by
+   * completion time. The action is part of the dedup key because distinct actions can legally share
+   * a requested time: a savepoint instant reuses the requested time of the commit it pins, and
+   * keying on requested time alone would drop it from the backlog.
+   */
+  private List<HoodieInstant> orderByCompletionTimeAndDedup(
+      List<HoodieInstant> list1, List<HoodieInstant> list2) {
+    Map<String, HoodieInstant> dedupedByRequestedTimeAndAction = new LinkedHashMap<>();
+    Stream.concat(list1.stream(), list2.stream())
+        .forEach(
+            hoodieInstant ->
+                dedupedByRequestedTimeAndAction.putIfAbsent(
+                    hoodieInstant.requestedTime() + "_" + hoodieInstant.getAction(),
+                    hoodieInstant));
+    return dedupedByRequestedTimeAndAction.values().stream()
+        .sorted(Comparator.comparing(HoodieInstant::getCompletionTime))
+        .collect(Collectors.toList());
+  }
+
   private CommitsPair getCompletedAndPendingCommitsAfterInstant(HoodieInstant commitInstant) {
+    if (usesCompletionTimeOrdering()) {
+      return getCompletedAndPendingCommitsAfterCompletionTime(commitInstant);
+    }
     // Table version 6 uses the old timeline view, so instants are selected and ordered by their
-    // requested (instant) time. Completion-time based handling will be added with table version 9
-    // support in a follow-up PR.
+    // requested (instant) time.
     List<HoodieInstant> allInstants =
         metaClient
             .getActiveTimeline()