[Enhancement] Support restore/rollback sync during conversion (1/2)

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

@@ -25,6 +25,7 @@
 import lombok.Builder;
 import lombok.Value;
 
+import org.apache.xtable.model.metadata.SourceMetadata;
 import org.apache.xtable.model.storage.PartitionFileGroup;
 
 /**
@@ -47,4 +48,6 @@ public class InternalSnapshot {
   List<PartitionFileGroup> partitionedDataFiles;
   // pending commits before latest commit on the table.
   @Builder.Default List<Instant> pendingCommits = Collections.emptyList();
+  // Metadata about source table
+  SourceMetadata sourceMetadata;
 }

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

@@ -21,6 +21,7 @@
 import lombok.Builder;
 import lombok.Value;
 
+import org.apache.xtable.model.metadata.SourceMetadata;
 import org.apache.xtable.model.storage.DataFilesDiff;
 
 /**
@@ -36,4 +37,6 @@ public class TableChange {
 
   /** The {@link InternalTable} at the commit time to which this table change belongs. */
   InternalTable tableAsOfChange;
+
+  SourceMetadata sourceMetadata;
 }

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

@@ -0,0 +1,56 @@
+/*
+ * 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.model.metadata;
+
+import java.io.IOException;
+
+import lombok.Builder;
+import lombok.Value;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import org.apache.xtable.model.exception.ParseException;
+
+@Value
+@Builder
+public class SourceMetadata {
+  // The source table snapshot identifier
+  // Snapshot ID in Iceberg, version ID in Delta, and instant <timestamp_action> in Hudi
+  String sourceIdentifier;
+  // The table format of the source table
+  String tableFormat;
+
+  private static final ObjectMapper MAPPER = new ObjectMapper();
+
+  public static SourceMetadata fromJson(String json) throws ParseException {
+    try {
+      return MAPPER.readValue(json, SourceMetadata.class);
+    } catch (IOException e) {
+      throw new ParseException("Failed to deserialize SourceMetadata", e);
+    }
+  }
+
+  public String toJson() {
+    try {
+      return MAPPER.writeValueAsString(this);
+    } catch (IOException e) {
+      throw new ParseException("Failed to serialize SourceMetadata", e);
+    }
+  }
+}

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

@@ -51,6 +51,8 @@ public class TableSyncMetadata {
   /** Property name for the XTABLE metadata in the table metadata/properties */
   public static final String XTABLE_METADATA = "XTABLE_METADATA";
 
+  public static final String XTABLE_SOURCE_METADATA = "XTABLE_SOURCE_METADATA";
+
   Instant lastInstantSynced;
   List<Instant> instantsToConsiderForNextSync;
   int version;

xtable-api/src/main/java/org/apache/xtable/spi/extractor/ConversionSource.java

@@ -78,4 +78,13 @@ public interface ConversionSource<COMMIT> extends Closeable {
    *     false.
    */
   boolean isIncrementalSyncSafeFrom(Instant instant);
+
+  /**
+   * Extract the identifier of the provided commit, the identifier defined as 1. Snapshot ID in
+   * Iceberg 2. Version ID in Delta 3. timestamp in Hudi
+   *
+   * @param commit The provided commit
+   * @return the string version of commit identifier
+   */
+  String getCommitIdentifier(COMMIT commit);
 }

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

@@ -25,6 +25,7 @@
 
 import org.apache.xtable.conversion.TargetTable;
 import org.apache.xtable.model.InternalTable;
+import org.apache.xtable.model.metadata.SourceMetadata;
 import org.apache.xtable.model.metadata.TableSyncMetadata;
 import org.apache.xtable.model.schema.InternalPartitionField;
 import org.apache.xtable.model.schema.InternalSchema;
@@ -76,8 +77,9 @@ public interface ConversionTarget {
    * Starts the sync and performs any initialization required
    *
    * @param table the table that will be synced
+   * @param sourceMetadata The source table metadata for current round of sync
    */
-  void beginSync(InternalTable table);
+  void beginSync(InternalTable table, SourceMetadata sourceMetadata);
 
   /** Completes the sync and performs any cleanup required. */
   void completeSync();
@@ -90,4 +92,7 @@ public interface ConversionTarget {
 
   /** Initializes the client with provided configuration */
   void init(TargetTable targetTable, Configuration configuration);
+
+  /** Return the commit identifier from target table */
+  Optional<String> getTargetCommitIdentifier(String sourceIdentifier);
 }

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

@@ -37,6 +37,7 @@
 import org.apache.xtable.model.InternalSnapshot;
 import org.apache.xtable.model.InternalTable;
 import org.apache.xtable.model.TableChange;
+import org.apache.xtable.model.metadata.SourceMetadata;
 import org.apache.xtable.model.metadata.TableSyncMetadata;
 import org.apache.xtable.model.sync.SyncMode;
 import org.apache.xtable.model.sync.SyncResult;
@@ -73,7 +74,8 @@ public Map<String, SyncResult> syncSnapshot(
                 internalTable,
                 target -> target.syncFilesForSnapshot(snapshot.getPartitionedDataFiles()),
                 startTime,
-                snapshot.getPendingCommits()));
+                snapshot.getPendingCommits(),
+                snapshot.getSourceMetadata()));
       } catch (Exception e) {
         log.error("Failed to sync snapshot", e);
         results.put(
@@ -121,7 +123,8 @@ public Map<String, List<SyncResult>> syncChanges(
                   change.getTableAsOfChange(),
                   target -> target.syncFilesForDiff(change.getFilesDiff()),
                   startTime,
-                  changes.getPendingCommits()));
+                  changes.getPendingCommits(),
+                  change.getSourceMetadata()));
         } catch (Exception e) {
           log.error("Failed to sync table changes", e);
           resultsForFormat.add(buildResultForError(SyncMode.INCREMENTAL, startTime, e));
@@ -149,9 +152,10 @@ private SyncResult getSyncResult(
       InternalTable tableState,
       SyncFiles fileSyncMethod,
       Instant startTime,
-      List<Instant> pendingCommits) {
+      List<Instant> pendingCommits,
+      SourceMetadata sourceMetadata) {
     // initialize the sync
-    conversionTarget.beginSync(tableState);
+    conversionTarget.beginSync(tableState, sourceMetadata);
     // sync schema updates
     conversionTarget.syncSchema(tableState.getReadSchema());
     // sync partition updates