[FLINK-39339][table-planner] Consider immutable cols to infer sink required updated mode traits

[xuyangzhong]

What is the purpose of the change

When deciding whether a sink can accept ONLY_UPDATE_AFTER (i.e., drop UPDATE_BEFORE), the current logic requires the input's upsert key to exactly match the sink's primary key. This is overly conservative when immutable columns are present.I f the extra columns in the sink PK (beyond the upsert key) are all immutable, then the sink PK is effectively a valid upsert key and UPDATE_BEFORE can still be safely dropped.

Brief change log

• In inferSinkRequiredTraits, query the input's immutable columns via FlinkRelMetadataQuery.getImmutableColumns
• Relax the upsert-key-vs-sink-PK check: instead of requiring exact equality, allow the sink PK to match if it contains a upsert key and all remaining columns are immutable
• Add corresponding test cases in ChangelogModeInferenceTest

Verifying this change

New tests are added to verify this pr.

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): no
• 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?

[flinkbot]

CI report:

• eb9efe4 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[lincoln-lil]

@xuyangzhong +1 for the optimization! I left some comments there.

For the tests, we should add at least one test case triggering the DELETE_BY_KEY path with immutable columns.

[lincoln-lil]

The word "Different" implies an equality comparison (the old method areUpsertKeysDifferentFromPk indeed used !exists(_.equals(sinkPks))). However, the new logic actually checks a subset/coverage relationship, so the method name should more accurately reflect the "coverage"(satisfy) semantics.

[xuyangzhong]

renamed it to canUpsertKeysWithImmutableColsSatisfyPk

[lincoln-lil]

The double negation might be readability issue here.

A minimal rewrite version(maybe slightly improvement or you can think of other writings):

val sinkPkCovered = changeLogUpsertKeys.exists { uk =>
  val ukWithinPk = sinkPks.contains(uk)
  val pkFullyCovered = uk.union(effectiveImmutableCols).contains(sinkPks)
  ukWithinPk && pkFullyCovered
}
!sinkPkCovered

[xuyangzhong]

I have updated the code here and add an example in the method comment.

[xuyangzhong]

@xuyangzhong +1 for the optimization! I left some comments there.

For the tests, we should add at least one test case triggering the DELETE_BY_KEY path with immutable columns.

Currently, immutable cols cannot be defined on sources that output rowkind DELETE, making it difficult to add tests. Therefore, I've decided to abandon this approach and revert the changes to SatisfyDeleteKindTraitVisitor, focusing only on modifications to SatisfyUpdateKindTraitVisitor. WDYT?

[lincoln-lil]

@xuyangzhong +1 for the optimization! I left some comments there.
For the tests, we should add at least one test case triggering the DELETE_BY_KEY path with immutable columns.

Currently, immutable cols cannot be defined on sources that output rowkind DELETE, making it difficult to add tests. Therefore, I've decided to abandon this approach and revert the changes to SatisfyDeleteKindTraitVisitor, focusing only on modifications to SatisfyUpdateKindTraitVisitor. WDYT?

Sounds reasonable.

[lincoln-lil]

@xuyangzhong Thanks for the update! +1