[FLINK-39286][table] Introduce immutable columns constraint in schema

[xuyangzhong]

What is the purpose of the change

This pr contains 2 parts:

1. [FLINK-39286][table] Introduce immutable columns constraint in schema
2. [FLINK-39287][table-planner] Introduce FlinkRelMdImmutableColumns to implement the derivation logic of some simple nodes and update the logic to infer upsert key with immutable columns
3. [FLINK-39314][table-planner] Allow input to drop update before if filter is applied on any of upsert keys

Brief change log

• Introduce immutable constraint in public apis such as Schema, ResovledSchema, Constraint, etc
• Introduce FlinkRelMdImmutableColumns to infer immutable columns for some simple nodes
• Update FlinkRelMdUniqueKeys to infer upsert key with immutable columns
• Update the logic to infer ONLY_AFTER changelog mode in FlinkChangelogModeInferenceProgram
• Add tests for all changes

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): yes
• 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? yes
• If yes, how is the feature documented? later

[flinkbot]

CI report:

• 0774f24 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[lincoln-lil]

@xuyangzhong Can you rebase the master? There was a known issue that just fixed (#27808).

[lincoln-lil]

@xuyangzhong Thanks for the pr! Overall looks good. I left some comments there.

One question regarding the SHOW CREATE TABLE path: I understand this FLIP does not yet expose the constraint via DDL syntax, but if we silently omit the new constraint from the output, it could break DDL round-trip fidelity, WDYT?

[lincoln-lil]

The description here may be misleading regarding its relationship with the PK. The JavaDoc for Schema.Builder.immutableColumns is clearer.

[xuyangzhong]

@xuyangzhong Thanks for the pr! Overall looks good. I left some comments there.

One question regarding the SHOW CREATE TABLE path: I understand this FLIP does not yet expose the constraint via DDL syntax, but if we silently omit the new constraint from the output, it could break DDL round-trip fidelity, WDYT?

Currently, asSummaryString in Constraint only works for toString. The schema printing for "SHOW CREATE TABLE" is handled in a different place: ShowCreateUtil.buildShowCreateTableRow. Regardless, I will add a test in ShowCreateUtilTest to prevent the immutable columns constraint from being included in the SHOW CREATE TABLE output.

[xuyangzhong]

Hi, @lincoln-lil I have addressed all comments and added a new commit about https://issues.apache.org/jira/browse/FLINK-39314 in this pr. Can you please take a look again?

[lincoln-lil]

Hi, @lincoln-lil I have addressed all comments and added a new commit about https://issues.apache.org/jira/browse/FLINK-39314 in this pr. Can you please take a look again?

I checked these changes, looks semantically correct and addresses a real non-opt issue where only the first upsert key was checked instead of all available upsert keys. The existing tests provide partial coverage, so additional tests for the multi-upsertkey scenario would be beneficial(not strictly required).

[xuyangzhong]

Thanks for reminder. I have added more tests for the commit [table-planner] Allow input to drop update before if filter is applied on any of upsert keys