[GOBBLIN-1707] Enhance IcebergDataset to detect when files already at dest then proceed with only delta

[phet]

Dear Gobblin maintainers,

Please accept this PR. I understand that it will not be reviewed until I have checked off all the steps below!

JIRA

• My PR addresses the following Gobblin JIRA issues and references them in the PR title. For example, "[GOBBLIN-XXX] My Gobblin PR"
  • https://issues.apache.org/jira/browse/GOBBLIN-1707

Description

• Here are some details about my PR, including screenshots (if applicable):

Enhance IcebergDataset to detect when files already at dest then proceed with only delta

Tests

• My PR adds the following unit tests OR does not need testing for this extremely good reason:

unit tests included

Commits

• My commits all reference JIRA issues in their subject lines, and I have squashed multiple commits if they address the same issue. In addition, my commits follow the guidelines from "How to write a good git commit message":
  1. Subject is separated from body by a blank line
  2. Subject is limited to 50 characters
  3. Subject does not end with a period
  4. Subject uses the imperative mood ("add", not "adding")
  5. Body wraps at 72 characters
  6. Body explains "what" and "why", not "how"

[codecov-commenter]

Codecov Report

Merging #3575 (72b5662) into master (71da34b) will decrease coverage by 0.00%.
The diff coverage is 67.59%.

@@             Coverage Diff              @@
##             master    #3575      +/-   ##
============================================
- Coverage     46.90%   46.90%   -0.01%     
- Complexity    10610    10651      +41     
============================================
  Files          2111     2114       +3     
  Lines         82533    82893     +360     
  Branches       9178     9237      +59     
============================================
+ Hits          38714    38877     +163     
- Misses        40261    40450     +189     
- Partials       3558     3566       +8     

Impacted Files	Coverage Δ
...obblin/util/function/CheckedExceptionFunction.java	0.00% <0.00%> (ø)
...bblin/util/measurement/GrowthMilestoneTracker.java	0.00% <0.00%> (ø)
...lin/data/management/copy/iceberg/IcebergTable.java	83.63% <70.00%> (-5.05%)	⬇️
...n/data/management/copy/iceberg/IcebergDataset.java	86.77% <94.02%> (+4.95%)	⬆️
...a/management/copy/iceberg/IcebergSnapshotInfo.java	89.47% <100.00%> (+1.23%)	⬆️
...bblin/service/monitoring/FsJobStatusRetriever.java	55.17% <100.00%> (ø)
...e/modules/orchestration/MysqlUserQuotaManager.java	47.34% <0.00%> (-26.53%)	⬇️
...a/org/apache/gobblin/util/limiter/NoopLimiter.java	40.00% <0.00%> (-20.00%)	⬇️
...lin/service/modules/flow/MultiHopFlowCompiler.java	50.28% <0.00%> (-18.58%)	⬇️
...lin/util/filesystem/FileSystemInstrumentation.java	85.71% <0.00%> (-14.29%)	⬇️
... and 30 more

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

[ZihanLi58]

I have several questions here:

1. From the algorithm, it seems as long as there is one new snapshot generate on source, we will go through all the data files available on source to do copy even there is only one new file added. Not sure if I'm missing anything here?
2. How do we plan to handle the file deletion on source? i.e. expire snapshot operation? If file A is deleted on source (I mean deleted from HDFS), how do we plan to apply the same on destination?

[phet]

1. it seems as long as there is one new snapshot generate on source, we will go through all the data files available on source to do copy even there is only one new file added.

actually we'll always go through the complete metadata on source to list every file reachable from at least one snapshot. that is, unless there's not any 'new', unreplicated snapshot. actual copy however only happens for files that are not present on the destination. further, we need not examine every file, to determine whether it exists on dest. rather, thanks to the immutability of iceberg files, we may short-circuit evaluation of an entire subtree of the iceberg metadata, when the root (e.g. manifest-list or manifest) is found already to exist at dest. to document details, I've added the comment // ALGO: in IcebergDataset.getFilePathsToFileStatus()

edit: upon further consideration, I added top-level short-circuiting of any manifest scanning, in the special case of the (root) metadata.json file already found to be replicated to destination (and changed text above accordingly).

2. How do we plan to handle the file deletion on source? i.e. expire snapshot operation?

good question! the answer is: distcp is not responsible. instead we expect reachability analysis and orphan file deletion to happen elsewhere. a good candidate would be to encapsulate within the destination catalog that replication will eventually--forthcoming enhancement!--come to register the copied 'metadata.json' file with. e.g. that catalog would hold the metadata version prior to the registration and could easily determine which snapshots 'expired' from the act of replacing the older metadata file with the newer one (just copied from source)

[ZihanLi58]

overall look good to me, just minor comments

[ZihanLi58]

Can we add comment here to mention that only oldest snapshot which most likely already been copied contains the all old files, the mfi from newer snapshot only contain new added file in those manifest?

[phet]

those are definitely the semantics of using IcebergTable.getIncrementalSnapshotInfosIterator(). it's in the javadoc there. where do you want me to repeat it over here? maybe should I note those semantics above here when I call that method?

[ZihanLi58]

Or we can add more clarity in IcebergTable.getIncrementalSnapshotInfosIterator(), at least I was missing the assumption that it return the snapshot from oldest to latest. and files only appear once in earliest snapshot that contains it. Might be just me though...

[phet]

sure thing. I added this:

This means the {@link IcebergSnapshotInfo#getManifestFiles()} for the (n+1)-th element of the iterator will omit all manifest files and listed data files, already reflected in a {@link IcebergSnapshotInfo#getManifestFiles()} from the n-th or prior elements. Given the order of the {@link Iterator<IcebergSnapshotInfo>} returned, this mirrors the snapshot-to-file dependencies: each file is returned exactly once with the (oldest) snapshot from which it first becomes reachable.

Only the final {@link IcebergSnapshotInfo#getMetadataPath()} is present (for the snapshot it itself deems current).

[ZihanLi58]

not sure whether this is true for all files, as metadata file is immutable, data file might be updated?

[phet]

it's technically possible to change files in place, but to do so, breaks the iceberg's repeatability. it's not something we should ever encourage... instead write new files and create a snapshot w/ those that substitutes out and hence replaces the original ones!

for distcp specifically, the real issue w/ in-place mods to data files is that every delta copy must devolve into a full comparison of the filestatus (between source and dest) for the entire iceberg table. that's a huge amount of effort in some cases... all because of misbehaving writers/updaters.

I suggest that if we do learn we're working w/ writers that do this, we later return here to add the necessary complexity. it's likely something we'll control via configuration, so it's not always on.

[ZihanLi58]

Makes sense, given you have the comment mention this behavior as well, it should be fine.

[ZihanLi58]

LGTM