[FLINK-39261][table] Add FROM_CHANGELOG built-in process table function

[raminqaf]

What is the purpose of the change

Implement the FROM_CHANGELOG built-in process table function as specified in FLIP-564 (append-only stream to retract stream, flat mode).

FROM_CHANGELOG converts an append-only table with an explicit operation code column into a (potentially updating) dynamic table. Each input record is mapped to one output record with the appropriate RowKind. The operation column is interpreted by the engine and removed from the output. The implementation is stateless and uses row semantics.

SELECT * FROM FROM_CHANGELOG(
    input => TABLE cdc_stream,
    op => DESCRIPTOR(op),
    op_mapping => MAP['c, r', 'INSERT', 'ub', 'UPDATE_BEFORE', 'ua', 'UPDATE_AFTER', 'd', 'DELETE']
)

Brief change log

• Add changelogModeResolver to BuiltInFunctionDefinition so built-in PTFs can declare their output changelog mode without implementing ChangelogFunction
• Update FlinkChangelogModeInferenceProgram to check for the resolver on built-in PTFs
• Allow retract mode (with UPDATE_BEFORE) for row-semantic PTFs in verifyPtfTableArgsForUpdates
• Add FromChangelogTypeStrategy with input validation (op column existence, STRING type, op_mapping value validation, UPDATE_BEFORE required) and output type inference (removes op column from output)
• Introduce ValidationOnlyInputTypeStrategy base class for PTFs that use static arguments
• Add FROM_CHANGELOG built-in function definition with ROW_SEMANTIC_TABLE and ChangelogMode.all() output
• Add FromChangelogFunction runtime implementation using ProjectedRowData for zero-copy projection
• Add fromChangelog() convenience method to Table API
• Add ChangelogContext.getArgumentValue() for inspecting function arguments during changelog inference
• Add documentation for FROM_CHANGELOG in changelog.md

Verifying this change

This change added tests and can be verified as follows:

• Added unit tests for FromChangelogTypeStrategy input validation (valid mapping, op column not found, wrong type, invalid descriptor, invalid RowKind, UPDATE_BEFORE required, duplicate RowKind, upsert mapping rejected)
• Added semantic tests covering: default op_mapping with all four change operations, custom op_mapping with comma-separated keys, unmapped codes dropped, custom op column name, Table API convenience method
• Added plan tests verifying changelog mode propagation (changelogMode=[I,UB,UA,D] for retract output, changelogMode=[I] for source input) and custom op_mapping
• Added plan test for retract mode with row semantics in ProcessTableFunctionTest
• Updated ProcessTableFunctionTest to verify upsert mode with row semantics is rejected

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 (Table.fromChangelog(), ChangelogContext.getArgumentValue())
• 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? docs / JavaDocs

[flinkbot]

CI report:

• 917d73a Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[gustavodemorais]

Thanks for the PR, @raminqaf! In general, it looks very consistent with to_changelog.

Here is my first set of reviews. In this one, I think there are some minor details we have to decide on and two bigger things:

• Going with retract by default as the output mode
• Finding a nice way of expressing the changelog mode with the BuiltInFunctionsDefinition

See below for more details

[gustavodemorais]

Thanks for the updates, Ramin! Added some comments, take a look

[twalthr]

Some initial comments.

[gustavodemorais]

Thanks for all the updates and the replies, @raminqaf. I think we clarified most of the major things and we're almost there!

[gustavodemorais]

Looks great, @raminqaf! Thanks for all the updates. LGTM! We can continue interacting in the next PRs.

[twalthr]

Excellent PR @raminqaf. I just had a couple of minor improvement comments. Should be good to merge in the next iteration.

[twalthr]

LGTM. Thanks @raminqaf.

[gustavodemorais]

Hey @raminqaf, I'd suggest doing changing this in the next PR so we close this and we can adjust it for both to and from changelog 🙂