This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch release-2.2
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-2.2 by this push:
new a9d367e0fd5 [FLINK-41244][table-runtime] Scope clearing of unnamed PTF
timers to the current key
a9d367e0fd5 is described below
commit a9d367e0fd5275d7b934fcd26f0985d688b5c934
Author: Timo Walther <[email protected]>
AuthorDate: Mon Nov 2 12:02:34 2026 +0100
[FLINK-41244][table-runtime] Scope clearing of unnamed PTF timers to the
current key
This closes #29672.
---
.../stream/ProcessTableFunctionSemanticTests.java | 1 +
.../stream/ProcessTableFunctionTestPrograms.java | 33 ++++++++++++++++++++++
.../exec/stream/ProcessTableFunctionTestUtils.java | 24 ++++++++++++++++
.../process/AbstractProcessTableOperator.java | 19 ++++++++++++-
.../process/WritableInternalTimeContext.java | 19 ++++++++++++-
5 files changed, 94 insertions(+), 2 deletions(-)
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 3b260344978..5a02b4ac966 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
@@ -98,6 +98,7 @@ public class ProcessTableFunctionSemanticTests extends
SemanticTestBase {
ProcessTableFunctionTestPrograms.PROCESS_MULTI_INPUT,
ProcessTableFunctionTestPrograms.PROCESS_STATEFUL_MULTI_INPUT_WITH_TIMEOUT,
ProcessTableFunctionTestPrograms.PROCESS_CLEAR_ALL_TIMERS_KEEPS_STATE,
+
ProcessTableFunctionTestPrograms.PROCESS_CLEAR_ALL_UNNAMED_TIMERS_PER_KEY,
ProcessTableFunctionTestPrograms.PROCESS_UPDATING_MULTI_INPUT);
}
}
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 ad776fd0898..9af9ce7e8ff 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
@@ -25,6 +25,7 @@ import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctio
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ChainedReceivingFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ChainedSendingFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ClearAllTimersKeepsStateFunction;
+import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ClearAllUnnamedTimersPerKeyFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ClearStateFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ContextFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.DescriptorFunction;
@@ -1497,6 +1498,38 @@ public class ProcessTableFunctionTestPrograms {
"INSERT INTO sink SELECT * FROM f(r => TABLE t
PARTITION BY name, on_time => DESCRIPTOR(ts))")
.build();
+ public static final TableTestProgram
PROCESS_CLEAR_ALL_UNNAMED_TIMERS_PER_KEY =
+ TableTestProgram.of(
+ "process-clear-all-unnamed-timers-per-key",
+ "clearing all timers only affects the timers of
the current key")
+ .setupTemporarySystemFunction("f",
ClearAllUnnamedTimersPerKeyFunction.class)
+ .setupTableSource(
+ SourceTestStep.newBuilder("t")
+ .addSchema(TIMED_SOURCE_SCHEMA)
+ .producedValues(
+ Row.of("Bob", 1,
Instant.ofEpochMilli(0)),
+ Row.of("Alice", 1,
Instant.ofEpochMilli(1)),
+ Row.of("Charly", 1,
Instant.ofEpochMilli(2)),
+ Row.of("Bob", 2,
Instant.ofEpochMilli(3)),
+ Row.of("Alice", 2,
Instant.ofEpochMilli(4)),
+ Row.of("Bob", 3,
Instant.ofEpochMilli(5)))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(KEYED_TIMED_BASE_SINK_SCHEMA)
+ .consumedValues(
+ "+I[Bob, count 1,
1970-01-01T00:00:00Z]",
+ "+I[Alice, count 1,
1970-01-01T00:00:00.001Z]",
+ "+I[Charly, count 1,
1970-01-01T00:00:00.002Z]",
+ "+I[Bob, count 2,
1970-01-01T00:00:00.003Z]",
+ "+I[Alice, count 2,
1970-01-01T00:00:00.004Z]",
+ "+I[Bob, count 3,
1970-01-01T00:00:00.005Z]",
+ "+I[Charly, timeout with count 1,
1970-01-01T00:00:01Z]")
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM f(r => TABLE t
PARTITION BY name, on_time => DESCRIPTOR(ts))")
+ .build();
+
public static final TableTestProgram PROCESS_UPDATING_MULTI_INPUT =
TableTestProgram.of(
"process-updating-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 7b1c9a511a7..6de1cf6ce84 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
@@ -966,6 +966,30 @@ public class ProcessTableFunctionTestUtils {
}
}
+ /** Testing function. */
+ public static class ClearAllUnnamedTimersPerKeyFunction extends
AppendProcessTableFunctionBase {
+ public void eval(
+ Context ctx,
+ @StateHint Tuple1<Integer> count,
+ @ArgumentHint({SET_SEMANTIC_TABLE, REQUIRE_ON_TIME}) Row r) {
+ final TimeContext<Long> timeCtx = ctx.timeContext(Long.class);
+ final int newCount = count.f0 == null ? 1 : count.f0 + 1;
+ if (newCount == 1) {
+ // All keys register an unnamed timer for the same time
+ timeCtx.registerOnTime(1000L);
+ } else if (newCount == 2) {
+ ctx.clearAllTimers();
+ }
+ // Written to the state of the current key even after clearing
+ count.f0 = newCount;
+ collect(Row.of("count " + newCount));
+ }
+
+ public void onTimer(OnTimerContext ctx, Tuple1<Integer> count) {
+ collect(Row.of("timeout with count " + count.f0));
+ }
+ }
+
/**
* Implements a custom join that acts like kind of an outer join and never
produces deletions.
* Both the score and city can change at any time. The join will output an
update if a matching
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 e9f0e005640..aa7ee630ad4 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
@@ -45,6 +45,7 @@ import org.apache.flink.table.api.dataview.MapView;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.binary.BinaryRowData;
import org.apache.flink.table.functions.ProcessTableFunction;
import org.apache.flink.table.functions.ProcessTableFunction.TimeContext;
import org.apache.flink.table.functions.TableSemantics;
@@ -165,6 +166,19 @@ public abstract class AbstractProcessTableOperator extends
AbstractStreamOperato
@Override
public void onProcessingTime(InternalTimer<RowData, Object> timer) throws
Exception {}
+ /**
+ * Every key enters the operator through this method. Timers and state
compare keys via {@link
+ * Object#equals(Object)}, which is only guaranteed to have value
semantics for binary keys.
+ */
+ @Override
+ public void setCurrentKey(Object key) {
+ if (!(key instanceof BinaryRowData)) {
+ throw new IllegalStateException(
+ "Partition keys must be BinaryRowData but was: " +
key.getClass().getName());
+ }
+ super.setCurrentKey(key);
+ }
+
/** Implementation of {@link ProcessTableFunction.Context}. */
@Internal
public class RunnerContext implements ProcessTableFunction.Context {
@@ -309,7 +323,10 @@ public abstract class AbstractProcessTableOperator extends
AbstractStreamOperato
if (shouldEnableTimers()) {
internalTimeContext =
new WritableInternalTimeContext(
- namedTimersMapState, namedTimerService,
unnamedTimerService);
+ getKeyedStateBackend(),
+ namedTimersMapState,
+ namedTimerService,
+ unnamedTimerService);
} else {
internalTimeContext = new ReadableInternalTimeContext();
}
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/WritableInternalTimeContext.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/WritableInternalTimeContext.java
index 5fd996fecaa..73946dc25b0 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/WritableInternalTimeContext.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/WritableInternalTimeContext.java
@@ -20,8 +20,10 @@ package org.apache.flink.table.runtime.operators.process;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.state.MapState;
+import org.apache.flink.runtime.state.KeyedStateBackend;
import org.apache.flink.runtime.state.VoidNamespace;
import org.apache.flink.streaming.api.operators.InternalTimerService;
+import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
import org.apache.flink.table.functions.ProcessTableFunction.TimeContext;
import org.apache.flink.util.FlinkRuntimeException;
@@ -37,14 +39,17 @@ import java.util.Map;
@Internal
class WritableInternalTimeContext extends ReadableInternalTimeContext {
+ private final KeyedStateBackend<RowData> keyedStateBackend;
private final MapState<StringData, Long> namedTimersMapState;
private final InternalTimerService<StringData> namedTimerService;
private final InternalTimerService<VoidNamespace> unnamedTimerService;
WritableInternalTimeContext(
+ KeyedStateBackend<RowData> keyedStateBackend,
MapState<StringData, Long> namedTimersMapState,
InternalTimerService<StringData> namedTimerService,
InternalTimerService<VoidNamespace> unnamedTimerService) {
+ this.keyedStateBackend = keyedStateBackend;
this.namedTimersMapState = namedTimersMapState;
this.namedTimerService = namedTimerService;
this.unnamedTimerService = unnamedTimerService;
@@ -116,11 +121,23 @@ class WritableInternalTimeContext extends
ReadableInternalTimeContext {
}
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(
+ (namespace, time) -> {
+ if
(currentKey.equals(keyedStateBackend.getCurrentKey())) {
+ unnamedTimers.add(time);
+ }
+ });
} catch (Exception e) {
throw new FlinkRuntimeException(e);
+ } finally {
+ keyedStateBackend.setCurrentKey(currentKey);
}
unnamedTimers.forEach(
time ->
unnamedTimerService.deleteEventTimeTimer(VoidNamespace.INSTANCE, time));
