This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch release-2.4
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-2.4 by this push:
new eb107bf98e5 [FLINK-41244][table-runtime] Scope clearing of unnamed PTF
timers to the current key
eb107bf98e5 is described below
commit eb107bf98e5bf84f2de69d2bd8989538365982a8
Author: Timo Walther <[email protected]>
AuthorDate: Mon Nov 2 12:03:19 2026 +0100
[FLINK-41244][table-runtime] Scope clearing of unnamed PTF timers to the
current key
This closes #29669.
---
.../stream/ProcessTableFunctionSemanticTests.java | 1 +
.../stream/ProcessTableFunctionTestPrograms.java | 33 +++++++++++
.../exec/stream/ProcessTableFunctionTestUtils.java | 25 +++++++++
.../process/AbstractProcessTableOperator.java | 19 ++++++-
.../process/WritableInternalTimeContext.java | 19 ++++++-
.../process/ProcessSetTableOperatorTest.java | 65 +++++++++++++++++++++-
6 files changed, 158 insertions(+), 4 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 1faf06beb60..17f2fd27bcf 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
@@ -109,6 +109,7 @@ public class ProcessTableFunctionSemanticTests extends
SemanticTestBase {
ProcessTableFunctionTestPrograms.PROCESS_MULTI_INPUT_WITH_SCALAR_ARGS,
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,
ProcessTableFunctionTestPrograms.PROCESS_ORDER_BY,
ProcessTableFunctionTestPrograms.PROCESS_MULTI_INPUT_ORDER_BY,
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 d2512eb0e54..966fa26df0a 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
@@ -30,6 +30,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.ComplexValueViewFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ContextFunction;
@@ -2119,6 +2120,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 1122c322a0e..81ca2f213bc 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
@@ -1266,6 +1266,31 @@ public class ProcessTableFunctionTestUtils {
}
}
+ /** Testing function. */
+ public static class ClearAllUnnamedTimersPerKeyFunction extends
AppendProcessTableFunctionBase {
+ public void eval(
+ Context ctx,
+ @StateHint ValueView<Integer> count,
+ @ArgumentHint({SET_SEMANTIC_TABLE, REQUIRE_ON_TIME}) Row r)
+ throws Exception {
+ final TimeContext<Long> timeCtx = ctx.timeContext(Long.class);
+ final int newCount = count.getValue() == null ? 1 :
count.getValue() + 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.setValue(newCount);
+ collect(Row.of("count " + newCount));
+ }
+
+ public void onTimer(OnTimerContext ctx, ValueView<Integer> count)
throws Exception {
+ collect(Row.of("timeout with count " + count.getValue()));
+ }
+ }
+
/**
* 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 b3906336968..22ba7019652 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
@@ -49,6 +49,7 @@ import org.apache.flink.table.api.dataview.ValueView;
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;
@@ -185,6 +186,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);
+ }
+
/**
* 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
@@ -362,7 +376,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 ce97c630c80..090647d6ae2 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;
@@ -111,11 +116,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));
diff --git
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/process/ProcessSetTableOperatorTest.java
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/process/ProcessSetTableOperatorTest.java
index 09953f4d9ec..b0e2332b247 100644
---
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/process/ProcessSetTableOperatorTest.java
+++
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/process/ProcessSetTableOperatorTest.java
@@ -55,6 +55,7 @@ import java.util.Optional;
import java.util.stream.Collectors;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/**
* Harness tests for {@link ProcessSetTableOperator}.
@@ -128,6 +129,18 @@ class ProcessSetTableOperatorTest {
}
}
+ @Test
+ void testNonBinaryKey() throws Exception {
+ try (StreamTaskMailboxTestHarness<RowData> harness =
+ createHarness(new SingleInputOperatorCreator(), new
GenericKeySelector())) {
+ assertThatThrownBy(() -> processMainRows(harness, 1L))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage(
+ "Partition keys must be BinaryRowData but was: "
+ + GenericRowData.class.getName());
+ }
+ }
+
@Test
void testNotifyStatefulSetsWithExpiredState() throws Exception {
try (KeyedMultiInputStreamOperatorTestHarness<RowData, RowData>
harness =
@@ -169,11 +182,19 @@ class ProcessSetTableOperatorTest {
private static StreamTaskMailboxTestHarness<RowData> createHarness(
OperatorCreator operatorCreator, InternalTypeInfo<?>...
sideInputTypes)
throws Exception {
+ return createHarness(operatorCreator, KEY_SELECTOR, sideInputTypes);
+ }
+
+ private static StreamTaskMailboxTestHarness<RowData> createHarness(
+ OperatorCreator operatorCreator,
+ RowDataKeySelector keySelector,
+ InternalTypeInfo<?>... sideInputTypes)
+ throws Exception {
final StreamTaskMailboxTestHarnessBuilder<RowData> builder =
new StreamTaskMailboxTestHarnessBuilder<>(
MultipleInputStreamTask::new, OUTPUT_TYPE_INFO)
- .setKeyType(KEY_SELECTOR.getProducedType())
- .addInput(MAIN_TYPE_INFO, 1, KEY_SELECTOR);
+ .setKeyType(keySelector.getProducedType())
+ .addInput(MAIN_TYPE_INFO, 1, keySelector);
for (InternalTypeInfo<?> sideInputType : sideInputTypes) {
builder.addInput(sideInputType, 1, null);
}
@@ -259,6 +280,46 @@ class ProcessSetTableOperatorTest {
.collect(Collectors.toList());
}
+ /** Key selector that violates the requirement of binary keys. */
+ private static class GenericKeySelector implements RowDataKeySelector {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public RowData getKey(RowData value) {
+ return GenericRowData.of(value.getLong(0));
+ }
+
+ @Override
+ public InternalTypeInfo<RowData> getProducedType() {
+ return KEY_SELECTOR.getProducedType();
+ }
+
+ @Override
+ public RowDataKeySelector copy() {
+ return new GenericKeySelector();
+ }
+ }
+
+ /** Operator creator for a single main table that never evaluates rows. */
+ private static class SingleInputOperatorCreator implements OperatorCreator
{
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public ProcessSetTableOperator
create(StreamOperatorParameters<RowData> parameters) {
+ return createOperator(
+ parameters,
+ List.of(mainTableSemantics()),
+ List.of(),
+ new ProcessTableRunner() {
+ @Override
+ public void callEval() {}
+
+ @Override
+ public void callOnTimer() {}
+ });
+ }
+ }
+
private static class TestOperatorFactory extends
AbstractStreamOperatorFactory<RowData> {
private static final long serialVersionUID = 1L;
