[722] Implement iceberg versions for TableFormat and HoodieTableMetadata

xtable-api/src/main/java/org/apache/xtable/model/InternalTable.java

@@ -52,4 +52,6 @@ public class InternalTable {
   Instant latestCommitTime;
   // Path to latest metadata
   String latestMetadataPath;
+  // latest operation on the table.
+  String latestTableOperationId;
 }

xtable-api/src/main/java/org/apache/xtable/model/metadata/TableSyncMetadata.java

@@ -56,6 +56,7 @@ public class TableSyncMetadata {
   int version;
   String sourceTableFormat;
   String sourceIdentifier;
+  String latestTableOperationId;
 
   /**
    * @deprecated Use {@link #of(Instant, List, String, String)} instead. This method exists for
@@ -64,20 +65,35 @@ public class TableSyncMetadata {
   @Deprecated
   public static TableSyncMetadata of(
       Instant lastInstantSynced, List<Instant> instantsToConsiderForNextSync) {
-    return TableSyncMetadata.of(lastInstantSynced, instantsToConsiderForNextSync, null, null);
+    return TableSyncMetadata.of(lastInstantSynced, instantsToConsiderForNextSync, null, null, null);
   }
 
   public static TableSyncMetadata of(
       Instant lastInstantSynced,
       List<Instant> instantsToConsiderForNextSync,
       String sourceTableFormat,
       String sourceIdentifier) {
+    return TableSyncMetadata.of(
+        lastInstantSynced,
+        instantsToConsiderForNextSync,
+        sourceTableFormat,
+        sourceIdentifier,
+        null);
+  }
+
+  public static TableSyncMetadata of(
+      Instant lastInstantSynced,
+      List<Instant> instantsToConsiderForNextSync,
+      String sourceTableFormat,
+      String sourceIdentifier,
+      String latestTableOperationId) {
     return new TableSyncMetadata(
         lastInstantSynced,
         instantsToConsiderForNextSync,
         CURRENT_VERSION,
         sourceTableFormat,
-        sourceIdentifier);
+        sourceIdentifier,
+        latestTableOperationId);
   }
 
   public String toJson() {

xtable-api/src/main/java/org/apache/xtable/spi/sync/TableFormatSync.java

@@ -168,7 +168,8 @@ private SyncResult getSyncResult(
             tableState.getLatestCommitTime(),
             pendingCommits,
             tableState.getTableFormat(),
-            sourceIdentifier);
+            sourceIdentifier,
+            tableState.getLatestTableOperationId());
     conversionTarget.syncMetadata(latestState);
     // sync schema updates
     conversionTarget.syncSchema(tableState.getReadSchema());
@@ -178,6 +179,11 @@ private SyncResult getSyncResult(
     fileSyncMethod.sync(conversionTarget);
     conversionTarget.completeSync();
 
+    log.info(
+        "Took {} sec in mode {} to sync table change for {}",
+        Duration.between(startTime, Instant.now()).getSeconds(),
+        mode,
+        conversionTarget.getTableFormat());
     return SyncResult.builder()
         .mode(mode)
         .tableFormatSyncStatus(SyncResult.SyncStatus.SUCCESS)

xtable-core/src/main/java/org/apache/xtable/conversion/ConversionTargetFactory.java

@@ -18,20 +18,23 @@
 
 package org.apache.xtable.conversion;
 
+import java.util.Iterator;
 import java.util.Properties;
+import java.util.ServiceConfigurationError;
 import java.util.ServiceLoader;
 
 import lombok.AccessLevel;
 import lombok.NoArgsConstructor;
+import lombok.extern.log4j.Log4j2;
 
 import org.apache.hadoop.conf.Configuration;
 
 import org.apache.xtable.delta.DeltaConversionTargetConfig;
 import org.apache.xtable.exception.NotSupportedException;
-import org.apache.xtable.kernel.DeltaKernelConversionTarget;
 import org.apache.xtable.model.storage.TableFormat;
 import org.apache.xtable.spi.sync.ConversionTarget;
 
+@Log4j2
 @NoArgsConstructor(access = AccessLevel.PRIVATE)
 public class ConversionTargetFactory {
   private static final ConversionTargetFactory INSTANCE = new ConversionTargetFactory();
@@ -87,7 +90,31 @@ public ConversionTarget createConversionTargetForName(
         TableFormat.DELTA.equalsIgnoreCase(tableFormatName)
             && DeltaConversionTargetConfig.fromProperties(properties).isUseKernel();
     ServiceLoader<ConversionTarget> loader = ServiceLoader.load(ConversionTarget.class);
-    for (ConversionTarget target : loader) {
+    Iterator<ConversionTarget> iterator = loader.iterator();
+    while (true) {
+      ConversionTarget target;
+      try {
+        // hasNext() also resolves provider classes lazily, so it can throw
+        // ServiceConfigurationError too - it must be inside the guard alongside next().
+        if (!iterator.hasNext()) {
+          break;
+        }
+        target = iterator.next();
+      } catch (ServiceConfigurationError | LinkageError error) {
+        // A registered target whose engine library is not on the classpath (e.g. Delta when only
+        // Hudi/Iceberg are provided). Skip it so a subset of engines can still be used; a missing
+        // engine for the requested format surfaces below as NotSupportedException. The offending
+        // provider is consumed before the error is thrown, so the next hasNext() advances past it.
+        log.warn(
+            "Skipping a registered ConversionTarget whose engine library is not on the classpath "
+                + "({}: {}); provide the missing engine if you need this target format. This is "
+                + "expected when an engine is intentionally absent, but indicates a linkage problem "
+                + "if the engine is present.",
+            error.getClass().getName(),
+            error.getMessage(),
+            error);
+        continue;
+      }
       if (target.getTableFormat().equalsIgnoreCase(tableFormatName)
           && isDeltaKernelTarget(target) == useKernel) {
         return target;
@@ -96,7 +123,10 @@ && isDeltaKernelTarget(target) == useKernel) {
     throw new NotSupportedException("Target format is not yet supported: " + tableFormatName);
   }
 
+  private static final String DELTA_KERNEL_TARGET_CLASS =
+      "org.apache.xtable.kernel.DeltaKernelConversionTarget";
+
   private static boolean isDeltaKernelTarget(ConversionTarget target) {
-    return target instanceof DeltaKernelConversionTarget;
+    return DELTA_KERNEL_TARGET_CLASS.equals(target.getClass().getName());
   }
 }

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

@@ -25,7 +25,9 @@
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -57,6 +59,8 @@
 import org.apache.hudi.common.table.view.TableFileSystemView;
 import org.apache.hudi.hadoop.fs.HadoopFSUtils;
 import org.apache.hudi.metadata.HoodieTableMetadata;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.StoragePathInfo;
 
 import org.apache.xtable.collectors.CustomCollectors;
 import org.apache.xtable.exception.NotSupportedException;
@@ -112,6 +116,21 @@ public HudiDataFileExtractor(
     this.fileStatsExtractor = hudiFileStatsExtractor;
   }
 
+  public HudiDataFileExtractor(
+      HoodieTableMetaClient metaClient,
+      PathBasedPartitionValuesExtractor hudiPartitionValuesExtractor,
+      HudiFileStatsExtractor hudiFileStatsExtractor,
+      FileSystemViewManager fileSystemViewManager) {
+    this.engineContext = new HoodieLocalEngineContext(metaClient.getStorageConf());
+    this.metadataConfig = HoodieMetadataConfig.newBuilder().enable(false).build();
+    this.basePath = HadoopFSUtils.convertToHadoopPath(metaClient.getBasePath());
+    this.tableMetadata = null;
+    this.fileSystemViewManager = fileSystemViewManager;
+    this.metaClient = metaClient;
+    this.partitionValuesExtractor = hudiPartitionValuesExtractor;
+    this.fileStatsExtractor = hudiFileStatsExtractor;
+  }
+
   public List<PartitionFileGroup> getFilesCurrentState(InternalTable table) {
     try {
       List<String> allPartitionPaths =
@@ -145,6 +164,110 @@ public InternalFilesDiff getDiffForCommit(
     return InternalFilesDiff.builder().filesAdded(filesAdded).filesRemoved(filesRemoved).build();
   }
 
+  public InternalFilesDiff getDiffForCommit(
+      InternalTable table, HoodieCommitMetadata commitMetadata, HoodieInstant commit) {
+    SyncableFileSystemView fsView = fileSystemViewManager.getFileSystemView(metaClient);
+    List<InternalDataFile> filesAddedWithoutStats = new ArrayList<>();
+    List<InternalDataFile> filesToRemove = new ArrayList<>();
+    Map<String, StoragePathInfo> fullPathInfo =
+        commitMetadata.getFullPathToInfo(metaClient.getStorage(), basePath.toString());
+    commitMetadata
+        .getPartitionToWriteStats()
+        .forEach(
+            (partitionPath, writeStats) -> {
+              List<PartitionValue> partitionValues =
+                  partitionValuesExtractor.extractPartitionValues(
+                      table.getPartitioningFields(), partitionPath);
+              Map<String, HoodieBaseFile> currentBaseFilesInPartition =
+                  fsView
+                      .getLatestBaseFiles(partitionPath)
+                      .collect(Collectors.toMap(HoodieBaseFile::getFileId, Function.identity()));
+              for (HoodieWriteStat writeStat : writeStats) {
+                if (FSUtils.isLogFile(new StoragePath(writeStat.getPath()))) {
+                  continue;
+                }
+                StoragePath baseFileFullPath =
+                    FSUtils.constructAbsolutePath(metaClient.getBasePath(), writeStat.getPath());
+                if (FSUtils.getCommitTimeWithFullPath(baseFileFullPath.toString())
+                    .equals(commit.requestedTime())) {
+                  // getFullPathToInfo keys the map by the absolute path, not the file name
+                  StoragePathInfo pathInfo = fullPathInfo.get(baseFileFullPath.toString());
+                  if (pathInfo == null) {
+                    throw new ReadException(
+                        "Commit metadata has no file info for base file " + baseFileFullPath);
+                  }
+                  filesAddedWithoutStats.add(
+                      buildFileWithoutStats(partitionValues, new HoodieBaseFile(pathInfo)));
+                }
+                if (currentBaseFilesInPartition.containsKey(writeStat.getFileId())) {
+                  filesToRemove.add(
+                      buildFileWithoutStats(
+                          partitionValues, currentBaseFilesInPartition.get(writeStat.getFileId())));
+                }
+              }
+            });
+    List<InternalDataFile> filesAdded =
+        fileStatsExtractor
+            .addStatsToFiles(tableMetadata, filesAddedWithoutStats.stream(), table.getReadSchema())
+            .collect(Collectors.toList());
+    return InternalFilesDiff.builder().filesAdded(filesAdded).filesRemoved(filesToRemove).build();
+  }
+
+  public InternalFilesDiff getDiffForReplaceCommit(
+      InternalTable table,
+      HoodieReplaceCommitMetadata replaceCommitMetadata,
+      HoodieInstant commit) {
+    SyncableFileSystemView fsView = fileSystemViewManager.getFileSystemView(metaClient);
+    List<InternalDataFile> filesAddedWithoutStats = new ArrayList<>();
+    List<InternalDataFile> filesToRemove = new ArrayList<>();
+    replaceCommitMetadata
+        .getPartitionToReplaceFileIds()
+        .forEach(
+            (partitionPath, fileIds) -> {
+              List<PartitionValue> partitionValues =
+                  partitionValuesExtractor.extractPartitionValues(
+                      table.getPartitioningFields(), partitionPath);
+              Map<String, HoodieBaseFile> currentBaseFilesInPartition =
+                  fsView
+                      .getLatestBaseFiles(partitionPath)
+                      .collect(Collectors.toMap(HoodieBaseFile::getFileId, Function.identity()));
+              filesToRemove.addAll(
+                  fileIds.stream()
+                      .map(
+                          fileId ->
+                              buildFileWithoutStats(
+                                  partitionValues, currentBaseFilesInPartition.get(fileId)))
+                      .collect(Collectors.toList()));
+            });
+    replaceCommitMetadata
+        .getPartitionToWriteStats()
+        .forEach(
+            (partitionPath, writeStats) -> {
+              List<PartitionValue> partitionValues =
+                  partitionValuesExtractor.extractPartitionValues(
+                      table.getPartitioningFields(), partitionPath);
+              filesAddedWithoutStats.addAll(
+                  writeStats.stream()
+                      .map(
+                          writeStat ->
+                              FSUtils.constructAbsolutePath(
+                                      metaClient.getBasePath(), writeStat.getPath())
+                                  .toString())
+                      .filter(
+                          baseFileFullPath ->
+                              FSUtils.getCommitTimeWithFullPath(baseFileFullPath)
+                                  .equals(commit.requestedTime()))
+                      .map(HoodieBaseFile::new)
+                      .map(hoodieBaseFile -> buildFileWithoutStats(partitionValues, hoodieBaseFile))
+                      .collect(Collectors.toList()));
+            });
+    List<InternalDataFile> filesAdded =
+        fileStatsExtractor
+            .addStatsToFiles(tableMetadata, filesAddedWithoutStats.stream(), table.getReadSchema())
+            .collect(Collectors.toList());
+    return InternalFilesDiff.builder().filesAdded(filesAdded).filesRemoved(filesToRemove).build();
+  }
+
   private AddedAndRemovedFiles getAddedAndRemovedPartitionInfo(
       HoodieTimeline timeline,
       HoodieInstant instant,

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

@@ -0,0 +1,92 @@
+/*
+ * 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.hudi;
+
+import java.util.Collections;
+import java.util.Iterator;
+
+import lombok.AllArgsConstructor;
+import lombok.Value;
+
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieReplaceCommitMetadata;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+
+import org.apache.xtable.model.IncrementalTableChanges;
+import org.apache.xtable.model.InternalTable;
+import org.apache.xtable.model.TableChange;
+import org.apache.xtable.model.storage.InternalFilesDiff;
+
+/**
+ * Computes {@link org.apache.xtable.model.IncrementalTableChanges} between current state of the
+ * table and new completed instant added to the timeline.
+ */
+@Value
+@AllArgsConstructor
+public class HudiIncrementalTableChangeExtractor {
+  HoodieTableMetaClient metaClient;
+  HudiTableExtractor tableExtractor;
+  HudiDataFileExtractor dataFileExtractor;
+
+  public IncrementalTableChanges extractTableChanges(
+      HoodieCommitMetadata commitMetadata, HoodieInstant completedInstant) {
+    InternalTable internalTable =
+        tableExtractor.table(metaClient, commitMetadata, completedInstant);
+    InternalFilesDiff dataFilesDiff;
+    if (commitMetadata instanceof HoodieReplaceCommitMetadata) {
+      dataFilesDiff =
+          dataFileExtractor.getDiffForReplaceCommit(
+              internalTable, (HoodieReplaceCommitMetadata) commitMetadata, completedInstant);
+    } else {
+      dataFilesDiff =
+          dataFileExtractor.getDiffForCommit(internalTable, commitMetadata, completedInstant);
+    }
+
+    Iterator<TableChange> tableChangeIterator =
+        Collections.singleton(
+                TableChange.builder()
+                    .tableAsOfChange(internalTable)
+                    .filesDiff(dataFilesDiff)
+                    .sourceIdentifier(completedInstant.getCompletionTime())
+                    .build())
+            .iterator();
+    return IncrementalTableChanges.builder()
+        .tableChanges(tableChangeIterator)
+        .pendingCommits(Collections.emptyList())
+        .build();
+  }
+
+  public IncrementalTableChanges extractTableChanges(HoodieInstant completedInstant) {
+    InternalTable internalTable = tableExtractor.table(metaClient, completedInstant);
+    Iterator<TableChange> tableChangeIterator =
+        Collections.singleton(
+                TableChange.builder()
+                    .tableAsOfChange(internalTable)
+                    .filesDiff(
+                        InternalFilesDiff.from(Collections.emptyList(), Collections.emptyList()))
+                    .sourceIdentifier(completedInstant.getCompletionTime())
+                    .build())
+            .iterator();
+    return IncrementalTableChanges.builder()
+        .tableChanges(tableChangeIterator)
+        .pendingCommits(Collections.emptyList())
+        .build();
+  }
+}