[FLINK-34380][table] Fix incorrect RowKind in MiniBatch join when INSERT is folded with retract

[Au-Miner]

What is the purpose of the change

Fix MiniBatch join output RowKind when an INSERT is folded away with a retract in the same batch. A later UPDATE_AFTER for the same key should be emitted as INSERT if no prior row exists downstream.

Brief change log

• Update InputSideHasUniqueKeyBundle#addRecord and foldRecord to track insert-and-cleared unique keys and convert the following UPDATE_AFTER to INSERT.
• Update JoinKeyContainsUniqueKeyBundle#addRecord and foldRecord with the same correction for join keys that contain the unique key.
• Clear the inserted-and-cleared tracking state when the bundle is cleared.

Verifying this change

Updated BufferBundleTest to assert that +I followed by +U keeps the latest record as INSERT. Updated StreamingMiniBatchJoinOperatorTest with MiniBatch join cases for both join-key-contains-unique-key and input-side-has-unique-key paths, asserting that a +I/-U/+U sequence emits INSERT.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: no
• The runtime per-record code paths (performance sensitive): yes
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no
• If yes, how is the feature documented? not applicable

[flinkbot]

CI report:

• 1e6b3e5 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[Au-Miner]

Hi @SteveStevenpoor , thanks for the review. If you have any other questions, please feel free to contact me

[SteveStevenpoor]

Hi @SteveStevenpoor , thanks for the review. If you have any other questions, please feel free to contact me

Thanks for your contribution @Au-Miner. The newly added tests look great. Though, I still have a concern. Suppose the first three records in the bundle are +I -U +U.

In this case:

• JoinKeyContainsUniqueKeyBundle results in +I
• InputSideHasUniqueKeyBundle results in +I
• InputSideHasNoUniqueKeyBundle results in +U

I understand that we usually treat +I and +U the same in many situations. However, if the first +U is not a problem for the NoUniqueKey case, why is it important in the other cases?

I’m also concerned that this could be confusing for users, because the resulting record’s RowKind would depend on the joined table configuration.

That’s my only remaining concern. Please let me know what you think.