[flink] optimize flink Data-Evolution-Merge-Into

[steFaiz]

Purpose

This PR optimize Data-Evolution-Merge-Into in several aspects:

1. Introduce a MergeIntoUpdateChecker to check if some global-indexed columns are updated (This is same as Spark's implementation)
2. Use calcite to rename target table (current implementation is based on regex, which is very unstable)
3. Use calcite to find _row_id field (if exists) in source table. We can eliminate join process.

Linked issue: None

Tests

Please see: org.apache.paimon.flink.action.DataEvolutionMergeIntoActionITCase.

API and Format

None

Documentation

None

Generative AI tooling

Full hand-writing.

[JingsongLi]

Maybe you can just use CommitOperator with commit.strict-mode.last-safe-snapshot configured in DataEvolutionMergeIntoAction.

[steFaiz]

Maybe you can just use CommitOperator with commit.strict-mode.last-safe-snapshot configured in DataEvolutionMergeIntoAction.

@JingsongLi Thanks for your advise! I've modified my code.

[JingsongLi]

I mean you can use configure scanned snapshot id to commit.strict-mode.last-safe-snapshot, this snapshot id should be used by MERGE INTO source.

And you don't need to introduce MergeIntoCommitterOperator, just use CommitterOperator, commit.strict-mode.last-safe-snapshot will solve your FilterAndCommit problem.

[steFaiz]

I mean you can use configure scanned snapshot id to commit.strict-mode.last-safe-snapshot, this snapshot id should be used by MERGE INTO source.

And you don't need to introduce MergeIntoCommitterOperator, just use CommitterOperator, commit.strict-mode.last-safe-snapshot will solve your FilterAndCommit problem.

@JingsongLi Thanks for your explaination! I've set the commit.strict-mode.last-safe-snapshot as the latestSnapshotId in the action's constructor.

I introduce the new committer operator wrapping a CommitterOperator to check the received committables, so that we could throw and error or drop partition index if merge into action modified some global-indexed columns.

An alternative approach is to introduce a new operator before committer e.g. UpdateCheckerOperator. Which one do you prefer, I'm willing to modify my code.

[JingsongLi]

@steFaiz Sounds good to me.

[steFaiz]

@steFaiz Sounds good to me.

@JingsongLi Appreciate for it! Do you mean I should introduce a UpdateCheckerOperator or keep the new committer(just resolve the conflict)?

[JingsongLi]

@steFaiz Sounds good to me.

@JingsongLi Appreciate for it! Do you mean I should introduce a UpdateCheckerOperator or keep the new committer(just resolve the conflict)?

introduce a UpdateCheckerOperator

[JingsongLi]

+1