twalthr commented on code in PR #29525:
URL: https://github.com/apache/flink/pull/29525#discussion_r4323615636
##########
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:
Good catch @MartijnVisser. getKeys isn't TTL-aware, so expired but not yet
cleaned up sets can be notified. A TTL-aware existence check would add extra
state reads for every key, also for lazily accessed state the function might
not touch. I would suggest that (similar to applyToKeyedState in the DataStream
API), we documented the behavior instead. I added it in ptfs.md and the
NOTIFY_STATEFUL_SETS JavaDoc.
--
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]
