Core, Spark: Scan only live entries in RewriteTablePathUtil

[dramaticlly]

Instead of scanning all entries in data/manifest for identifying list of content files to copy, scan only the live one. This is essential to prevent rewrite table path to carry the files already expired as part of snapshot expiration in the source table.

Existing logic fetch both added/existing/deleted entry from manifest to collect list of content files to be copied and rely on reducer for deduplicate based on file name.

However we want to avoid the scenario where the given content file with only deleted status in older manifest, as snapshot expiration might already removed the snapshot which reference the given content file, and deleted as part of snapshot expiration.

With some concrete examlpe to help with explanation,

assume we have 3 snapshots of overwrite operation
1. 8729031490038117099 (dataSeq=1, added d2.parquet)
2. 6024975807438659167 (dataSeq=2, removed d2,parquet and added d3.parquet)
3. 4358334817990999907 (dataSeq=3, removed d3.parquet and added d4.parquet)

the expiration of first snapshot 8729031490038117099, will remove d2.parquet on disk,
second snapshot 6024975807438659167 might still have data manifest entry of deleted (status=2) for d2.parquet.

However it's not desired to include d2.parquet as part of files for path rewrite.

CC @szehon-ho

[szehon-ho]

Yes, thanks for fixing the issue (found by our internal usage).

I wonder, because the deleted entry may be important for CDC (to mark that this file at some point existed), is another possibility to skip the deleted file for copy? cc @flyrain for thoughts as well