This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 77c3daea10d [FLINK-40961][table] Support notifying stateful sets for
broadcast tables in PTFs
77c3daea10d is described below
commit 77c3daea10dc224d3e84916d2cf77130f5e28fe2
Author: Timo Walther <[email protected]>
AuthorDate: Tue Oct 20 16:07:07 2026 +0200
[FLINK-40961][table] Support notifying stateful sets for broadcast tables
in PTFs
This closes #29525.
---
docs/content.zh/docs/dev/table/functions/ptfs.md | 199 ++++++++-
docs/content/docs/dev/table/functions/ptfs.md | 199 ++++++++-
.../flink/table/annotation/ArgumentTrait.java | 42 +-
.../table/functions/ProcessTableFunction.java | 49 ++-
.../table/types/inference/StaticArgumentTrait.java | 3 +
.../table/types/inference/SystemTypeInference.java | 38 +-
.../extraction/TypeInferenceExtractorTest.java | 17 +
.../stream/StreamExecProcessTableFunction.java | 1 +
.../stream/ProcessTableFunctionRestoreTests.java | 7 +-
.../stream/ProcessTableFunctionSemanticTests.java | 1 +
.../stream/ProcessTableFunctionTestPrograms.java | 72 ++-
.../exec/stream/ProcessTableFunctionTestUtils.java | 118 +++++
.../plan/stream/sql/ProcessTableFunctionTest.java | 17 +
.../plan/process-notify-stateful-sets-restore.json | 482 +++++++++++++++++++++
.../savepoint/_metadata | Bin 0 -> 17281 bytes
.../runtime/generated/ProcessTableRunner.java | 71 ++-
.../process/AbstractProcessTableOperator.java | 35 +-
.../operators/process/BroadcastEvalCollector.java | 5 +
.../operators/process/BroadcastStateAdapters.java | 5 +-
.../process/PassThroughCollectorBase.java | 5 +
.../operators/process/ProcessSetTableOperator.java | 6 +-
.../operators/process/RuntimeStateInfo.java | 15 +
.../operators/process/RuntimeTableSemantics.java | 7 +
...essSetTableOperatorInterruptibleTimersTest.java | 1 +
.../process/ProcessSetTableOperatorTest.java | 424 ++++++++++++++++++
25 files changed, 1770 insertions(+), 49 deletions(-)
diff --git a/docs/content.zh/docs/dev/table/functions/ptfs.md
b/docs/content.zh/docs/dev/table/functions/ptfs.md
index 39600e8132c..ee73302776f 100644
--- a/docs/content.zh/docs/dev/table/functions/ptfs.md
+++ b/docs/content.zh/docs/dev/table/functions/ptfs.md
@@ -1471,6 +1471,91 @@ The following rules apply to broadcast tables and
broadcast state:
- 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.
+#### Notifying Stateful Sets
+
+By default, a change to a broadcast state entry has no effect on sets that
have been processed before. Especially when
+building a rule engine, the question arises how to re-evaluate previously
processed sets with the new broadcast
+information. For this, a broadcast table can be declared with
`ArgumentTrait.NOTIFY_STATEFUL_SETS`.
+
+For every row of such a broadcast table, `eval()` is called multiple times:
+
+1. First, without scoping to a set (as described above): This call updates the
broadcast state, if available.
+2. Next, once for every set that holds at least one state entry: In this
phase, the `eval()` implementation
+ has access to the incoming broadcast row, as well as all (keyed) state
entries.
+
+The following example deduplicates events per key. A row of the broadcast
table resets the deduplication for all keys.
+
+{{< tabs "5ddc43b1-d719-4bb6-be26-4aed9476e5d3" >}}
+{{< tab "Java" >}}
+```java
+TableEnvironment env =
TableEnvironment.create(EnvironmentSettings.inStreamingMode());
+
+env.executeSql("CREATE VIEW Events(id, payload) AS VALUES (1, 'a'), (1, 'a'),
(2, 'b')");
+env.executeSql("CREATE VIEW Resets(reason) AS VALUES ('cache invalidation')");
+
+env.createFunction("ControllableDedup", ControllableDedup.class);
+
+env
+ .executeSql("SELECT * FROM ControllableDedup(data => TABLE Events PARTITION
BY id, resets => TABLE Resets)")
+ .print();
+
+// --------------------
+// Function declaration
+// --------------------
+
+// Function that deduplicates events per key until a reset is broadcast
+public static class ControllableDedup extends ProcessTableFunction<String> {
+
+ public void eval(
+ Context ctx,
+ @StateHint ValueView<String> seen,
+ @ArgumentHint(SET_SEMANTIC_TABLE) Row data,
+ @ArgumentHint({BROADCAST_SEMANTIC_TABLE, NOTIFY_STATEFUL_SETS}) Row
resets
+ ) throws Exception {
+ // Process row from table 'Events': emit only the first payload per key
+ if (data != null) {
+ if (seen.getValue() == null) {
+ String payload = data.getFieldAs("payload");
+ seen.setValue(payload);
+ collect(payload);
+ }
+ return;
+ }
+
+ // Process row from table 'Resets' - broadcast phase:
+ // state scoped to a set is null, nothing to do
+ if (resets != null && seen == null) {
+ return;
+ }
+
+ // Process row from table 'Resets' - notification phase:
+ // state scoped to a set is present, act on notification
+ if (resets != null && seen != null) {
+ ctx.clearAllState();
+ }
+ }
+}
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+The following rules apply to notifying stateful sets:
+
+- At least one state entry scoped to a set must be declared. Sets without
state entries are not notified. Registered timers alone don't count as state.
+- Sets whose state entries have expired due to [state TTL](#state-ttl) might
still be notified until the state backend
+ has cleaned them up. In this case, the state entries of the set are empty,
and writing to them during notification
+ keeps the set alive.
+- During notification, state entries of the set and timers are accessible and
broadcast state is read-only. Similar to processing a row
+ of the main table(s), the PTF can emit results via `collect()` and register
timers. Results contain the partition key
+ of the set.
+
+{{< hint warning >}}
+Notifying stateful sets is expensive. For every broadcast row, all keys in
state are loaded and a PTF's eval() is called for
+each of them. Processing of other rows and checkpointing is blocked until all
sets have been notified. Thus, this trait
+should only be used if broadcast rows are rare compared to rows of the main
table(s) and the overall key space is
+relatively low.
+{{< /hint >}}
+
### Efficiency and Design Principles
A high number of input tables can negatively impact a single TaskManager or
subtask. Network buffers must be allocated
@@ -2224,12 +2309,124 @@ for `Charly` could not be found.
+----+-------------+-------------+--------------------------------+--------------------------------+
```
+{{< top >}}
+
+### Blockable Score Aggregation
+
+The following example combines various PTF concepts for influencing the
aggregation of scores:
+- a table with set semantics,
+- broadcast table with broadcast state that notifies previous stateful sets,
+- and an updating output.
+
+The PTF sums up scores per user and emits the current sum as an upsert. A
broadcast table contains blocked
+users that are stored in broadcast state. Score events of blocked users are
ignored. If a user is blocked after being
+processed, the notification removes the user's state and sends a
`RowKind.DELETE` downstream.
+
+{{< tabs "16dccd99-2d64-415b-9f5e-9683e30e9300" >}}
+{{< tab "Java" >}}
+```java
+TableEnvironment env =
TableEnvironment.create(EnvironmentSettings.inStreamingMode());
+
+env.executeSql("CREATE VIEW ScoreEvents(name, score) AS VALUES ('Bob', 12),
('Alice', 42), ('Bob', 7), ('Alice', 3)");
+env.executeSql("CREATE VIEW BlockedUsers(name) AS VALUES ('Bob')");
+
+// The PTF emits upserts, thus, the sink declares the PARTITION BY key as
primary key
+env.executeSql(
+ "CREATE TABLE UserScores (name STRING, total INT, PRIMARY KEY (name) NOT
ENFORCED) WITH ("
+ + "'connector' = 'upsert-kafka', "
+ + "'topic' = 'user-scores', "
+ + "'properties.bootstrap.servers' = 'localhost:9092', "
+ + "'key.format' = 'json', "
+ + "'value.format' = 'json')");
+
+env.createFunction("BlockableScoreSum", BlockableScoreSum.class);
+
+env.executeSql(
+ "INSERT INTO UserScores SELECT * FROM BlockableScoreSum("
+ + "scores => TABLE ScoreEvents PARTITION BY name, "
+ + "blockedUsers => TABLE BlockedUsers)");
+
+// --------------------
+// Function declaration
+// --------------------
+
+// Function that sums up and continuously updates scores per user.
+// It removes blocked users from the result when instructed to do so from a
+// controlling BlockedUsers table.
+@DataTypeHint("ROW<total INT>")
+public static class BlockableScoreSum
+ extends ProcessTableFunction<Row>
+ implements ChangelogFunction {
+
+ @Override
+ public ChangelogMode getChangelogMode(ChangelogContext changelogContext) {
+ // Tells the system that the PTF produces upserts with deletes that
contain all fields
+ return ChangelogMode.upsert(false);
+ }
+
+ public static class UserScore {
+ public String name;
+ public int total;
+ }
+
+ public void eval(
+ Context ctx,
+ @StateHint UserScore user,
+ @StateHint(StateKind.BROADCAST) MapView<String, Boolean> blockList,
+ @ArgumentHint(SET_SEMANTIC_TABLE) Row scores,
+ @ArgumentHint({BROADCAST_SEMANTIC_TABLE, NOTIFY_STATEFUL_SETS}) Row
blockedUsers
+ ) throws Exception {
+ // Process row from table 'ScoreEvents': ignore blocked users, otherwise
emit the current sum
+ if (scores != null) {
+ String name = scores.getFieldAs("name");
+ if (blockList.contains(name)) {
+ return;
+ }
+ RowKind kind = user.name == null ? RowKind.INSERT : RowKind.UPDATE_AFTER;
+ user.name = name;
+ user.total += scores.<Integer>getFieldAs("score");
+ collect(Row.ofKind(kind, user.total));
+ return;
+ }
+
+ String blockedName = blockedUsers.getFieldAs("name");
+
+ // Process row from table 'BlockedUsers' - broadcast phase:
+ // state scoped to a set is null, store the blocked user
+ if (user == null) {
+ blockList.put(blockedName, true);
+ return;
+ }
+
+ // Process row from table 'BlockedUsers' - notification phase:
+ // state scoped to a set is present, remove the blocked user from the
result and from state
+ if (blockedName.equals(user.name)) {
+ collect(Row.ofKind(RowKind.DELETE, user.total));
+ ctx.clearAllState();
+ }
+ }
+}
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+If `Bob` is blocked after his first score event has been processed, the sink
receives the following changelog:
+
+```text
++I[Bob, 12]
+-D[Bob, 12]
++I[Alice, 42]
++U[Alice, 45]
+```
+
+If the broadcast row is processed before any score event of `Bob`, `Bob` never
appears in the changelog. In both cases,
+the materialized result only contains `Alice` with a total of 45.
+
Limitations
-----------
PTFs are in an early stage. The following limitations apply:
- PTFs cannot run in batch mode.
-- Broadcast state
Testing Process Table Functions
-------------------------------
diff --git a/docs/content/docs/dev/table/functions/ptfs.md
b/docs/content/docs/dev/table/functions/ptfs.md
index 9210292c6dd..63d47252401 100644
--- a/docs/content/docs/dev/table/functions/ptfs.md
+++ b/docs/content/docs/dev/table/functions/ptfs.md
@@ -1474,6 +1474,91 @@ The following rules apply to broadcast tables and
broadcast state:
- 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.
+#### Notifying Stateful Sets
+
+By default, a change to a broadcast state entry has no effect on sets that
have been processed before. Especially when
+building a rule engine, the question arises how to re-evaluate previously
processed sets with the new broadcast
+information. For this, a broadcast table can be declared with
`ArgumentTrait.NOTIFY_STATEFUL_SETS`.
+
+For every row of such a broadcast table, `eval()` is called multiple times:
+
+1. First, without scoping to a set (as described above): This call updates the
broadcast state, if available.
+2. Next, once for every set that holds at least one state entry: In this
phase, the `eval()` implementation
+ has access to the incoming broadcast row, as well as all (keyed) state
entries.
+
+The following example deduplicates events per key. A row of the broadcast
table resets the deduplication for all keys.
+
+{{< tabs "5ddc43b1-d719-4bb6-be26-4aed9476e5d3" >}}
+{{< tab "Java" >}}
+```java
+TableEnvironment env =
TableEnvironment.create(EnvironmentSettings.inStreamingMode());
+
+env.executeSql("CREATE VIEW Events(id, payload) AS VALUES (1, 'a'), (1, 'a'),
(2, 'b')");
+env.executeSql("CREATE VIEW Resets(reason) AS VALUES ('cache invalidation')");
+
+env.createFunction("ControllableDedup", ControllableDedup.class);
+
+env
+ .executeSql("SELECT * FROM ControllableDedup(data => TABLE Events PARTITION
BY id, resets => TABLE Resets)")
+ .print();
+
+// --------------------
+// Function declaration
+// --------------------
+
+// Function that deduplicates events per key until a reset is broadcast
+public static class ControllableDedup extends ProcessTableFunction<String> {
+
+ public void eval(
+ Context ctx,
+ @StateHint ValueView<String> seen,
+ @ArgumentHint(SET_SEMANTIC_TABLE) Row data,
+ @ArgumentHint({BROADCAST_SEMANTIC_TABLE, NOTIFY_STATEFUL_SETS}) Row
resets
+ ) throws Exception {
+ // Process row from table 'Events': emit only the first payload per key
+ if (data != null) {
+ if (seen.getValue() == null) {
+ String payload = data.getFieldAs("payload");
+ seen.setValue(payload);
+ collect(payload);
+ }
+ return;
+ }
+
+ // Process row from table 'Resets' - broadcast phase:
+ // state scoped to a set is null, nothing to do
+ if (resets != null && seen == null) {
+ return;
+ }
+
+ // Process row from table 'Resets' - notification phase:
+ // state scoped to a set is present, act on notification
+ if (resets != null && seen != null) {
+ ctx.clearAllState();
+ }
+ }
+}
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+The following rules apply to notifying stateful sets:
+
+- At least one state entry scoped to a set must be declared. Sets without
state entries are not notified. Registered timers alone don't count as state.
+- Sets whose state entries have expired due to [state TTL](#state-ttl) might
still be notified until the state backend
+ has cleaned them up. In this case, the state entries of the set are empty,
and writing to them during notification
+ keeps the set alive.
+- During notification, state entries of the set and timers are accessible and
broadcast state is read-only. Similar to processing a row
+ of the main table(s), the PTF can emit results via `collect()` and register
timers. Results contain the partition key
+ of the set.
+
+{{< hint warning >}}
+Notifying stateful sets is expensive. For every broadcast row, all keys in
state are loaded and a PTF's eval() is called for
+each of them. Processing of other rows and checkpointing is blocked until all
sets have been notified. Thus, this trait
+should only be used if broadcast rows are rare compared to rows of the main
table(s) and the overall key space is
+relatively low.
+{{< /hint >}}
+
### Efficiency and Design Principles
A high number of input tables can negatively impact a single TaskManager or
subtask. Network buffers must be allocated
@@ -2227,12 +2312,124 @@ for `Charly` could not be found.
+----+-------------+-------------+--------------------------------+--------------------------------+
```
+{{< top >}}
+
+### Blockable Score Aggregation
+
+The following example combines various PTF concepts for influencing the
aggregation of scores:
+- a table with set semantics,
+- broadcast table with broadcast state that notifies previous stateful sets,
+- and an updating output.
+
+The PTF sums up scores per user and emits the current sum as an upsert. A
broadcast table contains blocked
+users that are stored in broadcast state. Score events of blocked users are
ignored. If a user is blocked after being
+processed, the notification removes the user's state and sends a
`RowKind.DELETE` downstream.
+
+{{< tabs "16dccd99-2d64-415b-9f5e-9683e30e9300" >}}
+{{< tab "Java" >}}
+```java
+TableEnvironment env =
TableEnvironment.create(EnvironmentSettings.inStreamingMode());
+
+env.executeSql("CREATE VIEW ScoreEvents(name, score) AS VALUES ('Bob', 12),
('Alice', 42), ('Bob', 7), ('Alice', 3)");
+env.executeSql("CREATE VIEW BlockedUsers(name) AS VALUES ('Bob')");
+
+// The PTF emits upserts, thus, the sink declares the PARTITION BY key as
primary key
+env.executeSql(
+ "CREATE TABLE UserScores (name STRING, total INT, PRIMARY KEY (name) NOT
ENFORCED) WITH ("
+ + "'connector' = 'upsert-kafka', "
+ + "'topic' = 'user-scores', "
+ + "'properties.bootstrap.servers' = 'localhost:9092', "
+ + "'key.format' = 'json', "
+ + "'value.format' = 'json')");
+
+env.createFunction("BlockableScoreSum", BlockableScoreSum.class);
+
+env.executeSql(
+ "INSERT INTO UserScores SELECT * FROM BlockableScoreSum("
+ + "scores => TABLE ScoreEvents PARTITION BY name, "
+ + "blockedUsers => TABLE BlockedUsers)");
+
+// --------------------
+// Function declaration
+// --------------------
+
+// Function that sums up and continuously updates scores per user.
+// It removes blocked users from the result when instructed to do so from a
+// controlling BlockedUsers table.
+@DataTypeHint("ROW<total INT>")
+public static class BlockableScoreSum
+ extends ProcessTableFunction<Row>
+ implements ChangelogFunction {
+
+ @Override
+ public ChangelogMode getChangelogMode(ChangelogContext changelogContext) {
+ // Tells the system that the PTF produces upserts with deletes that
contain all fields
+ return ChangelogMode.upsert(false);
+ }
+
+ public static class UserScore {
+ public String name;
+ public int total;
+ }
+
+ public void eval(
+ Context ctx,
+ @StateHint UserScore user,
+ @StateHint(StateKind.BROADCAST) MapView<String, Boolean> blockList,
+ @ArgumentHint(SET_SEMANTIC_TABLE) Row scores,
+ @ArgumentHint({BROADCAST_SEMANTIC_TABLE, NOTIFY_STATEFUL_SETS}) Row
blockedUsers
+ ) throws Exception {
+ // Process row from table 'ScoreEvents': ignore blocked users, otherwise
emit the current sum
+ if (scores != null) {
+ String name = scores.getFieldAs("name");
+ if (blockList.contains(name)) {
+ return;
+ }
+ RowKind kind = user.name == null ? RowKind.INSERT : RowKind.UPDATE_AFTER;
+ user.name = name;
+ user.total += scores.<Integer>getFieldAs("score");
+ collect(Row.ofKind(kind, user.total));
+ return;
+ }
+
+ String blockedName = blockedUsers.getFieldAs("name");
+
+ // Process row from table 'BlockedUsers' - broadcast phase:
+ // state scoped to a set is null, store the blocked user
+ if (user == null) {
+ blockList.put(blockedName, true);
+ return;
+ }
+
+ // Process row from table 'BlockedUsers' - notification phase:
+ // state scoped to a set is present, remove the blocked user from the
result and from state
+ if (blockedName.equals(user.name)) {
+ collect(Row.ofKind(RowKind.DELETE, user.total));
+ ctx.clearAllState();
+ }
+ }
+}
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+If `Bob` is blocked after his first score event has been processed, the sink
receives the following changelog:
+
+```text
++I[Bob, 12]
+-D[Bob, 12]
++I[Alice, 42]
++U[Alice, 45]
+```
+
+If the broadcast row is processed before any score event of `Bob`, `Bob` never
appears in the changelog. In both cases,
+the materialized result only contains `Alice` with a total of 45.
+
Limitations
-----------
PTFs are in an early stage. The following limitations apply:
- PTFs cannot run in batch mode.
-- Broadcast state
Testing Process Table Functions
-------------------------------
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/ArgumentTrait.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/ArgumentTrait.java
index dd94778624f..b9641e60fc9 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/ArgumentTrait.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/ArgumentTrait.java
@@ -93,8 +93,10 @@ public enum ArgumentTrait {
* state entries. State entries that are scoped to a set are passed as
null. The PTF must not
* emit results via {@code collect()} or register/clear timers.
*
- * <p>A change to a broadcast state entry has no effect on rows of the
main table(s) that have
- * been processed before. Rows of different tables are not processed in a
deterministic order.
+ * <p>By default, a change to a broadcast state entry has no effect on
rows of the main table(s)
+ * that have been processed before. Rows of different tables are not
processed in a
+ * deterministic order. Use {@link #NOTIFY_STATEFUL_SETS} to re-evaluate
all stateful sets based
+ * on the broadcast information.
*
* <p>A signature may declare multiple broadcast table arguments, but must
declare at least one
* {@link #ROW_SEMANTIC_TABLE} or {@link #SET_SEMANTIC_TABLE} argument
next to them.
@@ -109,6 +111,42 @@ public enum ArgumentTrait {
*/
OPTIONAL_PARTITION_BY(false, StaticArgumentTrait.OPTIONAL_PARTITION_BY),
+ /**
+ * Notifies all existing stateful sets to re-evaluate based on the given
broadcast row and the
+ * potentially updated broadcast state entries.
+ *
+ * <p>For every broadcast row, {@code eval()} is called multiple times:
+ *
+ * <ol>
+ * <li>First, without scoping to a set (as described in {@link
#BROADCAST_SEMANTIC_TABLE}):
+ * This call updates the broadcast state, if available.
+ * <li>Next, once for every set that holds at least one state entry: In
this phase, the {@code
+ * eval()} implementation has access to the incoming broadcast row,
as well as all (keyed)
+ * state entries.
+ * </ol>
+ *
+ * <p>At least one state entry scoped to a set must be declared. Sets
without state entries are
+ * not notified. Registered timers alone don't count as state. Sets whose
state entries have
+ * expired due to a time-to-live (TTL) might still be notified until the
state backend has
+ * cleaned them up. In this case, the state entries of the set are empty,
and writing to them
+ * during notification keeps the set alive.
+ *
+ * <p>During notification, state entries of the set and timers are
accessible and broadcast
+ * state is read-only. Similar to processing a row of the main table(s),
the PTF can emit
+ * results via {@code collect()} and register timers. Results contain the
partition key of the
+ * set.
+ *
+ * <p>Note: Notifying stateful sets is expensive. For every broadcast row,
all keys in state are
+ * loaded and a PTF's {@code eval()} is called for each of them.
Processing of other rows and
+ * checkpointing is blocked until all sets have been notified. Thus, this
trait should only be
+ * used if broadcast rows are rare compared to rows of the main table(s)
and the overall key
+ * space is relatively low.
+ *
+ * <p>Note: This trait is only valid for {@link #BROADCAST_SEMANTIC_TABLE}
arguments and
+ * requires at least one {@link #SET_SEMANTIC_TABLE} argument.
+ */
+ NOTIFY_STATEFUL_SETS(false, StaticArgumentTrait.NOTIFY_STATEFUL_SETS),
+
/**
* Defines that all columns of a table argument (i.e. {@link
#ROW_SEMANTIC_TABLE} or {@link
* #SET_SEMANTIC_TABLE}) are included in the output of the PTF. By
default, only columns of the
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/ProcessTableFunction.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/ProcessTableFunction.java
index 4fca7bbfd8d..89f7cc1915c 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/ProcessTableFunction.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/ProcessTableFunction.java
@@ -401,27 +401,6 @@ import java.time.LocalDateTime;
* {@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> {
@@ -446,6 +425,34 @@ import java.time.LocalDateTime;
* }
* }</pre>
*
+ * <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.
+ *
+ * <p>Use {@link ArgumentTrait#NOTIFY_STATEFUL_SETS} to re-evaluate previously
processed sets. For
+ * every broadcast row, {@code eval()} is then called multiple times: First,
without scoping to a
+ * set to update the broadcast state, if available. Next, once for every set
that holds at least one
+ * state entry: In this phase, the {@code eval()} implementation has access to
the incoming
+ * broadcast row, as well as all (keyed) state entries. Notifying stateful
sets is expensive and
+ * should only be used if broadcast rows are rare and the overall key space is
relatively low.
+ *
* <h1>Time and Timers</h1>
*
* <p>A PTF supports event time natively. Time-based services are available
via {@link
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/StaticArgumentTrait.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/StaticArgumentTrait.java
index 19fe3aea6df..ee495993ea8 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/StaticArgumentTrait.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/StaticArgumentTrait.java
@@ -50,6 +50,9 @@ public enum StaticArgumentTrait {
// For SET_SEMANTIC_TABLE
OPTIONAL_PARTITION_BY(SET_SEMANTIC_TABLE),
+ // For BROADCAST_SEMANTIC_TABLE
+ NOTIFY_STATEFUL_SETS(BROADCAST_SEMANTIC_TABLE),
+
// For SUPPORT_UPDATES
REQUIRE_UPDATE_BEFORE(SUPPORT_UPDATES),
REQUIRE_FULL_DELETE(SUPPORT_UPDATES);
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/SystemTypeInference.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/SystemTypeInference.java
index a7b7d6f5837..3e72cdc280f 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/SystemTypeInference.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/SystemTypeInference.java
@@ -113,7 +113,7 @@ public class SystemTypeInference {
origin.getInputTypeStrategy(),
origin.disableSystemArguments()));
if (functionKind == FunctionKind.PROCESS_TABLE) {
- checkBroadcastState(systemArgs, origin.getStateTypeStrategies());
+ checkBroadcastTraits(systemArgs, origin.getStateTypeStrategies());
}
builder.stateTypeStrategies(origin.getStateTypeStrategies());
builder.outputTypeStrategy(
@@ -277,17 +277,17 @@ public class SystemTypeInference {
}
}
+ private static void checkBroadcastTraits(
+ List<StaticArgument> staticArgs, Map<String, StateTypeStrategy>
stateTypeStrategies) {
+ checkBroadcastState(staticArgs, stateTypeStrategies);
+ checkNotifyStatefulSets(staticArgs);
+ }
+
private static void checkBroadcastState(
- @Nullable List<StaticArgument> staticArgs,
- Map<String, StateTypeStrategy> stateTypeStrategies) {
+ List<StaticArgument> staticArgs, Map<String, StateTypeStrategy>
stateTypeStrategies) {
final boolean hasBroadcastTables =
- staticArgs != null
- && staticArgs.stream()
- .anyMatch(
- arg ->
- arg.is(
- StaticArgumentTrait
-
.BROADCAST_SEMANTIC_TABLE));
+ staticArgs.stream()
+ .anyMatch(arg ->
arg.is(StaticArgumentTrait.BROADCAST_SEMANTIC_TABLE));
if (hasBroadcastTables) {
return;
}
@@ -304,6 +304,24 @@ public class SystemTypeInference {
});
}
+ private static void checkNotifyStatefulSets(List<StaticArgument>
staticArgs) {
+ if (staticArgs.stream().anyMatch(arg ->
arg.is(StaticArgumentTrait.SET_SEMANTIC_TABLE))) {
+ return;
+ }
+ staticArgs.stream()
+ .filter(arg ->
arg.is(StaticArgumentTrait.NOTIFY_STATEFUL_SETS))
+ .findFirst()
+ .ifPresent(
+ arg -> {
+ throw new ValidationException(
+ String.format(
+ "Table argument '%s' notifies
stateful sets which "
+ + "requires at least one
table argument with "
+ + "set semantics.",
+ arg.getName()));
+ });
+ }
+
private static void checkPassThroughColumns(List<StaticArgument>
staticArgs) {
final Set<StaticArgumentTrait> traits =
staticArgs.stream()
diff --git
a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/TypeInferenceExtractorTest.java
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/TypeInferenceExtractorTest.java
index 2fb20c161b6..ef611150c9e 100644
---
a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/TypeInferenceExtractorTest.java
+++
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/TypeInferenceExtractorTest.java
@@ -905,6 +905,12 @@ class TypeInferenceExtractorTest {
.expectErrorMessage(
"Invalid argument traits for argument 'rules'.
"
+ "Trait BROADCAST_SEMANTIC_TABLE is
incompatible with PASS_COLUMNS_THROUGH."),
+ // ---
+ TestSpec.forProcessTableFunction(
+
NotifyStatefulSetsWithoutBroadcastProcessTableFunction.class)
+ .expectErrorMessage(
+ "Invalid argument traits for argument 'data'. "
+ + "Trait NOTIFY_STATEFUL_SETS requires
BROADCAST_SEMANTIC_TABLE."),
TestSpec.forScalarFunction("Bitmap in scalar function",
BitmapTypeFunction.class)
.expectStaticArgument(
StaticArgument.scalar("bm",
DataTypes.BITMAP(), false))
@@ -2761,6 +2767,17 @@ class TypeInferenceExtractorTest {
@ArgumentHint(ArgumentTrait.SET_SEMANTIC_TABLE) Row data) {}
}
+ private static class NotifyStatefulSetsWithoutBroadcastProcessTableFunction
+ extends ProcessTableFunction<Integer> {
+ public void eval(
+ @StateHint MyFirstState count,
+ @ArgumentHint({
+ ArgumentTrait.SET_SEMANTIC_TABLE,
+ ArgumentTrait.NOTIFY_STATEFUL_SETS
+ })
+ Row data) {}
+ }
+
private static class PassThroughBroadcastTableProcessTableFunction
extends ProcessTableFunction<Integer> {
public void eval(
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecProcessTableFunction.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecProcessTableFunction.java
index 657f2e15cd6..e7cada6d076 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecProcessTableFunction.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecProcessTableFunction.java
@@ -344,6 +344,7 @@ public class StreamExecProcessTableFunction extends
ExecNodeBase<RowData>
tableArg.is(StaticArgumentTrait.PASS_COLUMNS_THROUGH),
tableArg.is(StaticArgumentTrait.SET_SEMANTIC_TABLE),
tableArg.is(StaticArgumentTrait.BROADCAST_SEMANTIC_TABLE),
+ tableArg.is(StaticArgumentTrait.NOTIFY_STATEFUL_SETS),
timeColumn,
upsertKeys);
}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionRestoreTests.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionRestoreTests.java
index a2d5cfa00ab..906ff0a6456 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionRestoreTests.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionRestoreTests.java
@@ -49,13 +49,16 @@ public class ProcessTableFunctionRestoreTests extends
RestoreTestBase {
ProcessTableFunctionTestPrograms.PROCESS_ORDER_BY_RESTORE,
ProcessTableFunctionTestPrograms.PROCESS_MULTI_INPUT_ORDER_BY_RESTORE,
ProcessTableFunctionTestPrograms.PROCESS_LATE_EVENTS_RESTORE,
-
ProcessTableFunctionTestPrograms.PROCESS_BROADCAST_STATE_RESTORE);
+
ProcessTableFunctionTestPrograms.PROCESS_BROADCAST_STATE_RESTORE,
+
ProcessTableFunctionTestPrograms.PROCESS_NOTIFY_STATEFUL_SETS_RESTORE);
}
@Override
protected void awaitSavepointReady(TableTestProgram program,
List<CompletableFuture<?>> futures)
throws Exception {
- if (program !=
ProcessTableFunctionTestPrograms.PROCESS_BROADCAST_STATE_RESTORE) {
+ if (program !=
ProcessTableFunctionTestPrograms.PROCESS_BROADCAST_STATE_RESTORE
+ && program
+ !=
ProcessTableFunctionTestPrograms.PROCESS_NOTIFY_STATEFUL_SETS_RESTORE) {
super.awaitSavepointReady(program, futures);
return;
}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java
index bbf6a8644c2..1faf06beb60 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java
@@ -102,6 +102,7 @@ public class ProcessTableFunctionSemanticTests extends
SemanticTestBase {
ProcessTableFunctionTestPrograms.PROCESS_BROADCAST_ON_TIME,
ProcessTableFunctionTestPrograms.PROCESS_BROADCAST_CLEAR_STATE,
ProcessTableFunctionTestPrograms.PROCESS_BROADCAST_MULTI_INPUT,
+ ProcessTableFunctionTestPrograms.PROCESS_NOTIFY_STATEFUL_SETS,
ProcessTableFunctionTestPrograms.PROCESS_INVALID_BROADCAST_COLLECT,
ProcessTableFunctionTestPrograms.PROCESS_INVALID_BROADCAST_TIMER,
ProcessTableFunctionTestPrograms.PROCESS_INVALID_BROADCAST_STATE_WRITE,
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java
index 9c7ebe85980..d2512eb0e54 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java
@@ -57,6 +57,8 @@ import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctio
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.MultiStateFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.NamedTimersFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.NonNullMapStateFunction;
+import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.NotifyRestoredSetFunction;
+import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.NotifyStatefulSetsRulesFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.OptionalOnTimeFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.OptionalPartitionOnTimeFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.OrderByFunction;
@@ -1852,6 +1854,71 @@ public class ProcessTableFunctionTestPrograms {
"INSERT INTO sink SELECT * FROM f(input => TABLE t
PARTITION BY name, rule => TABLE rules)")
.build();
+ public static final TableTestProgram PROCESS_NOTIFY_STATEFUL_SETS =
+ TableTestProgram.of(
+ "process-notify-stateful-sets",
+ "broadcast rows are applied to new sets or notify
existing sets")
+ .setupTemporarySystemFunction("f",
NotifyStatefulSetsRulesFunction.class)
+ .setupTableSource(TIMED_SOURCE)
+ .setupTableSource(TIMED_BROADCAST_RULES_SOURCE)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(KEYED_TIMED_BASE_SINK_SCHEMA)
+ .consumedValues(
+ // Rule Eve triggers the release
of Dave to all keys
+ "+I[Bob, {{Dave=2}},
1970-01-01T00:00:00.003Z]",
+ "+I[Alice, {{Dave=2}},
1970-01-01T00:00:00.003Z]",
+ // 6 Bob events processed with all
rules broadcasted and
+ // applied
+ "+I[Bob, {6, {Dave=2, Eve=5},
{Dave=2, Eve=5}}, 1970-01-01T00:00:01Z]",
+ // 1 Alice event processed with
all rules broadcasted
+ // and applied
+ "+I[Alice, {1, {Dave=2, Eve=5},
{Dave=2, Eve=5}}, 1970-01-01T00:00:01Z]")
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM f("
+ + "input => TABLE t PARTITION BY name, "
+ + "rule => TABLE rules, "
+ + "on_time => DESCRIPTOR(ts))")
+ .build();
+
+ public static final TableTestProgram PROCESS_NOTIFY_STATEFUL_SETS_RESTORE =
+ TableTestProgram.of(
+ "process-notify-stateful-sets-restore",
+ "broadcast rows notify sets that have been
restored")
+ .setupTemporarySystemFunction("f",
NotifyRestoredSetFunction.class)
+ .setupTableSource(BROADCAST_RESTORE_SOURCE)
+ .setupTableSource(
+ SourceTestStep.newBuilder("rules")
+ .addSchema(
+ "name STRING",
+ "weight INT",
+ "ts TIMESTAMP_LTZ(3)",
+ "WATERMARK FOR ts AS ts")
+ .producedBeforeRestore(
+ Row.of("Dave", 3,
Instant.ofEpochMilli(0)))
+ // The rule after restore targets the
restored set
+ .producedAfterRestore(Row.of("Bob", 4,
Instant.ofEpochMilli(4)))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(KEYED_TIMED_BASE_SINK_SCHEMA)
+ .consumedBeforeRestore(
+ "+I[Bob, {+I[Bob, -1,
1970-01-01T00:00:00Z]}, 1970-01-01T00:00:00Z]")
+ .consumedAfterRestore(
+ "+I[Bob, {+I[Bob, 99,
1970-01-01T00:00:00.001Z]}, 1970-01-01T00:00:00.001Z]",
+ "+I[Alice, {+I[Alice, 42,
1970-01-01T00:00:00.002Z]}, 1970-01-01T00:00:00.002Z]",
+ "+I[Charly, {+I[Charly, 7,
1970-01-01T00:00:00.003Z]}, 1970-01-01T00:00:00.003Z]",
+ // Notification carries the
rowtime of the broadcast row
+ "+I[Bob, {+I[Bob, 4,
1970-01-01T00:00:00.004Z], {Bob=4, Dave=3}}, 1970-01-01T00:00:00.004Z]")
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM f("
+ + "input => TABLE t PARTITION BY name, "
+ + "rule => TABLE rules, "
+ + "on_time => DESCRIPTOR(ts))")
+ .build();
+
public static final TableTestProgram PROCESS_INVALID_BROADCAST_COLLECT =
TableTestProgram.of(
"process-invalid-broadcast-collect",
@@ -1889,8 +1956,9 @@ public class ProcessTableFunctionTestPrograms {
.runFailingSql(
"SELECT * FROM f(input => TABLE t PARTITION BY
name, rule => TABLE rules)",
TableRuntimeException.class,
- "Broadcast state entry 'weights' is read-only
while processing a table "
- + "with row or set semantics.")
+ "Broadcast state entry 'weights' is read-only. It
can only be modified while "
+ + "processing a row of a table with
broadcast semantics, not while "
+ + "processing other tables, firing timers,
or notifying stateful sets.")
.build();
public static final TableTestProgram PROCESS_MULTI_INPUT =
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestUtils.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestUtils.java
index 1e5dad159c3..1122c322a0e 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestUtils.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestUtils.java
@@ -62,6 +62,7 @@ import java.util.Optional;
import java.util.stream.Collectors;
import static
org.apache.flink.table.annotation.ArgumentTrait.BROADCAST_SEMANTIC_TABLE;
+import static
org.apache.flink.table.annotation.ArgumentTrait.NOTIFY_STATEFUL_SETS;
import static
org.apache.flink.table.annotation.ArgumentTrait.OPTIONAL_PARTITION_BY;
import static
org.apache.flink.table.annotation.ArgumentTrait.PASS_COLUMNS_THROUGH;
import static
org.apache.flink.table.annotation.ArgumentTrait.REQUIRE_FULL_DELETE;
@@ -1492,6 +1493,123 @@ public class ProcessTableFunctionTestUtils {
}
}
+ /**
+ * Testing function that applies every broadcast rule exactly once to
every set. Depending on
+ * the order in which tables are processed, a rule is applied when a set
is created or when an
+ * existing set is notified. Timers act as synchronization barriers as
they only fire after all
+ * broadcast rows up to the timer's time have been processed.
+ */
+ public static class NotifyStatefulSetsRulesFunction extends
AppendProcessTableFunctionBase {
+ @SuppressWarnings("unused")
+ public void eval(
+ Context ctx,
+ @StateHint ValueView<Integer> count,
+ @StateHint MapView<String, Long> appliedRules,
+ @StateHint(StateKind.BROADCAST) MapView<String, Long> allRules,
+ @ArgumentHint(SET_SEMANTIC_TABLE) Row input,
+ @ArgumentHint({BROADCAST_SEMANTIC_TABLE,
NOTIFY_STATEFUL_SETS}) Row rule)
+ throws Exception {
+ final TimeContext<Long> timeCtx = ctx.timeContext(Long.class);
+
+ // Process row from table 't'
+ if (input != null) {
+ if (count.getValue() == null) {
+ // New set: catch up with all rules received so far
+ for (Map.Entry<String, Long> r : allRules.entries()) {
+ appliedRules.put(r.getKey(), r.getValue());
+ }
+ // Fires once all rules up to this time have been processed
+ // (in other words: when Eve triggers the release of Dave
to all keys)
+ timeCtx.registerOnTime("middle", 3L);
+ // Exceeds all input timestamps, thus, only fires at the
end of all tables
+ timeCtx.registerOnTime("end", 1000L);
+ count.setValue(1);
+ } else {
+ count.setValue(count.getValue() + 1);
+ }
+ return;
+ }
+
+ final String ruleName = rule.getFieldAs("name");
+ final long ruleTime = timeCtx.time();
+
+ // Process row from table 'rules' - broadcast phase:
+ // state scoped to a set is null, store the rule
+ if (count == null) {
+ allRules.put(ruleName, ruleTime);
+ return;
+ }
+
+ // Process row from table 'rules' - notification phase:
+ // state scoped to a set is present, apply the stored rule exactly
once
+ assertThat(allRules.get(ruleName)).isEqualTo(ruleTime);
+ assertThat(appliedRules.contains(ruleName)).isFalse();
+ assertThatThrownBy(() -> allRules.remove(ruleName))
+ .isInstanceOf(TableRuntimeException.class)
+ .hasMessageContaining("is read-only");
+ appliedRules.put(ruleName, ruleTime);
+ }
+
+ public void onTimer(
+ OnTimerContext ctx,
+ ValueView<Integer> count,
+ MapView<String, Long> appliedRules,
+ MapView<String, Long> allRules)
+ throws Exception {
+ if (ctx.currentTimer().equals("end")) {
+ // All rules have been applied to every set
+ collectObjects(
+ count.getValue(), toSortedString(appliedRules),
toSortedString(allRules));
+ return;
+ }
+ // Only rules up to the timer's time are guaranteed to be applied
+ final long time = ctx.timeContext(Long.class).time();
+ final MapView<String, Long> passedRules = new MapView<>();
+ for (Map.Entry<String, Long> r : appliedRules.entries()) {
+ if (r.getValue() <= time) {
+ passedRules.put(r.getKey(), r.getValue());
+ }
+ }
+ collectObjects(toSortedString(passedRules));
+ }
+ }
+
+ /**
+ * Testing function that notifies the set targeted by a rule. All output
is independent of the
+ * order in which main and broadcast tables are processed.
+ */
+ public static class NotifyRestoredSetFunction extends
AppendProcessTableFunctionBase {
+ @SuppressWarnings("unused")
+ public void eval(
+ @StateHint ValueView<String> setName,
+ @StateHint(StateKind.BROADCAST) MapView<String, Integer> rules,
+ @ArgumentHint(SET_SEMANTIC_TABLE) Row input,
+ @ArgumentHint({BROADCAST_SEMANTIC_TABLE,
NOTIFY_STATEFUL_SETS}) Row rule)
+ throws Exception {
+ // Process row from table 't'
+ if (input != null) {
+ setName.setValue(input.getFieldAs("name"));
+ collectObjects(input);
+ return;
+ }
+
+ final String ruleName = rule.getFieldAs("name");
+
+ // Process row from table 'rules' - broadcast phase:
+ // state scoped to a set is null, store the rule
+ if (setName == null) {
+ rules.put(ruleName, rule.getFieldAs("weight"));
+ return;
+ }
+
+ // Process row from table 'rules' - notification phase:
+ // only the targeted set reacts, other sets might not exist yet
+ if (ruleName.equals(setName.getValue())) {
+ collectObjects(rule, toSortedString(rules));
+ }
+ }
+ }
+
/** POJO for broadcast state. */
public static class RuleStats {
public int rules;
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ProcessTableFunctionTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ProcessTableFunctionTest.java
index d1f06a38418..5e45f045098 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ProcessTableFunctionTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ProcessTableFunctionTest.java
@@ -80,6 +80,7 @@ import java.util.stream.Stream;
import static java.util.Collections.singletonList;
import static org.apache.flink.core.testutils.FlinkAssertions.anyCauseMatches;
import static
org.apache.flink.table.annotation.ArgumentTrait.BROADCAST_SEMANTIC_TABLE;
+import static
org.apache.flink.table.annotation.ArgumentTrait.NOTIFY_STATEFUL_SETS;
import static
org.apache.flink.table.annotation.ArgumentTrait.OPTIONAL_PARTITION_BY;
import static
org.apache.flink.table.annotation.ArgumentTrait.PASS_COLUMNS_THROUGH;
import static
org.apache.flink.table.annotation.ArgumentTrait.ROW_SEMANTIC_TABLE;
@@ -562,6 +563,12 @@ class ProcessTableFunctionTest extends TableTestBase {
InvalidBroadcastStateFunction.class,
"SELECT * FROM f(input => TABLE t PARTITION BY name)",
"Broadcast state entry 's' requires at least one table
argument with broadcast semantics."),
+ ErrorSpec.ofSelect(
+ "notify stateful sets without set semantic table",
+ InvalidNotifyStatefulSetsWithRowFunction.class,
+ "SELECT * FROM f(input => TABLE t, rule => TABLE
t_rules)",
+ "Table argument 'rule' notifies stateful sets which
requires at least one "
+ + "table argument with set semantics."),
ErrorSpec.ofSelect(
"missing on_time for broadcast table",
BroadcastTimersFunction.class,
@@ -837,6 +844,16 @@ class ProcessTableFunctionTest extends TableTestBase {
@ArgumentHint(SET_SEMANTIC_TABLE) Row input) {}
}
+ /** Testing function. */
+ public static class InvalidNotifyStatefulSetsWithRowFunction
+ extends ProcessTableFunction<String> {
+ @SuppressWarnings("unused")
+ public void eval(
+ @StateHint MapView<String, Integer> s,
+ @ArgumentHint(ROW_SEMANTIC_TABLE) Row input,
+ @ArgumentHint({BROADCAST_SEMANTIC_TABLE,
NOTIFY_STATEFUL_SETS}) Row rule) {}
+ }
+
/** Testing function. */
public static class UpdatingBroadcastFunction extends
ProcessTableFunction<String> {
@SuppressWarnings("unused")
diff --git
a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-process-table-function_1/process-notify-stateful-sets-restore/plan/process-notify-stateful-sets-restore.json
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-process-table-function_1/process-notify-stateful-sets-restore/plan/process-notify-stateful-sets-restore.json
new file mode 100644
index 00000000000..ec1868b6694
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-process-table-function_1/process-notify-stateful-sets-restore/plan/process-notify-stateful-sets-restore.json
@@ -0,0 +1,482 @@
+{
+ "flinkVersion" : "2.4",
+ "nodes" : [ {
+ "id" : 46,
+ "type" : "stream-exec-table-source-scan_2",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "name",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "score",
+ "dataType" : "INT"
+ }, {
+ "name" : "ts",
+ "dataType" : {
+ "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "precision" : 3,
+ "kind" : "ROWTIME"
+ }
+ } ],
+ "watermarkSpecs" : [ {
+ "rowtimeAttribute" : "ts",
+ "expression" : {
+ "rexNode" : {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$-$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "TIMESTAMP(3) WITH LOCAL TIME ZONE"
+ }, {
+ "kind" : "LITERAL",
+ "value" : "1",
+ "type" : "INTERVAL SECOND(6) NOT NULL"
+ } ],
+ "type" : "TIMESTAMP(3) WITH LOCAL TIME ZONE"
+ },
+ "serializableString" : "`ts` - INTERVAL '0.001' SECOND"
+ }
+ } ]
+ }
+ }
+ },
+ "abilities" : [ {
+ "type" : "WatermarkPushDown",
+ "watermarkExpr" : {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$-$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "TIMESTAMP(3) WITH LOCAL TIME ZONE"
+ }, {
+ "kind" : "LITERAL",
+ "value" : "1",
+ "type" : "INTERVAL SECOND(6) NOT NULL"
+ } ],
+ "type" : "TIMESTAMP(3) WITH LOCAL TIME ZONE"
+ },
+ "rowtimeExpr" : {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "TIMESTAMP(3) WITH LOCAL TIME ZONE"
+ },
+ "idleTimeoutMillis" : -1,
+ "producedType" : {
+ "type" : "ROW",
+ "nullable" : false,
+ "fields" : [ {
+ "name" : "name",
+ "fieldType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "score",
+ "fieldType" : "INT"
+ }, {
+ "name" : "ts",
+ "fieldType" : {
+ "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "precision" : 3,
+ "kind" : "ROWTIME"
+ }
+ } ]
+ },
+ "watermarkParams" : {
+ "emitStrategy" : "ON_EVENT",
+ "alignGroupName" : null,
+ "alignMaxDrift" : "PT0S",
+ "alignUpdateInterval" : "PT1S",
+ "sourceIdleTimeout" : -1
+ }
+ } ]
+ },
+ "outputType" : {
+ "type" : "ROW",
+ "fields" : [ {
+ "name" : "name",
+ "fieldType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "score",
+ "fieldType" : "INT"
+ }, {
+ "name" : "ts",
+ "fieldType" : {
+ "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "precision" : 3,
+ "kind" : "ROWTIME"
+ }
+ } ]
+ },
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, t, watermark=[-(ts, 1:INTERVAL SECOND)],
watermarkEmitStrategy=[on-event]]], fields=[name, score, ts])"
+ }, {
+ "id" : 47,
+ "type" : "stream-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : {
+ "type" : "ROW",
+ "fields" : [ {
+ "name" : "name",
+ "fieldType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "score",
+ "fieldType" : "INT"
+ }, {
+ "name" : "ts",
+ "fieldType" : {
+ "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "precision" : 3,
+ "kind" : "ROWTIME"
+ }
+ } ]
+ },
+ "description" : "Exchange(distribution=[hash[name]])"
+ }, {
+ "id" : 48,
+ "type" : "stream-exec-table-source-scan_2",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`rules`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "name",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "weight",
+ "dataType" : "INT"
+ }, {
+ "name" : "ts",
+ "dataType" : {
+ "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "precision" : 3,
+ "kind" : "ROWTIME"
+ }
+ } ],
+ "watermarkSpecs" : [ {
+ "rowtimeAttribute" : "ts",
+ "expression" : {
+ "rexNode" : {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "TIMESTAMP(3) WITH LOCAL TIME ZONE"
+ },
+ "serializableString" : "`ts`"
+ }
+ } ]
+ }
+ }
+ },
+ "abilities" : [ {
+ "type" : "WatermarkPushDown",
+ "watermarkExpr" : {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "TIMESTAMP(3) WITH LOCAL TIME ZONE"
+ },
+ "rowtimeExpr" : {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "TIMESTAMP(3) WITH LOCAL TIME ZONE"
+ },
+ "idleTimeoutMillis" : -1,
+ "producedType" : {
+ "type" : "ROW",
+ "nullable" : false,
+ "fields" : [ {
+ "name" : "name",
+ "fieldType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "weight",
+ "fieldType" : "INT"
+ }, {
+ "name" : "ts",
+ "fieldType" : {
+ "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "precision" : 3,
+ "kind" : "ROWTIME"
+ }
+ } ]
+ },
+ "watermarkParams" : {
+ "emitStrategy" : "ON_EVENT",
+ "alignGroupName" : null,
+ "alignMaxDrift" : "PT0S",
+ "alignUpdateInterval" : "PT1S",
+ "sourceIdleTimeout" : -1
+ }
+ } ]
+ },
+ "outputType" : {
+ "type" : "ROW",
+ "fields" : [ {
+ "name" : "name",
+ "fieldType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "weight",
+ "fieldType" : "INT"
+ }, {
+ "name" : "ts",
+ "fieldType" : {
+ "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "precision" : 3,
+ "kind" : "ROWTIME"
+ }
+ } ]
+ },
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, rules, watermark=[ts], watermarkEmitStrategy=[on-event]]],
fields=[name, weight, ts])"
+ }, {
+ "id" : 49,
+ "type" : "stream-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "BROADCAST"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : {
+ "type" : "ROW",
+ "fields" : [ {
+ "name" : "name",
+ "fieldType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "weight",
+ "fieldType" : "INT"
+ }, {
+ "name" : "ts",
+ "fieldType" : {
+ "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "precision" : 3,
+ "kind" : "ROWTIME"
+ }
+ } ]
+ },
+ "description" : "Exchange(distribution=[broadcast])"
+ }, {
+ "id" : 50,
+ "type" : "stream-exec-process-table-function_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ }, {
+ "requiredDistribution" : {
+ "type" : "BROADCAST"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : {
+ "type" : "ROW",
+ "fields" : [ {
+ "name" : "name",
+ "fieldType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "out",
+ "fieldType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "rowtime",
+ "fieldType" : {
+ "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "nullable" : false,
+ "precision" : 3,
+ "kind" : "ROWTIME"
+ }
+ } ]
+ },
+ "description" : "ProcessTableFunction(invocation=[f(TABLE(#0) PARTITION
BY($0), TABLE(#1), DESCRIPTOR(_UTF-16LE'ts'), DEFAULT())], uid=[f],
select=[name,out,rowtime], rowType=[RecordType(VARCHAR(2147483647) name,
VARCHAR(2147483647) out, TIMESTAMP_LTZ(3) *ROWTIME* rowtime)])",
+ "uid" : "f",
+ "functionCall" : {
+ "kind" : "CALL",
+ "systemName" : "f",
+ "operands" : [ {
+ "kind" : "TABLE_ARG_CALL",
+ "inputIndex" : 0,
+ "partitionKeys" : [ 0 ],
+ "type" : {
+ "type" : "ROW",
+ "nullable" : false,
+ "fields" : [ {
+ "name" : "name",
+ "fieldType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "score",
+ "fieldType" : "INT"
+ }, {
+ "name" : "ts",
+ "fieldType" : {
+ "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "precision" : 3,
+ "kind" : "ROWTIME"
+ }
+ } ]
+ }
+ }, {
+ "kind" : "TABLE_ARG_CALL",
+ "inputIndex" : 1,
+ "type" : {
+ "type" : "ROW",
+ "nullable" : false,
+ "fields" : [ {
+ "name" : "name",
+ "fieldType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "weight",
+ "fieldType" : "INT"
+ }, {
+ "name" : "ts",
+ "fieldType" : {
+ "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "precision" : 3,
+ "kind" : "ROWTIME"
+ }
+ } ]
+ }
+ }, {
+ "kind" : "CALL",
+ "internalName" : "$DESCRIPTOR$1",
+ "operands" : [ {
+ "kind" : "LITERAL",
+ "value" : "ts",
+ "type" : "CHAR(2) NOT NULL"
+ } ],
+ "type" : "DESCRIPTOR NOT NULL"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$DEFAULT$1",
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "type" : {
+ "type" : "ROW",
+ "nullable" : false,
+ "fields" : [ {
+ "name" : "name",
+ "fieldType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "out",
+ "fieldType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "rowtime",
+ "fieldType" : {
+ "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "nullable" : false,
+ "precision" : 3,
+ "kind" : "ROWTIME"
+ }
+ } ]
+ }
+ },
+ "inputChangelogModes" : [ [ "INSERT" ], [ "INSERT" ] ],
+ "outputChangelogMode" : [ "INSERT" ],
+ "inputUpsertKeys" : [ [ ], [ ] ]
+ }, {
+ "id" : 51,
+ "type" : "stream-exec-sink_2",
+ "configuration" : {
+ "table.exec.sink.keyed-shuffle" : "AUTO",
+ "table.exec.sink.not-null-enforcer" : "ERROR",
+ "table.exec.sink.rowtime-inserter" : "ENABLED",
+ "table.exec.sink.type-length-enforcer" : "IGNORE",
+ "table.exec.sink.upsert-materialize" : "AUTO"
+ },
+ "dynamicTableSink" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`sink`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "name",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "out",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "rowtime",
+ "dataType" : "TIMESTAMP(3) WITH LOCAL TIME ZONE"
+ } ]
+ }
+ }
+ }
+ },
+ "inputChangelogMode" : [ "INSERT" ],
+ "upsertMaterializeStrategy" : "MAP",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : {
+ "type" : "ROW",
+ "fields" : [ {
+ "name" : "name",
+ "fieldType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "out",
+ "fieldType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "rowtime",
+ "fieldType" : {
+ "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "nullable" : false,
+ "precision" : 3,
+ "kind" : "ROWTIME"
+ }
+ } ]
+ },
+ "description" : "Sink(table=[default_catalog.default_database.sink],
fields=[name, out, rowtime])"
+ } ],
+ "edges" : [ {
+ "source" : 46,
+ "target" : 47,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 48,
+ "target" : 49,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 47,
+ "target" : 50,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 49,
+ "target" : 50,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 50,
+ "target" : 51,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
diff --git
a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-process-table-function_1/process-notify-stateful-sets-restore/savepoint/_metadata
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-process-table-function_1/process-notify-stateful-sets-restore/savepoint/_metadata
new file mode 100644
index 00000000000..fd8a68f5b73
Binary files /dev/null and
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-process-table-function_1/process-notify-stateful-sets-restore/savepoint/_metadata
differ
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/generated/ProcessTableRunner.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/generated/ProcessTableRunner.java
index e64dd3398e3..7bf9c914a02 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/generated/ProcessTableRunner.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/generated/ProcessTableRunner.java
@@ -87,19 +87,54 @@ public abstract class ProcessTableRunner extends
AbstractRichFunction {
this.onTimerCollector = onTimerCollector;
}
+ /**
+ * Ingests a row of a table with row or set semantics for the next {@link
#processEval()} call.
+ * The output prefix is derived from the row (i.e. partition keys or all
columns).
+ *
+ * @param pos position of the table argument
+ * @param row incoming row
+ * @param timeColumn position of the time attribute in the row, or -1 if
not available
+ * @param watermark current watermark of the table
+ */
public void ingestTableEvent(int pos, RowData row, int timeColumn, long
watermark) {
evalCollector.setPrefix(pos, row);
ingestRow(pos, row, timeColumn, watermark);
- if (emitRowtime && rowtime != null) {
- evalCollector.setRowtime(rowtime);
- }
+ setOutputRowtime();
}
+ /**
+ * Ingests a row of a table with broadcast semantics for the next {@link
+ * #processBroadcastEval()} call. Broadcast rows have no key context and
must not emit results,
+ * thus, no output prefix is set.
+ *
+ * @param pos position of the table argument
+ * @param row incoming row
+ * @param timeColumn position of the time attribute in the row, or -1 if
not available
+ * @param watermark current watermark of the table
+ */
public void ingestBroadcastTableEvent(int pos, RowData row, int
timeColumn, long watermark) {
- // Broadcast rows have no key context and never emit, thus, there is
no output prefix
ingestRow(pos, row, timeColumn, watermark);
}
+ /**
+ * Ingests the key context of a stateful set for the next {@link
#processEval()} call that
+ * notifies the set about the broadcast row. The broadcast row must have
been ingested before
+ * via {@link #ingestBroadcastTableEvent}, only the key context changes.
The output prefix is
+ * derived from the key.
+ *
+ * @param key key of the stateful set
+ */
+ public void ingestBroadcastNotification(RowData key) {
+ evalCollector.setKeyPrefix(key);
+ setOutputRowtime();
+ }
+
+ private void setOutputRowtime() {
+ if (emitRowtime && rowtime != null) {
+ evalCollector.setRowtime(rowtime);
+ }
+ }
+
private void ingestRow(int pos, RowData row, int timeColumn, long
watermark) {
if (timeColumn == -1) {
rowtime = null;
@@ -111,6 +146,14 @@ public abstract class ProcessTableRunner extends
AbstractRichFunction {
tableWatermark = watermark;
}
+ /**
+ * Ingests a fired timer for the next {@link #processOnTimer()} call. The
output prefix is
+ * derived from the key.
+ *
+ * @param key key of the set that registered the timer
+ * @param name name of the timer, or null for an unnamed timer
+ * @param timerTime time for which the timer was registered
+ */
public void ingestTimerEvent(RowData key, @Nullable StringData name, long
timerTime) {
onTimerCollector.setPrefix(-1, key);
if (emitRowtime) {
@@ -217,8 +260,8 @@ public abstract class ProcessTableRunner extends
AbstractRichFunction {
}
if (stateHandle.kind == Kind.EAGER_VALUE) {
if (!processingBroadcast && stateHandle.broadcast) {
- // Broadcast state is read-only while processing other
tables, thus,
- // modifications of eager value state are discarded
+ // Broadcast state is read-only while processing main
tables or notifying
+ // stateful sets, thus, modifications of eager value state
are discarded
continue;
}
moveValueStateFromFunction(stateHandle);
@@ -278,6 +321,7 @@ public abstract class ProcessTableRunner extends
AbstractRichFunction {
EAGER_VALUE
}
+ private final String name;
private final Kind kind;
/** Whether {@link #state} is an adapter for Flink's broadcast state.
*/
@@ -302,11 +346,13 @@ public abstract class ProcessTableRunner extends
AbstractRichFunction {
private boolean cleared;
public StateHandle(
+ String name,
Kind kind,
boolean broadcast,
State state,
@Nullable HashFunction hashFunction,
@Nullable RecordEqualiser equaliser) {
+ this.name = name;
this.kind = kind;
this.broadcast = broadcast;
this.state = state;
@@ -333,5 +379,18 @@ public abstract class ProcessTableRunner extends
AbstractRichFunction {
public boolean isCleared() {
return cleared;
}
+
+ @Override
+ public String toString() {
+ return "StateHandle{"
+ + "name='"
+ + name
+ + '\''
+ + ", kind="
+ + kind
+ + ", broadcast="
+ + broadcast
+ + '}';
+ }
}
}
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/AbstractProcessTableOperator.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/AbstractProcessTableOperator.java
index b2d614e3228..b3906336968 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/AbstractProcessTableOperator.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/AbstractProcessTableOperator.java
@@ -33,6 +33,7 @@ import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeutils.base.LongSerializer;
import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.runtime.state.KeyedStateBackend;
import org.apache.flink.runtime.state.VoidNamespace;
import org.apache.flink.runtime.state.VoidNamespaceSerializer;
import org.apache.flink.streaming.api.operators.AbstractStreamOperatorV2;
@@ -75,10 +76,12 @@ import javax.annotation.Nullable;
import java.time.Instant;
import java.time.LocalDateTime;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
+import java.util.stream.Stream;
/** Base class for operators for {@link ProcessTableFunction}. */
@Internal
@@ -100,6 +103,7 @@ public abstract class AbstractProcessTableOperator extends
AbstractStreamOperato
private transient PassAllCollector onTimerCollector;
private transient StateDescriptor<?, ?>[] stateDescriptors;
private transient StateHandle[] stateHandles;
+ private transient List<String> keyedStateNames;
private transient @Nullable MapState<StringData, Long> namedTimersMapState;
private transient @Nullable InternalTimerService<StringData>
namedTimerService;
@@ -181,6 +185,27 @@ public abstract class AbstractProcessTableOperator extends
AbstractStreamOperato
@Override
public void onProcessingTime(InternalTimer<RowData, Object> timer) throws
Exception {}
+ /**
+ * Calls the function for every set that holds at least one state entry on
this virtual
+ * processor. The previously ingested broadcast row is passed together
with the set's key
+ * context.
+ */
+ protected void notifyStatefulSets() throws Exception {
+ // Keys are materialized upfront because state backends don't support
modifications of
+ // state while iterating over its keys
+ final KeyedStateBackend<RowData> keyedStateBackend =
getKeyedStateBackend();
+ final List<RowData> keys;
+ try (Stream<RowData> keyStream =
+ keyedStateBackend.getKeys(keyedStateNames,
VoidNamespace.INSTANCE)) {
+ keys = keyStream.collect(Collectors.toList());
+ }
+ for (RowData key : keys) {
+ setCurrentKey(key);
+ processTableRunner.ingestBroadcastNotification(key);
+ processTableRunner.processEval();
+ }
+ }
+
//
--------------------------------------------------------------------------------------------
// Context implementation
//
--------------------------------------------------------------------------------------------
@@ -447,6 +472,7 @@ public abstract class AbstractProcessTableOperator extends
AbstractStreamOperato
private void setStateHandles() throws Exception {
final StateHandle[] stateHandles = new
StateHandle[stateDescriptors.length];
+ final List<String> keyedStateNames = new ArrayList<>();
for (int i = 0; i < stateDescriptors.length; i++) {
final RuntimeStateInfo stateInfo = stateInfos.get(i);
// The kind must be derived from the declared type, not the
descriptor, because both
@@ -458,6 +484,7 @@ public abstract class AbstractProcessTableOperator extends
AbstractStreamOperato
stateHandle =
createBroadcastStateHandle(stateInfo.getStateName(), kind, i);
} else {
stateHandle = createKeyedStateHandle(i);
+ keyedStateNames.add(stateInfo.getStateName());
}
// Hash function and equaliser are only used by eager value state
to reduce state
// updates; views access state lazily and don't need them.
@@ -466,9 +493,15 @@ public abstract class AbstractProcessTableOperator extends
AbstractStreamOperato
final RecordEqualiser equaliser = isEagerValue ? stateEquals[i] :
null;
stateHandles[i] =
new StateHandle(
- kind, stateInfo.isBroadcast(), stateHandle,
hashFunction, equaliser);
+ stateInfo.getStateName(),
+ kind,
+ stateInfo.isBroadcast(),
+ stateHandle,
+ hashFunction,
+ equaliser);
}
this.stateHandles = stateHandles;
+ this.keyedStateNames = keyedStateNames;
}
@SuppressWarnings({"unchecked", "rawtypes"})
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/BroadcastEvalCollector.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/BroadcastEvalCollector.java
index b507585b2b6..28b06a3a852 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/BroadcastEvalCollector.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/BroadcastEvalCollector.java
@@ -53,6 +53,11 @@ public class BroadcastEvalCollector extends
PassThroughCollectorBase {
evalCollector.setPrefix(pos, input);
}
+ @Override
+ public void setKeyPrefix(RowData key) {
+ evalCollector.setKeyPrefix(key);
+ }
+
@Override
public void setRowtime(Long time) {
evalCollector.setRowtime(time);
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/BroadcastStateAdapters.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/BroadcastStateAdapters.java
index 2ef0c9f4f0e..03d04587d88 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/BroadcastStateAdapters.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/BroadcastStateAdapters.java
@@ -218,8 +218,9 @@ public final class BroadcastStateAdapters {
public static TableRuntimeException broadcastStateIsReadOnly(String
stateName) {
return new TableRuntimeException(
String.format(
- "Broadcast state entry '%s' is read-only while
processing a table "
- + "with row or set semantics.",
+ "Broadcast state entry '%s' is read-only. It can only
be modified while "
+ + "processing a row of a table with broadcast
semantics, not while "
+ + "processing other tables, firing timers, or
notifying stateful sets.",
stateName));
}
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/PassThroughCollectorBase.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/PassThroughCollectorBase.java
index 5b032c3891f..799b203d821 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/PassThroughCollectorBase.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/PassThroughCollectorBase.java
@@ -59,6 +59,11 @@ public abstract class PassThroughCollectorBase extends
StreamRecordCollector<Row
public abstract void setPrefix(int pos, RowData input);
+ /** Sets the prefix from a key, i.e. the partition keys of all tables with
set semantics. */
+ public void setKeyPrefix(RowData key) {
+ prefix = key;
+ }
+
public void setRowtime(Long time) {
rowtime = GenericRowData.of(TimestampData.fromEpochMillis(time));
}
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/ProcessSetTableOperator.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/ProcessSetTableOperator.java
index 402622c5b44..7b02d063e30 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/ProcessSetTableOperator.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/ProcessSetTableOperator.java
@@ -52,7 +52,8 @@ import java.util.stream.IntStream;
* least one table with set semantics or at least one table with broadcast
semantics.
*
* <p>Tables with broadcast semantics have no key context. Similar to tables
with set semantics,
- * they contribute to the operator's watermark.
+ * they contribute to the operator's watermark. If configured, every broadcast
row is additionally
+ * forwarded to all sets that hold state.
*/
public class ProcessSetTableOperator extends AbstractProcessTableOperator
implements MultipleInputStreamOperator<RowData> {
@@ -113,6 +114,9 @@ public class ProcessSetTableOperator extends
AbstractProcessTableOperator
if
(inputSemantics.hasBroadcastSemantics()) {
processBroadcastTableEvent(
inputIdx, element.getValue(),
timeColumn);
+ if
(inputSemantics.notifiesStatefulSets()) {
+ notifyStatefulSets();
+ }
return;
}
final InputSortBuffer sortBuffer =
inputSortBuffers[inputIdx];
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/RuntimeStateInfo.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/RuntimeStateInfo.java
index 501c81d6a58..ca19574e42d 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/RuntimeStateInfo.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/RuntimeStateInfo.java
@@ -59,4 +59,19 @@ public class RuntimeStateInfo implements Serializable {
public boolean isBroadcast() {
return broadcast;
}
+
+ @Override
+ public String toString() {
+ return "RuntimeStateInfo{"
+ + "stateName='"
+ + stateName
+ + '\''
+ + ", dataType="
+ + dataType
+ + ", timeToLive="
+ + timeToLive
+ + ", broadcast="
+ + broadcast
+ + '}';
+ }
}
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/RuntimeTableSemantics.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/RuntimeTableSemantics.java
index e70f201ec5d..59275e509d8 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/RuntimeTableSemantics.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/RuntimeTableSemantics.java
@@ -46,6 +46,7 @@ public class RuntimeTableSemantics implements TableSemantics,
Serializable {
private final boolean passColumnsThrough;
private final boolean hasSetSemantics;
private final boolean hasBroadcastSemantics;
+ private final boolean notifiesStatefulSets;
private final int timeColumn;
private final List<int[]> upsertKeyColumns;
@@ -62,6 +63,7 @@ public class RuntimeTableSemantics implements TableSemantics,
Serializable {
boolean passColumnsThrough,
boolean hasSetSemantics,
boolean hasBroadcastSemantics,
+ boolean notifiesStatefulSets,
int timeColumn,
List<int[]> upsertKeyColumns) {
this.argName = argName;
@@ -74,6 +76,7 @@ public class RuntimeTableSemantics implements TableSemantics,
Serializable {
this.passColumnsThrough = passColumnsThrough;
this.hasSetSemantics = hasSetSemantics;
this.hasBroadcastSemantics = hasBroadcastSemantics;
+ this.notifiesStatefulSets = notifiesStatefulSets;
this.timeColumn = timeColumn;
this.upsertKeyColumns = upsertKeyColumns;
}
@@ -98,6 +101,10 @@ public class RuntimeTableSemantics implements
TableSemantics, Serializable {
return hasBroadcastSemantics;
}
+ public boolean notifiesStatefulSets() {
+ return notifiesStatefulSets;
+ }
+
public ChangelogMode getChangelogMode() {
if (changelogMode == null) {
changelogMode = consumedChangelogMode.deserialize();
diff --git
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/process/ProcessSetTableOperatorInterruptibleTimersTest.java
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/process/ProcessSetTableOperatorInterruptibleTimersTest.java
index 4fe65a92485..0c995e2771d 100644
---
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/process/ProcessSetTableOperatorInterruptibleTimersTest.java
+++
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/process/ProcessSetTableOperatorInterruptibleTimersTest.java
@@ -251,6 +251,7 @@ class ProcessSetTableOperatorInterruptibleTimersTest {
/* passColumnsThrough */ false,
/* hasSetSemantics */ true,
/* hasBroadcastSemantics */ false,
+ /* notifiesStatefulSets */ false,
/* timeColumn */ 1,
/* upsertKeyColumns */ List.of());
}
diff --git
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/process/ProcessSetTableOperatorTest.java
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/process/ProcessSetTableOperatorTest.java
new file mode 100644
index 00000000000..09953f4d9ec
--- /dev/null
+++
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/process/ProcessSetTableOperatorTest.java
@@ -0,0 +1,424 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.operators.process;
+
+import org.apache.flink.api.common.serialization.SerializerConfigImpl;
+import org.apache.flink.api.common.state.ValueState;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.MultipleInputStreamTask;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskMailboxTestHarness;
+import
org.apache.flink.streaming.runtime.tasks.StreamTaskMailboxTestHarnessBuilder;
+import
org.apache.flink.streaming.util.KeyedMultiInputStreamOperatorTestHarness;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.TableRuntimeException;
+import org.apache.flink.table.api.dataview.ValueView;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.runtime.generated.HashFunction;
+import org.apache.flink.table.runtime.generated.ProcessTableRunner;
+import org.apache.flink.table.runtime.generated.RecordComparator;
+import org.apache.flink.table.runtime.generated.RecordEqualiser;
+import org.apache.flink.table.runtime.keyselector.RowDataKeySelector;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.table.runtime.util.RuntimeChangelogMode;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.utils.HandwrittenSelectorUtil;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.Serializable;
+import java.util.Collection;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Harness tests for {@link ProcessSetTableOperator}.
+ *
+ * <p>The tests use hand-written {@link ProcessTableRunner}s instead of
generated code. This allows
+ * for full control over the order of input rows, watermarks, and state access.
+ */
+class ProcessSetTableOperatorTest {
+
+ private static final DataType MAIN_TYPE =
+ DataTypes.ROW(DataTypes.FIELD("k", DataTypes.BIGINT()));
+
+ // Output layout is driven by PassPartitionKeysCollector /
PassAllCollector in
+ // AbstractProcessTableOperator: [partition-key, PTF-emitted label].
+ private static final DataType OUTPUT_TYPE =
+ DataTypes.ROW(
+ DataTypes.FIELD("k", DataTypes.BIGINT()),
+ DataTypes.FIELD("label", DataTypes.STRING()));
+
+ private static final InternalTypeInfo<RowData> MAIN_TYPE_INFO =
+ InternalTypeInfo.of(MAIN_TYPE.getLogicalType());
+
+ private static final InternalTypeInfo<RowData> OUTPUT_TYPE_INFO =
+ InternalTypeInfo.of(OUTPUT_TYPE.getLogicalType());
+
+ private static final RowDataKeySelector KEY_SELECTOR =
+ HandwrittenSelectorUtil.getRowDataSelector(
+ new int[] {0}, MAIN_TYPE_INFO.toRowFieldTypes());
+
+ private static final RuntimeChangelogMode INSERT_ONLY =
+ RuntimeChangelogMode.serialize(ChangelogMode.insertOnly());
+
+ @Test
+ void testNotifyStatefulSets() throws Exception {
+ try (StreamTaskMailboxTestHarness<RowData> harness =
+ createBroadcastHarness(new
NotifyStatefulSetsOperatorCreator(true, 0))) {
+ processMainRows(harness, 1L, 2L, 1L);
+ // Notifies sets 1 and 2
+ processBroadcastRow(harness, "a");
+ // Notifies sets 1 and 2, which clear their state while iterating
over the keys
+ processBroadcastRow(harness, "clear");
+ // No stateful sets left
+ processBroadcastRow(harness, "b");
+ // Fires timers registered during notification
+ processWatermark(harness, TIMER_TIME, 2);
+
+ final List<String> output = describeOutput(harness);
+ assertThat(output).hasSize(9);
+ assertThat(output.subList(0, 3)).containsExactly("1:count-1",
"2:count-1", "1:count-2");
+ // The order of sets is defined by the state backend
+ assertThat(output.subList(3, 5))
+ .containsExactlyInAnyOrder(
+ "1:notified-2-a-read-only",
"2:notified-1-a-read-only");
+ assertThat(output.subList(5, 7))
+ .containsExactlyInAnyOrder(
+ "1:notified-2-clear-read-only",
"2:notified-1-clear-read-only");
+ assertThat(output.subList(7,
9)).containsExactlyInAnyOrder("1:fired", "2:fired");
+ }
+ }
+
+ @Test
+ void testWithoutNotifyStatefulSets() throws Exception {
+ try (StreamTaskMailboxTestHarness<RowData> harness =
+ createBroadcastHarness(new
NotifyStatefulSetsOperatorCreator(false, 0))) {
+ processMainRows(harness, 1L, 2L, 1L);
+ processBroadcastRow(harness, "a");
+ processWatermark(harness, TIMER_TIME, 2);
+
+ assertThat(describeOutput(harness))
+ .containsExactly("1:count-1", "2:count-1", "1:count-2");
+ }
+ }
+
+ @Test
+ void testNotifyStatefulSetsWithExpiredState() throws Exception {
+ try (KeyedMultiInputStreamOperatorTestHarness<RowData, RowData>
harness =
+ createTtlBroadcastHarness(new
NotifyStatefulSetsOperatorCreator(true, 10))) {
+ harness.setStateTtlProcessingTime(0);
+ processMainRows(harness, 1L);
+ harness.setStateTtlProcessingTime(12);
+ // Notifies set 1 with expired state, reading it cleans it up in
the heap backend
+ processBroadcastRow(harness, "a");
+ // No stateful sets left
+ processBroadcastRow(harness, "b");
+
+ processMainRows(harness, 2L);
+ harness.setStateTtlProcessingTime(24);
+ // Notifies set 2 with expired state, writing it keeps the set
alive
+ processBroadcastRow(harness, "reset");
+ harness.setStateTtlProcessingTime(30);
+ processBroadcastRow(harness, "c");
+
+ assertThat(describeOutput(harness.getOutput()))
+ .containsExactly(
+ "1:count-1",
+ "1:notified-null-a-read-only",
+ "2:count-1",
+ "2:notified-null-reset-read-only",
+ "2:notified-0-c-read-only");
+ }
+ }
+
+ //
--------------------------------------------------------------------------------------------
+ // Helper methods
+ //
--------------------------------------------------------------------------------------------
+
+ /** Creates a {@link ProcessSetTableOperator} for the given operator
parameters. */
+ private interface OperatorCreator extends Serializable {
+ ProcessSetTableOperator create(StreamOperatorParameters<RowData>
parameters);
+ }
+
+ private static StreamTaskMailboxTestHarness<RowData> createHarness(
+ OperatorCreator operatorCreator, InternalTypeInfo<?>...
sideInputTypes)
+ throws Exception {
+ final StreamTaskMailboxTestHarnessBuilder<RowData> builder =
+ new StreamTaskMailboxTestHarnessBuilder<>(
+ MultipleInputStreamTask::new, OUTPUT_TYPE_INFO)
+ .setKeyType(KEY_SELECTOR.getProducedType())
+ .addInput(MAIN_TYPE_INFO, 1, KEY_SELECTOR);
+ for (InternalTypeInfo<?> sideInputType : sideInputTypes) {
+ builder.addInput(sideInputType, 1, null);
+ }
+ return builder.setupOperatorChain(new
TestOperatorFactory(operatorCreator))
+ .name("ptf")
+ .finishForSingletonOperatorChain(
+ OUTPUT_TYPE_INFO.createSerializer(new
SerializerConfigImpl()))
+ .build();
+ }
+
+ private static ProcessSetTableOperator createOperator(
+ StreamOperatorParameters<RowData> parameters,
+ List<RuntimeTableSemantics> tableSemantics,
+ List<RuntimeStateInfo> stateInfos,
+ ProcessTableRunner runner) {
+ final int inputCount = tableSemantics.size();
+ final int stateCount = stateInfos.size();
+ return new ProcessSetTableOperator(
+ parameters,
+ tableSemantics,
+ stateInfos,
+ new RecordComparator[inputCount],
+ runner,
+ new HashFunction[stateCount],
+ new RecordEqualiser[stateCount],
+ INSERT_ONLY,
+ List.of());
+ }
+
+ /** Table with set semantics that is partitioned by its first column. */
+ private static RuntimeTableSemantics mainTableSemantics() {
+ return new RuntimeTableSemantics(
+ "main",
+ 0,
+ MAIN_TYPE,
+ new int[] {0},
+ new int[0],
+ new RuntimeTableSemantics.SortDirection[0],
+ INSERT_ONLY,
+ /* passColumnsThrough */ false,
+ /* hasSetSemantics */ true,
+ /* hasBroadcastSemantics */ false,
+ /* notifiesStatefulSets */ false,
+ /* timeColumn */ -1,
+ /* upsertKeyColumns */ List.of());
+ }
+
+ private static void processMainRows(StreamTaskMailboxTestHarness<RowData>
harness, long... keys)
+ throws Exception {
+ for (long key : keys) {
+ harness.processElement(new StreamRecord<>(GenericRowData.of(key)),
0);
+ }
+ }
+
+ private static void processMainRows(
+ KeyedMultiInputStreamOperatorTestHarness<RowData, RowData>
harness, long... keys)
+ throws Exception {
+ for (long key : keys) {
+ harness.processElement(0, new
StreamRecord<>(GenericRowData.of(key)));
+ }
+ }
+
+ /** Advances the watermark of all inputs. */
+ private static void processWatermark(
+ StreamTaskMailboxTestHarness<RowData> harness, long watermark, int
inputCount)
+ throws Exception {
+ for (int input = 0; input < inputCount; input++) {
+ harness.processElement(new Watermark(watermark), input);
+ }
+ harness.processAll();
+ }
+
+ /** Describes emitted rows as "key:label". */
+ private static List<String>
describeOutput(StreamTaskMailboxTestHarness<RowData> harness) {
+ return describeOutput(harness.getOutput());
+ }
+
+ private static List<String> describeOutput(Collection<Object> output) {
+ return output.stream()
+ .filter(StreamRecord.class::isInstance)
+ .map(r -> (RowData) ((StreamRecord<?>) r).getValue())
+ .map(r -> r.getLong(0) + ":" + r.getString(1))
+ .collect(Collectors.toList());
+ }
+
+ private static class TestOperatorFactory extends
AbstractStreamOperatorFactory<RowData> {
+ private static final long serialVersionUID = 1L;
+
+ private final OperatorCreator operatorCreator;
+
+ TestOperatorFactory(OperatorCreator operatorCreator) {
+ this.operatorCreator = operatorCreator;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <T extends StreamOperator<RowData>> T createStreamOperator(
+ StreamOperatorParameters<RowData> parameters) {
+ return (T) operatorCreator.create(parameters);
+ }
+
+ @Override
+ @SuppressWarnings("rawtypes")
+ public Class<? extends StreamOperator>
getStreamOperatorClass(ClassLoader classLoader) {
+ return ProcessSetTableOperator.class;
+ }
+ }
+
+ //
--------------------------------------------------------------------------------------------
+ // Broadcast
+ //
--------------------------------------------------------------------------------------------
+
+ private static final DataType BROADCAST_TYPE =
+ DataTypes.ROW(DataTypes.FIELD("rule", DataTypes.STRING()));
+
+ private static final InternalTypeInfo<RowData> BROADCAST_TYPE_INFO =
+ InternalTypeInfo.of(BROADCAST_TYPE.getLogicalType());
+
+ private static final long TIMER_TIME = 100L;
+
+ private static StreamTaskMailboxTestHarness<RowData>
createBroadcastHarness(
+ OperatorCreator operatorCreator) throws Exception {
+ return createHarness(operatorCreator, BROADCAST_TYPE_INFO);
+ }
+
+ /**
+ * Creates an operator-level harness instead of a task-level harness,
because only the former
+ * allows controlling the time of state TTL.
+ */
+ private static KeyedMultiInputStreamOperatorTestHarness<RowData, RowData>
+ createTtlBroadcastHarness(OperatorCreator operatorCreator) throws
Exception {
+ final KeyedMultiInputStreamOperatorTestHarness<RowData, RowData>
harness =
+ new KeyedMultiInputStreamOperatorTestHarness<>(
+ new TestOperatorFactory(operatorCreator),
KEY_SELECTOR.getProducedType());
+ harness.setKeySelector(0, KEY_SELECTOR);
+ harness.open();
+ return harness;
+ }
+
+ private static RuntimeTableSemantics broadcastTableSemantics(boolean
notifyStatefulSets) {
+ return new RuntimeTableSemantics(
+ "broadcast",
+ 1,
+ BROADCAST_TYPE,
+ new int[0],
+ new int[0],
+ new RuntimeTableSemantics.SortDirection[0],
+ INSERT_ONLY,
+ /* passColumnsThrough */ false,
+ /* hasSetSemantics */ false,
+ /* hasBroadcastSemantics */ true,
+ notifyStatefulSets,
+ /* timeColumn */ -1,
+ /* upsertKeyColumns */ List.of());
+ }
+
+ private static void processBroadcastRow(
+ StreamTaskMailboxTestHarness<RowData> harness, String rule) throws
Exception {
+ harness.processElement(
+ new
StreamRecord<>(GenericRowData.of(StringData.fromString(rule))), 1);
+ }
+
+ private static void processBroadcastRow(
+ KeyedMultiInputStreamOperatorTestHarness<RowData, RowData>
harness, String rule)
+ throws Exception {
+ harness.processElement(
+ 1, new
StreamRecord<>(GenericRowData.of(StringData.fromString(rule))));
+ }
+
+ /** Operator for {@link NotifyStatefulSetsRunner} with a keyed count and a
broadcast rule. */
+ private static class NotifyStatefulSetsOperatorCreator implements
OperatorCreator {
+ private static final long serialVersionUID = 1L;
+
+ private final boolean notifyStatefulSets;
+ private final long countTimeToLive;
+
+ NotifyStatefulSetsOperatorCreator(boolean notifyStatefulSets, long
countTimeToLive) {
+ this.notifyStatefulSets = notifyStatefulSets;
+ this.countTimeToLive = countTimeToLive;
+ }
+
+ @Override
+ public ProcessSetTableOperator
create(StreamOperatorParameters<RowData> parameters) {
+ return createOperator(
+ parameters,
+ List.of(mainTableSemantics(),
broadcastTableSemantics(notifyStatefulSets)),
+ List.of(
+ new RuntimeStateInfo(
+ "count",
+
ValueView.newValueViewDataType(DataTypes.INT()),
+ countTimeToLive,
+ false),
+ new RuntimeStateInfo(
+ "rule",
+
ValueView.newValueViewDataType(DataTypes.STRING()),
+ 0,
+ true)),
+ new NotifyStatefulSetsRunner());
+ }
+ }
+
+ /**
+ * {@link ProcessTableRunner} that counts rows of the main table per set.
A broadcast row
+ * updates a broadcast rule. If notified, every stateful set emits its
count together with the
+ * rule and registers a timer. The rule "clear" clears the set's state,
the rule "reset" resets
+ * the set's count. State handles are accessed directly instead of
converting state like
+ * generated code does.
+ */
+ private static class NotifyStatefulSetsRunner extends ProcessTableRunner {
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public void callEval() throws Exception {
+ final ValueState<Integer> count = (ValueState<Integer>)
stateHandles[0].getState();
+ final ValueState<String> rule = (ValueState<String>)
stateHandles[1].getState();
+ if (inputIndex == 0) {
+ final int c = Optional.ofNullable(count.value()).orElse(0) + 1;
+ count.update(c);
+ emit("count-" + c);
+ } else if (isProcessingBroadcast()) {
+ rule.update(inputRow.getString(0).toString());
+ } else {
+ String label = "notified-" + count.value() + "-" +
rule.value();
+ try {
+ rule.update("invalid");
+ } catch (TableRuntimeException e) {
+ label += "-read-only";
+ }
+ emit(label);
+
runnerContext.timeContext(Long.class).registerOnTime(TIMER_TIME);
+ if (rule.value().equals("clear")) {
+ count.clear();
+ } else if (rule.value().equals("reset")) {
+ count.update(0);
+ }
+ }
+ }
+
+ @Override
+ public void callOnTimer() {
+
onTimerCollector.collect(GenericRowData.of(StringData.fromString("fired")));
+ }
+
+ private void emit(String label) {
+
evalCollector.collect(GenericRowData.of(StringData.fromString(label)));
+ }
+ }
+}
