Flink: IcebergFilesCommitter validate dataFile exist start from last committed snapshot.

[Reo-LEI]

This PR is trying to address the issue: #3102

Now, IcebergFilesCommitter will only travel all snapshot history in committer start, and then will keep check snapshot history between last committed snapshot id and current snapshot id.

Fixes #3102

[Reo-LEI]

@openinx @stevenzwu @kbendick could you help to review this PR? :)

[ayush-san]

What will be the value of lastCommittedSnapshotId if I restore the job from a savepoint?

Shouldn't we do something similar for it as we do for maxCommittedCheckpointId but here many things to consider as if we run expire snapshot maintenance procedure before restoring the job from the savepoint(#2482 (comment))

this.lastCommittedSnapshotId = getLastCommittedSnapshotId(table);

[Reo-LEI]

Thanks for your review @ayush-san. As I listed in #2867, restart or restore from checkpoint/savepoint, the lastCommittedSnapshotId will just simply reset to null, and rowDelta will travel all snapshot history to ensure data files still exist and guarantee all snapshot history are valid. If we delete the expired snapshots by maintenance procedure before restoring the job, the validation will travel to the snapshot which is commited by expire snapshot maintenance procedure. I think that whould be ok.

This optimization just trying to speed up commit result in runtime, and I want to keep it simple, so I think we don't need to save the lastCommittedSnapshotId and restore it from checkpoint/savepoint. but I want to hear @openinx @jackye1995 @stevenzwu @kbendick opinions.

[ayush-san]

@Reo-LEI Can you please help me understand how the #2867 help in solving this validation error

I agree with you that we can handle the lastCommittedSnapshotId in a separate PR and get this reviewed because this will really speed up the commit time which increases with time. I have seen my flink job checkpoint time increases to 10-15mins from 100-200ms

[Reo-LEI]

@ayush-san I'm very sorry, this is my mistake, I want to link to #3102 (comment), but not #2867 .

And I think this PR will not solving #2482. if the validation error not fixed, this PR will encounter the same problem. I think we should follow #2603 (comment) this to check the exists files to fix the validation error.

[ayush-san]

Yes, but with your PR we can run the snapshot expire task with the flink job running since you are updating the lastCommittedSnapshotId. The only case now left is that when we start the flink job from a checkpoint, we will encounter the same problem.

But if are doing that for one case, we can mimic the same for case when we restore the job from a checkpoint. Anyways we can tackle this in a separate PR and discuss it with @rdblue and @openinx

[Reo-LEI]

but with your PR we can run the snapshot expire task with the flink job running since you are updating the lastCommittedSnapshotId.

I think the snapshot expire task don’t cause the validation error is becasue the validationHistory will just travel to the lastCommittedSnapshot and stop valid the removed snapshots which is older than the lastCommittedSnapshot. But once the lastCommittedSnapshot be removed(e.g. the snapshot expire task run multiple time between two checkpoints) , you will encounter the validation error again.

I think store the lastCommittedSnapshotId is not a correct way to resolve the validation error(#2482). Because we cann't guarantee the lastCommittedSnapshot what we store its snapshot id will alway exists when we restore the flink job. The lastCommittedSnapshot probably has been removed when we restore the flink job, and the validation error will raise again. @ayush-san

[Reo-LEI]

Close for #3258