MartijnVisser commented on code in PR #29525:
URL: https://github.com/apache/flink/pull/29525#discussion_r4315933397
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/AbstractProcessTableOperator.java:
##########
@@ -181,6 +185,27 @@ public void onEventTime(InternalTimer<RowData, Object>
timer) throws Exception {
@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)) {
Review Comment:
I think this also returns sets whose TTL has expired but that aren't cleaned
up yet. Locally they were notified with empty state, and a write during
notification revived them. Is that intended?
##########
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
Review Comment:
`Timers are ignored` reads as if timers can't be used during notification,
which the next sentence contradicts. I would prefer `Registered timers alone
don't count as state`, also in `ptfs.md`.
--
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]
