twalthr commented on code in PR #29657:
URL: https://github.com/apache/flink/pull/29657#discussion_r4403051593
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/WritableInternalTimeContext.java:
##########
@@ -111,11 +116,23 @@ private void deleteUnnamedTimer(long time) {
}
private void clearUnnamedTimers() {
+ final RowData currentKey = keyedStateBackend.getCurrentKey();
final List<Long> unnamedTimers = new ArrayList<>();
try {
- unnamedTimerService.forEachEventTimeTimer((namespace, time) ->
unnamedTimers.add(time));
+ // The timer service iterates the timers of all keys and sets the
key context to the
+ // key of each visited timer. Only timers of the current key must
be cleared.
+ // The equals is safe: keys are BinaryRowData, which compares the
serialized bytes and
+ // not the object identity, so a key deserialized by the timer
service still matches.
+ unnamedTimerService.forEachEventTimeTimer(
Review Comment:
This is what AI proposed to me as well. It this means more computational and
storage overhead plus state migration for existing PTFs. Not a valid option.
Unnamed timers should give you a performance benefit, even if clearing all of
them has downsides.
--
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]
