[722] Implement iceberg versions for TableFormat and HoodieTableMetadata

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

@@ -52,4 +52,15 @@ public class InternalTable {
   Instant latestCommitTime;
   // Path to latest metadata
   String latestMetadataPath;
+  /**
+   * Identifies the source-table operation this state was derived from, or null when the source does
+   * not supply one. Written by the source's table extractor and carried into the target's metadata
+   * by {@link org.apache.xtable.spi.sync.TableFormatSync}, so that a target reading its own
+   * metadata back can tell which source operation it last applied.
+   *
+   * <p>The contents are specific to the source format. The Hudi extractor writes a serialised Hudi
+   * instant, which only Hudi-aware code should parse; every other target must treat this as an
+   * opaque string.
+   */
+  String latestTableOperationIdentifier;
 }

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

@@ -28,6 +28,7 @@
 import lombok.Value;
 
 import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.SerializationFeature;
 import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
@@ -46,6 +47,10 @@ public class TableSyncMetadata {
       new ObjectMapper()
           .registerModule(new JavaTimeModule())
           .configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false)
+          // A reader on an older version has to tolerate fields a newer writer added, since this
+          // blob is persisted in target-table metadata and is read back by whichever version
+          // happens to open the table next.
+          .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
           .setSerializationInclusion(JsonInclude.Include.NON_NULL);
 
   /** Property name for the XTABLE metadata in the table metadata/properties */
@@ -56,6 +61,12 @@ public class TableSyncMetadata {
   int version;
   String sourceTableFormat;
   String sourceIdentifier;
+  /**
+   * Identifies the source-table operation this sync corresponds to. The contents are specific to
+   * the source format, so a target must treat them as opaque. See {@link
+   * org.apache.xtable.model.InternalTable#latestTableOperationIdentifier}.
+   */
+  String latestTableOperationIdentifier;
 
   /**
    * @deprecated Use {@link #of(Instant, List, String, String)} instead. This method exists for
@@ -64,20 +75,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 latestTableOperationIdentifier) {
     return new TableSyncMetadata(
         lastInstantSynced,
         instantsToConsiderForNextSync,
         CURRENT_VERSION,
         sourceTableFormat,
-        sourceIdentifier);
+        sourceIdentifier,
+        latestTableOperationIdentifier);
   }
 
   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.getLatestTableOperationIdentifier());
     conversionTarget.syncMetadata(latestState);
     // sync schema updates
     conversionTarget.syncSchema(tableState.getReadSchema());

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,21 @@ 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() resolves provider classes lazily, so it throws too and has to be guarded.
+        if (!iterator.hasNext()) {
+          break;
+        }
+        target = iterator.next();
+      } catch (ServiceConfigurationError | LinkageError error) {
+        // A registered target whose engine library is absent. Skip it so a subset of engines works;
+        // a missing engine for the requested format still fails below as NotSupportedException.
+        log.warn("Skipping a ConversionTarget whose engine library is not on the classpath", error);
+        continue;
+      }
       if (target.getTableFormat().equalsIgnoreCase(tableFormatName)
           && isDeltaKernelTarget(target) == useKernel) {
         return target;
@@ -96,7 +113,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,121 @@ public InternalFilesDiff getDiffForCommit(
     return InternalFilesDiff.builder().filesAdded(filesAdded).filesRemoved(filesRemoved).build();
   }
 
+  /**
+   * Derives the file diff from the metadata of the commit being written, rather than by comparing
+   * two committed states as {@link #getDiffForCommit(HoodieInstant, InternalTable, HoodieInstant,
+   * HoodieTimeline)} does. Requires the constructor taking a {@link FileSystemViewManager}, since
+   * it reads the live file system view. Log files are skipped, so merge-on-read updates are not
+   * represented.
+   */
+  public InternalFilesDiff getDiffFromCommitMetadata(
+      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();
+  }
+
+  /**
+   * Replace-commit counterpart of {@link #getDiffFromCommitMetadata}. Files the replace commit
+   * supersedes are reported as removed, files it wrote as added.
+   */
+  public InternalFilesDiff getDiffFromReplaceCommitMetadata(
+      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,