You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "twalthr (via GitHub)" <gi...@apache.org> on 2026/10/07 13:09:28 UTC
[PR] [FLINK-39883][table] Support broadcast state for process table functions [flink]
twalthr opened a new pull request, #29419:
URL: https://github.com/apache/flink/pull/29419
## What is the purpose of the change
This implements broadcast state for process table functions (PTFs) as proposed in [FLIP-565](https://cwiki.apache.org/confluence/display/FLINK/FLIP-565%3A+Broadcast+State+for+Process+Table+Functions).
A table argument with `ArgumentTrait.BROADCAST_SEMANTIC_TABLE` is sent to all virtual processors next to tables with row or set semantics. State entries declared with `@StateHint(StateKind.BROADCAST)` are backed by Flink's broadcast state, shared across all sets, and can only be modified while processing a broadcast table. This enables the broadcast state pattern (e.g. dynamic rules or configuration) in SQL and the Table API.
`NOTIFY_STATEFUL_SETS` is out of scope and will be addressed separately. Support in `ProcessTableFunctionTestHarness` will follow in a separate PR.
## Brief change log
- `flink-table-common`
- New `StateKind { PER_SET, BROADCAST }` referenced by `StateHint#value()`
- New `ArgumentTrait`/`StaticArgumentTrait.BROADCAST_SEMANTIC_TABLE`
- `StateTypeStrategy` carries a broadcast flag; extraction rejects `ListView` and TTL for broadcast state
- `SystemTypeInference` validates that at least one table with row or set semantics exists and rejects `PARTITION BY`/`ORDER BY` on broadcast tables. Similar to tables with set semantics, broadcast tables participate in event-time processing and require `on_time` if declared.
- `flink-table-planner`
- Broadcast tables are distributed via `BROADCAST_DISTRIBUTED` (`BroadcastPartitioner` in `StreamExecExchange`)
- `StreamExecProcessTableFunction` creates a keyed or non-keyed multi-input transformation and passes broadcast information to the runtime
- Code generation passes `null` for keyed state while processing a broadcast table
- `flink-table-runtime`
- `BroadcastStateAdapters` expose operator broadcast state as `MapView`, `ValueView`, and eager value state; writes are rejected while processing tables with row or set semantics
- `BroadcastEvalCollector` and `BroadcastInternalTimeContext` reject emitting results and registering timers on the broadcast path
- `Context#clearState`/`clearAllState` are scoped to the currently processed table
- JavaDoc in `ProcessTableFunction`, `StateHint`, `StateKind`, `ArgumentTrait` and a new "Broadcast State" section in `ptfs.md` (en and zh)
## Verifying this change
This change added tests and can be verified as follows:
- `TypeInferenceExtractorTest`: extraction of broadcast state and table arguments, including invalid declarations (ListView, TTL, missing main table, incompatible traits)
- `ProcessTableFunctionTest`: plan tests for set/row semantic main tables, multiple tables, `on_time`, updating inputs, and validation errors
- `ProcessTableFunctionSemanticTests`: broadcast `MapView`, `ValueView`, and eager state with SQL and Table API, row semantic main tables, multiple broadcast tables, `on_time`, `clearAllState()`, as well as runtime errors for `collect()`, timers, and writes from the main path. End-of-input timers verify the final broadcast state content deterministically.
- `ProcessTableFunctionRestoreTests`: restore of broadcast state from a savepoint
- `BroadcastStateAdaptersTest`: read-only behavior of the state adapters
## 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 (`StateKind`, `StateHint`, `ArgumentTrait`, `StaticArgumentTrait`, `StateTypeStrategy`, as proposed in FLIP-565)
- The serializers: no
- The runtime per-record code paths (performance sensitive): yes (PTF runner and operators; the code paths of PTFs without broadcast tables remain unchanged)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no (PTF operators use operator broadcast state which is checkpointed as usual)
- 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)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39883][table] Support broadcast state for process table functions [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #29419:
URL: https://github.com/apache/flink/pull/29419#discussion_r4220456132
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalProcessTableFunction.java:
##########
@@ -371,6 +371,46 @@ private static void verifyInputSize(TableConfig tableConfig, int providedInputAr
// Shared utilities
// --------------------------------------------------------------------------------------------
+ /**
+ * Derives per-input {@link InputProperty}s. Broadcast table arguments require a broadcast
+ * distribution, all other inputs use the default.
+ */
+ private List<InputProperty> deriveInputProperties(RexCall call) {
+ final boolean[] broadcastByInput = broadcastByInput(call, getInputs().size());
+ return IntStream.range(0, getInputs().size())
+ .mapToObj(
+ i ->
+ broadcastByInput[i]
+ ? InputProperty.builder()
+ .requiredDistribution(
+ InputProperty.BROADCAST_DISTRIBUTION)
+ .build()
+ : InputProperty.DEFAULT)
+ .collect(Collectors.toList());
+ }
+
+ /** Returns whether the given signature declares a table argument with broadcast semantics. */
+ public static boolean hasBroadcastTables(List<StaticArgument> staticArgs) {
+ return staticArgs.stream()
+ .anyMatch(arg -> arg.is(StaticArgumentTrait.BROADCAST_SEMANTIC_TABLE));
+ }
+
+ /**
+ * Returns a flag per input indicating whether it is a table argument with broadcast semantics.
+ */
+ public static boolean[] broadcastByInput(RexCall call, int inputCount) {
Review Comment:
I feel this is a micro optimization with little impact. Given that we limit inputs to 20. Boolean is easier for debugging.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39883][table] Support broadcast state for process table functions [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #29419:
URL: https://github.com/apache/flink/pull/29419#discussion_r4210195672
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/ProcessTableFunction.java:
##########
@@ -378,6 +389,63 @@
* }
* }</pre>
*
+ * <h2>Broadcast State</h2>
+ *
+ * <p>Broadcast state enables patterns such as a rule engine where rules are dynamically updated at
+ * runtime, or dynamic configuration that influences the processing of the main table(s).
+ *
+ * <p>Broadcast state is tightly coupled to a broadcast semantic table. A table argument declared
+ * with {@link ArgumentTrait#BROADCAST_SEMANTIC_TABLE} acts as a "side" or "control" input. Every
+ * row of a broadcast table is sent to all virtual processors, regardless of any PARTITION BY
+ * clause. A PTF can store this broadcast information in state entries declared with
+ * {@code @StateHint(StateKind.BROADCAST)}. This broadcast state is shared across all sets and can
+ * be read when processing rows from the main table(s).
+ *
+ * <p>The following rules apply to broadcast tables and broadcast state:
+ *
+ * <ul>
+ * <li>At least one table argument with row or set semantics must be declared next to broadcast
+ * tables. Multiple broadcast tables and multiple broadcast state entries are supported.
+ * <li>Broadcast state can only be modified while processing a broadcast row. This includes
+ * clearing broadcast state. When processing rows of the main table(s), broadcast state is
+ * read-only.
+ * <li>While processing a broadcast row, there is no key context. State entries that are scoped to
+ * a set are passed as null, results can not be emitted via {@code collect()}, and timers
+ * cannot be registered or cleared. {@link Context#clearAllState()} only clears broadcast
+ * state in this case.
+ * <li>It is the responsibility of the PTF implementer to maintain identical broadcast state
+ * across all virtual processors, i.e. broadcast state should only be updated
+ * deterministically based on the broadcast rows.
+ * </ul>
+ *
+ * <p>Note: The system decides which input row is streamed through the virtual processor next. A
+ * change to a broadcast state entry has no effect on rows of the main table(s) that have been
+ * processed before.
+ *
+ * <pre>{@code
+ * // Function that filters sentences using a dynamically updated list of bad words
+ * class RuleFunction extends ProcessTableFunction<String> {
+ * public void eval(
+ * @StateHint(StateKind.BROADCAST) MapView<String, Boolean> badWords,
+ * @ArgumentHint(ROW_SEMANTIC_TABLE) Row data,
+ * @ArgumentHint(BROADCAST_SEMANTIC_TABLE) Row rules) throws Exception {
+ * // Write access to broadcast state for the broadcast table
+ * if (rules != null) {
+ * badWords.put(rules.getFieldAs("word"), true);
+ * return;
+ * }
+ * // Read access to broadcast state for the main table
+ * String sentence = data.getFieldAs("sentence");
+ * for (String word : sentence.split(" ")) {
Review Comment:
the example in docs also uses `toLowerCase()`
is it intentional?
https://github.com/apache/flink/pull/29419/changes#diff-3454533d1ebf67d5630b4f39257c7ec9b5b172f12f7191cb26a35644a66ce55bR1434
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39883][table] Support broadcast state for process table functions [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #29419:
URL: https://github.com/apache/flink/pull/29419#discussion_r4210114403
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ProcessTableFunctionTest.java:
##########
@@ -696,6 +791,37 @@ public void eval(@ArgumentHint({ROW_SEMANTIC_TABLE, SUPPORT_UPDATES}) User u, In
}
/** Testing function. */
+ /** Testing function. */
Review Comment:
I guess one should be enough
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39883][table] Support broadcast state for process table functions [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr merged PR #29419:
URL: https://github.com/apache/flink/pull/29419
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39883][table] Support broadcast state for process table functions [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #29419:
URL: https://github.com/apache/flink/pull/29419#discussion_r4210432031
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalProcessTableFunction.java:
##########
@@ -371,6 +371,46 @@ private static void verifyInputSize(TableConfig tableConfig, int providedInputAr
// Shared utilities
// --------------------------------------------------------------------------------------------
+ /**
+ * Derives per-input {@link InputProperty}s. Broadcast table arguments require a broadcast
+ * distribution, all other inputs use the default.
+ */
+ private List<InputProperty> deriveInputProperties(RexCall call) {
+ final boolean[] broadcastByInput = broadcastByInput(call, getInputs().size());
+ return IntStream.range(0, getInputs().size())
+ .mapToObj(
+ i ->
+ broadcastByInput[i]
+ ? InputProperty.builder()
+ .requiredDistribution(
+ InputProperty.BROADCAST_DISTRIBUTION)
+ .build()
+ : InputProperty.DEFAULT)
+ .collect(Collectors.toList());
+ }
+
+ /** Returns whether the given signature declares a table argument with broadcast semantics. */
+ public static boolean hasBroadcastTables(List<StaticArgument> staticArgs) {
+ return staticArgs.stream()
+ .anyMatch(arg -> arg.is(StaticArgumentTrait.BROADCAST_SEMANTIC_TABLE));
+ }
+
+ /**
+ * Returns a flag per input indicating whether it is a table argument with broadcast semantics.
+ */
+ public static boolean[] broadcastByInput(RexCall call, int inputCount) {
Review Comment:
Since most of the code is generated I wonder if we can use BitSets which are more compact
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39883][table] Support broadcast state for process table functions [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #29419:
URL: https://github.com/apache/flink/pull/29419#issuecomment-6038952430
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "94848a2627c6b488d91c4a2d70afa4c9ff7923e7",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "94848a2627c6b488d91c4a2d70afa4c9ff7923e7",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 94848a2627c6b488d91c4a2d70afa4c9ff7923e7 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org