raminqaf commented on code in PR #29525:
URL: https://github.com/apache/flink/pull/29525#discussion_r4315836547
##########
docs/content.zh/docs/dev/table/functions/ptfs.md:
##########
@@ -1471,6 +1471,88 @@ 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. Timers are ignored.
+- During notification, state entries of the set and timers are accessible and
broadcast state is read-only. Similar to processing a row
Review Comment:
"Timers are ignored" contradicts "timers are accessible"
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/SystemTypeInference.java:
##########
@@ -277,17 +277,17 @@ private static void
checkMultipleTableArgs(List<StaticArgument> staticArgs) {
}
}
+ private static void checkStatefulBroadcast(
Review Comment:
How about?
```suggestion
private static void checkBroadcastTraits(
```
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/BroadcastStateAdapters.java:
##########
@@ -218,8 +218,8 @@ private static <T> T rethrow(SupplierWithException<T,
Exception> stateAccess)
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 is only
writable when "
+ + "processing a table with broadcast
semantics.",
Review Comment:
This error message is a bi ambiguous to me
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/ArgumentTrait.java:
##########
@@ -109,6 +111,36 @@ 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. Timers are ignored. 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.
+ */
+ NOTIFY_STATEFUL_SETS(false, StaticArgumentTrait.NOTIFY_STATEFUL_SETS),
Review Comment:
Should `NOTIFY_STATEFUL_SETS` also require a set semantic table?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
