Core: Fix RewriteTablePath Incremental Replication

[barronfuentes]

This contribution attempts to resolve a bug with the RewriteTablePath action in Apache Iceberg. While the current implementation of RewriteTablePath supports the ability to replicate a portion of an Iceberg tables history, it does not support the ability to do so repeatedly while maintaining history of previous replicated portions in the target table.

For example, assume an Iceberg table has two versions: A, B. Immediately after the first version (A) was created in the table, the Rewrite Table Path action was used to replicate the table for disaster recovery. After the second version (B) was added to the source table, Rewrite Table Path was used once again to replicate (append) only the second version to the target table. In the current state, the target table will have all of the metadata and data files associated with both versions A & B, but only the records contained within the most recent replicated version (B) will be available to queries. This pull request aims to address this issue and provide the ability to incrementally replicate Iceberg tables in a manner which results in a target table with contents that are consistent with the source table.

The cause of the issue exists in how new Manifest List files are created. A filter which prevents rewriting Manifest Files out of scope for replication is also applied to the content of rewritten Manifest Lists. The exclusion of these historical references is the reason why queries can only access records contained in the latest replication and not the previous replications.

[barronfuentes]

This approach assumes that all versions before startVersion have been replicated to the target table.

[barronfuentes]

@szehon-ho - Would love to get your feedback on these changes.

[dramaticlly]

I did some look on this, I think right now the only strict delta manifest-list/manifests/data files are rewritten if start_version is provided. So for incremental rewrite and copy does not result in cumulative of all deltas when read target table. Collected some details from provided unit test examples to help visualize the problem

1st snapshot: 4355148708777346214
2nd snapshot: 936971667881185972

• source table

+--------------------------------------------------------------------------------+-----------------------------------------------------------+-------------------+
|manifest-list                                                                   |path                                                       |added_snapshot_id  |
+--------------------------------------------------------------------------------+-----------------------------------------------------------+-------------------+
|1224/metadata/snap-936971667881185972-1-7a1b9514-44b9-418e-a6b3-dd724a5daa01.avro|1224/metadata/7a1b9514-44b9-418e-a6b3-dd724a5daa01-m0.avro|936971667881185972 |
|1224/metadata/snap-936971667881185972-1-7a1b9514-44b9-418e-a6b3-dd724a5daa01.avro|1224/metadata/a3362153-e2aa-4e7c-a04e-7ea9f525d0ce-m0.avro|4355148708777346214|
+--------------------------------------------------------------------------------+-----------------------------------------------------------+-------------------+

+----------------------------------------------------------+------+-------------------+-----------------------------------------------------------------------+
|manifest-file                                             |status|snapshot_id       |file_path                                                               |
+----------------------------------------------------------+------+-------------------+-----------------------------------------------------------------------+
|1224/metadata/7a1b9514-44b9-418e-a6b3-dd724a5daa01-m0.avro|1     |936971667881185972 |1224/data/00000-16-2c43c06a-e56e-4c74-a498-393398ac66df-0-00001.parquet|
|1224/metadata/a3362153-e2aa-4e7c-a04e-7ea9f525d0ce-m0.avro|1     |4355148708777346214|1224/data/00000-2-f37fff9f-1075-4e06-85e1-fa7c3f22ac48-0-00001.parquet |
+----------------------------------------------------------+------+-------------------+-----------------------------------------------------------------------+

• target table

+--------------------------------------------------------------------------------+-----------------------------------------------------------+------------------+
|manifest-list                                                                   |path                                                       |added_snapshot_id |
+--------------------------------------------------------------------------------+-----------------------------------------------------------+------------------+
|4614/metadata/snap-936971667881185972-1-7a1b9514-44b9-418e-a6b3-dd724a5daa01.avro|4614/metadata/7a1b9514-44b9-418e-a6b3-dd724a5daa01-m0.avro|936971667881185972|
+--------------------------------------------------------------------------------+-----------------------------------------------------------+------------------+

+----------------------------------------------------------+------+-------------------+-----------------------------------------------------------------------+
|manifest-file                                             |status|snapshot_id       |file_path                                                               |
+----------------------------------------------------------+------+-------------------+-----------------------------------------------------------------------+
|4614/metadata/7a1b9514-44b9-418e-a6b3-dd724a5daa01-m0.avro|1     |936971667881185972|4614/data/00000-16-2c43c06a-e56e-4c74-a498-393398ac66df-0-00001.parquet |
+----------------------------------------------------------+------+-------------------+-----------------------------------------------------------------------+

So incremental copy after 1st snapshot, read of target table ends up with only the delta for all iceberg metadata related to snapshot 936971667881185972 and manifest-list snap-936971667881185972 in source contains 2 manifest entry but only 1 in target table due to manifest filtering

[dramaticlly]

So if I understand the problem correctly, today we filter the manifest based on snapshot when rewrites manifest list for incremental rewrite. snapshot based filtering shall only apply to copy plan but not change the content when rewrite the manifest list with the new path.

[dramaticlly]

change LGTM, some comments on unit tests

[dramaticlly]

FYI @flyrain @amogh-jahagirdar I think we might want take this fix as part of 1.8 release if possible.

[szehon-ho]

This looks good, it looks like a similar issue as #12006 but for metadata files.

[szehon-ho]

Thanks a lot @barronfuentes and also @dramaticlly for evaluating and reviewing it!