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

danielhumanmod · 2024-10-27T23:22:52Z

Important Read

Aim to address Properly reflect rollbacks/restores in target tables #40

What is the purpose of the pull request

Previously, if a rollback/restore occurred in the source table, XTable would reflect it as file changes (added or deleted) in the target table. In this PR, we aim to improve this by issuing a rollback command in the target tables, ensuring more consistent histories between the source and target. This approach is also more efficient, as it allows us to restore directly to a specific version/snapshot instead of computing a large diff against the table’s current state.

Brief change log

This is the first part of this enhancement (1/2), focusing primarily on supporting commit-level information for all target table formats and the ability to locate certain target commit with given source identifier.

Add a source identifier in target transaction

snapshot ID in Iceberg, version ID in Delta, and instant timestamp in Hudi

Locate target commit with given source identifier

Additional Info

Fallback scenarios

Fallback will happen when a rollback or restore is detected in the source table, but the corresponding commit is not found in the target table. We will still leverage the rollback information from the source, but this round of sync will be treated as file changes in the target table, following the previous behavior.

Here’s an example:

Iceberg (Source)          Delta (Target)  
┌────────────┐      ┌─────────────────────┐
│ Snapshot 0 │ ◀  ▶ │ Version 0 (Synced)  │  
│ Snapshot 1 │ ◀  ▶ │ Version 1 (Synced)  │  
│ Snapshot 2 │ ◀  ▶ │ Version 2 (Synced)  │  
│ Snapshot 3 │ ◀  ▶ │ Version 3 (Synced)  │  
│ Snapshot 4 │      │                     │  
│ Snapshot 5 │ ◀  ▶ │ Version 4 (Synced)  │
└────────────┘      └─────────────────────┘

If rollback to Snapshot 0, 1, 2, 3, 5, can find a corresponding commit in target
If rollback to snapshot 4, can not find a corresponding commit in target. This makes it impossible to issue a rollback command, but we can still make use of the info of source snapshot to issue a file changes

In this case, we can not guarantee complete metadata consistency between the source and target, but it helps reduce some computation.

Verify this pull request

This pull request is already covered by existing tests, all existing tests should pass

the-other-tim-brown · 2024-11-12T14:03:07Z

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

@@ -47,9 +49,20 @@ public IncrementalTableChanges extractTableChanges(
        commitsBacklog.getCommitsToProcess().stream()
            .map(conversionSource::getTableChangeForCommit)


I think that we'll want the identifier on the commit level, right?

I think that we'll want the identifier on the commit level, right?

Thanks for the response @the-other-tim-brown.

Yes, ideally, every commit in source table should directly map to one in target table. However, based on my understanding of how XTable works, this isn’t guaranteed. Instead, the mapping (Source -> Target) is more like a N:1 mapping, which means:

Every commit in the target table has a corresponding mapping in the source table.

Not every commit in the source table has a one-to-one mapping in the target table.

The reason is, between each sync(), there could be multiple changes on source, and all these changes will sync as only one commit in target, just like this example

Iceberg (Source) Delta (Target) ┌────────────┐ ┌─────────────────────┐ │ Snapshot 0 │ ◀ ▶ │ Version 0 (Synced) │ (can map to snapshot 0) │ Snapshot 1 │ │ │ │ Snapshot 2 │ │ │ │ Snapshot 3 │ │ │ │ Snapshot 4 │ │ │ │ Snapshot 5 │ ◀ ▶ │ Version 1 (Synced) │ (can map to snapshot 5) └────────────┘ └─────────────────────┘

Given this, I’ve chosen to use the information from the latest commit in the source table as the source identifier.
But my understanding might be wrong, appreciate if there is any feedback or suggestion :)

This is true for snapshot sync but with incremental sync, there are multiple commits all synced to the target as their own commits. One thing we should confirm is whether we are able to track at a per commit level in each target. I am unsure if the metadata history is tracked in Iceberg and Delta. It is tracked in the Hudi target

Thanks for pointing this out—it’s an important detail! I’ll investigate further into the ability to track commit-level information in Iceberg and Delta. I will get back to you once I figure it out

danielhumanmod · 2024-11-20T08:51:04Z

Hi @the-other-tim-brown, based on my investigation, both Iceberg and Delta support storing commit-level information, but we might need to adjust our current code. Here’s a summary of the findings:

Iceberg
- Can use the snapshot summary to store snapshot-level information.
- However, we currently store sync metadata in properties, which is table-level information.
Delta
- Can use commit tags to store commit-level information.
- However, we currently store sync metadata in configurations, which is table-level information.

To align with these capabilities, some code adjustments may be needed for both Iceberg and Delta. I’ll start working on a proof of concept to explore this, and will get back to you once it’s completed.

the-other-tim-brown · 2024-11-22T04:35:10Z

@danielhumanmod Thanks for looking into it. I think being able to store this at a commit level better aligns with my our intentions so it would be great to fix this.

danielhumanmod · 2024-11-24T22:35:32Z

@danielhumanmod Thanks for looking into it. I think being able to store this at a commit level better aligns with my our intentions so it would be great to fix this.

Hi @the-other-tim-brown, I have updated the code to support the commit level info, appreciate any review and suggestion!

the-other-tim-brown · 2024-12-01T22:57:56Z

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

@@ -90,4 +91,7 @@ public interface ConversionTarget {

  /** Initializes the client with provided configuration */
  void init(TargetTable targetTable, Configuration configuration);
+
+  /** Return the commit identifier from target table */


Can you be more specific here? I am assuming this is the latest commit identifier but it should be more specific if the table can have more than one over time.

Can you be more specific here? I am assuming this is the latest commit identifier but it should be more specific if the table can have more than one over time.

Do you mean the snapshot sync scenario? Because In incremental sync, we will ensure target and source commit has 1:1 mapping. However, in snapshot sync, we will use the commit id of current (latest) snapshot even though the changes might contains more than one commit.

To ensure we’re aligned, the context of this function is:

It is designed to retrieve the corresponding target commit based on a given source identifier during a sync operation. Here’s the scenario:

Assume:

Last completed sync (1 snapshot, 1 incremental) is source commit 3

We’re starting the new sync process from source commit 4

Source table commit history:
• 1 (UPDATE)
• 2 (UPDATE)
• 3 (UPDATE)
• 4 (ROLLBACK to 2)
• 5 (UPDATE)
• 6 (UPDATE)
Target table commit history (mapped by source identifiers):
• 1 (mapped to source id 2)
• 2 (mapped to source id 3)

When syncing the ROLLBACK operation (source commit 4), we need to identify the corresponding target commit that aligns with the source identifier 2 (which is 1 in target).

That makes sense to me but the java doc can be a bit more clear. It should also include context on when the Optional will be empty

the-other-tim-brown · 2024-12-01T22:59:13Z

xtable-core/src/main/java/org/apache/xtable/delta/DeltaConversionTarget.java

@@ -215,13 +224,64 @@ public String getTableFormat() {
    return TableFormat.DELTA;
  }

+  @Override
+  public Optional<String> getTargetCommitIdentifier(String sourceIdentifier) {


If there is no commit Identifier in the latest commit, should this just return an empty Optional?

If there is no commit Identifier in the latest commit, should this just return an empty Optional?

Yes, if we can not find the corresponding target commit, we will return an empty Optional

the-other-tim-brown · 2024-12-01T23:00:37Z

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

   */
-  void beginSync(InternalTable table);
+  void beginSync(InternalTable table, String sourceIdentifier);


I am wondering if it makes sense to attach the source format as part of the identifier. Then we can also have some metadata about what the writer format was.

I am wondering if it makes sense to attach the source format as part of the identifier. Then we can also have some metadata about what the writer format was.

Good idea. I create a SourceMetadata as a commit-level information attached in target table commit so that It would be much easier if we want add more information in the future

the-other-tim-brown · 2024-12-07T15:34:21Z

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";


Instead of having a separate metadata, is it possible to embed this within the metadata object that we have today?

Good catch! The current metadata is table-level information, while the new one is commit-level, but it is true that we can reuse this key.

And maybe we could consider moving the source format to table-level information and leaving the source identifier at the commit level to reduce redundancy.

I am wondering if all of the information should actually be stored at the commit level instead of table level. In the Hudi target, the information is already stored at the commit level since that was the intention and the format I knew best. The others, it seems like I did not add the metadata to the correct location.

The source format should be allowed to change over time for flexibility so I think it is best to keep that at the commit level as well.

Ah I see your point, that's a good way. I’ll get started on it.

the-other-tim-brown · 2024-12-07T16:24:22Z

xtable-core/src/main/java/org/apache/xtable/delta/DeltaConversionTarget.java

@@ -264,7 +330,8 @@ private void commitTransaction() {
      transaction.updateMetadata(metadata, false);
      transaction.commit(
          actions,
-          new DeltaOperations.Update(Option.apply(Literal.fromObject("xtable-delta-sync"))));
+          new DeltaOperations.Update(Option.apply(Literal.fromObject("xtable-delta-sync"))),
+          ScalaUtils.convertJavaMapToScala(getCommitTags()));


What is the difference between putting this info in tags compared to the metadata we are currently using? Should we consolidate?

Here xtable-delta-sync is the operation, and the tags is more recommended to store the metadata as k-v pairs

the-other-tim-brown · 2024-12-07T16:25:30Z

xtable-core/src/test/java/org/apache/xtable/delta/TestDeltaSync.java

    validateDeltaTable(basePath, new HashSet<>(Arrays.asList(dataFile2, dataFile3)), null);
+    assertTrue(targetIdentifier2.isPresent());
+    assertEquals("1", targetIdentifier2.get());


Let's have 1 test where we return an empty option as well

the-other-tim-brown · 2024-12-07T16:27:18Z