twalthr opened a new pull request, #29525: URL: https://github.com/apache/flink/pull/29525
## What is the purpose of the change This implements `ArgumentTrait.NOTIFY_STATEFUL_SETS` for process table functions (PTFs) as proposed in [FLIP-565](https://cwiki.apache.org/confluence/spaces/FLINK/pages/406620840/FLIP-565+Improve+ProcessTableFunctions+for+late+data+handling+and+state+access). It builds on the broadcast state support of FLINK-39883. For every row of a broadcast table with this trait, `eval()` is called multiple times: first without scoping to a set, to update broadcast state (broadcast phase); next once per set that holds state on the virtual processor (notification phase). During notification, the state of the set is accessible, broadcast state is read-only, and the PTF can emit results whose output is prefixed with the set's key and carries the rowtime of the broadcast row. This enables reacting to dynamic rules or configuration for sets that have been seen before, e.g. resetting or removing them. ## Brief change log - `flink-table-common` - New `ArgumentTrait`/`StaticArgumentTrait.NOTIFY_STATEFUL_SETS`, only valid next to `BROADCAST_SEMANTIC_TABLE` - `SystemTypeInference` validates that at least one state entry scoped to a set is declared - `flink-table-planner` - `StreamExecProcessTableFunction` passes the trait to the runtime via `RuntimeTableSemantics` - `flink-table-runtime` - After the broadcast phase, `ProcessSetTableOperator` iterates all keys of the per-set state entries via `KeyedStateBackend#getKeys`, sets the key context, and calls `eval()` again. Keys are materialized upfront because state backends don't support modifications while iterating keys. - `ProcessTableRunner#ingestBroadcastNotification` derives the output prefix from the key and sets the output rowtime - Read-only error message of broadcast state entries now also covers the notification phase - JavaDoc in `ArgumentTrait` and `ProcessTableFunction`; a new "Notifying Stateful Sets" section and a "Blockable Score Aggregation" advanced example in `ptfs.md` (en and zh). The outdated "Broadcast state" limitation has been removed. ## Verifying this change This change added tests and can be verified as follows: - `TypeInferenceExtractorTest`: trait without a broadcast table is rejected - `ProcessTableFunctionTest`: notification without state scoped to a set is rejected - `ProcessTableFunctionSemanticTests`: rules are applied to new sets or notify existing sets. Named timers act as watermark barriers so that the output is independent of the order in which main and broadcast tables are processed. The function asserts that broadcast state is read-only during notification. - `ProcessTableFunctionRestoreTests`: a set restored from a savepoint is notified by a broadcast row after restore; the emitted row carries the rowtime of the broadcast row - `ProcessSetTableOperatorTest` (new harness test): notification with deterministic ordering, clearing state during notification, timers, read-only broadcast state, and no notification without the trait ## 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 (`ArgumentTrait`, `StaticArgumentTrait`, as proposed in FLIP-565) - The serializers: no - The runtime per-record code paths (performance sensitive): yes (each broadcast row of a notifying table iterates all keys of the virtual processor, which is documented as expensive; code paths without the trait remain unchanged) - 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 --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes (please specify the tool below) Generated-by: Claude Code (Claude Opus 5.5) 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
