This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 7bd618a7910 [FLINK-40571][table-planner] Fix flaky
process-stateful-multi-input-with-timeout PTF test (#29352)
7bd618a7910 is described below
commit 7bd618a7910b21f010f107fb5c8347ac1c1f4e40
Author: Purushottam Sinha <[email protected]>
AuthorDate: Mon Oct 12 15:08:38 2026 +0530
[FLINK-40571][table-planner] Fix flaky
process-stateful-multi-input-with-timeout PTF test (#29352)
* [FLINK-40571][table-planner] Fix flaky
process-stateful-multi-input-with-timeout PTF test
The two inputs of the PTF are processed in arrival order, not event-time
order. TimedJoinFunction keeps a single score per key, so if the city row
arrived after several scores, the earlier scores were overwritten and only
one join row was emitted at the city row's timestamp. The pinned output
only matched the city-first interleaving.
Restructure the test data so the output is independent of the
interleaving: a key with a city has a single score with the same timestamp
as its city, while keys without a city may have multiple scores. Timeouts
stay deterministic as timers only fire once both inputs have finished.
Generated-by: Claude Code (claude-opus-5-5)
* [FLINK-40571][table-planner] Add single-input PTF test for clearAllTimers
keeping state
The reworked multi-input timeout test no longer has a key receiving an
event after its join, so nothing verified that ctx.clearAllTimers()
leaves state intact. Add a dedicated single-input program for it and
inline the city source into the only program that uses it.
Generated-by: Claude Code (claude-opus-5-5)
---
.../stream/ProcessTableFunctionSemanticTests.java | 1 +
.../stream/ProcessTableFunctionTestPrograms.java | 69 +++++++++++++++++++---
.../exec/stream/ProcessTableFunctionTestUtils.java | 31 ++++++----
3 files changed, 82 insertions(+), 19 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 d8f5e8294b8..bbf6a8644c2 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
@@ -107,6 +107,7 @@ public class ProcessTableFunctionSemanticTests extends
SemanticTestBase {
ProcessTableFunctionTestPrograms.PROCESS_INVALID_BROADCAST_STATE_WRITE,
ProcessTableFunctionTestPrograms.PROCESS_MULTI_INPUT_WITH_SCALAR_ARGS,
ProcessTableFunctionTestPrograms.PROCESS_STATEFUL_MULTI_INPUT_WITH_TIMEOUT,
+
ProcessTableFunctionTestPrograms.PROCESS_CLEAR_ALL_TIMERS_KEEPS_STATE,
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 d91847a1710..9c7ebe85980 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
@@ -29,6 +29,7 @@ import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctio
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.BroadcastTimersFunction;
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.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;
@@ -118,7 +119,6 @@ import static
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTable
import static
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.PASS_THROUGH_BASE_SINK_SCHEMA;
import static
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TIMED_BASE_SINK_SCHEMA;
import static
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TIMED_BROADCAST_RULES_SOURCE;
-import static
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TIMED_CITY_SOURCE;
import static
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TIMED_MULTI_BASE_SINK_SCHEMA;
import static
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TIMED_SOURCE;
import static
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TIMED_SOURCE_LATE_EVENTS;
@@ -1979,19 +1979,42 @@ public class ProcessTableFunctionTestPrograms {
"process-stateful-multi-input-with-timeout",
"joins two tables and emits the left side after a
timeout if there is no right side")
.setupTemporarySystemFunction("f", TimedJoinFunction.class)
- .setupTableSource(TIMED_SOURCE)
- .setupTableSource(TIMED_CITY_SOURCE)
+ // The two inputs may reach eval() in any interleaving. To
keep the output
+ // independent of it, a key with a city has a single score
with the same
+ // timestamp as its city. Keys without a city may have
multiple scores.
+ .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", 3,
Instant.ofEpochMilli(2)),
+ Row.of("Alice", 2,
Instant.ofEpochMilli(3)),
+ Row.of("Dave", 4,
Instant.ofEpochMilli(4)),
+ Row.of("Frank", 5,
Instant.ofEpochMilli(6)))
+ .build())
+ .setupTableSource(
+ SourceTestStep.newBuilder("city")
+ .addSchema(
+ "name STRING",
+ "city STRING",
+ "ts TIMESTAMP_LTZ(3)",
+ "WATERMARK FOR ts AS ts - INTERVAL
'0.001' SECOND")
+ .producedValues(
+ Row.of("Bob", "London",
Instant.ofEpochMilli(0)),
+ Row.of("Charly", "Paris",
Instant.ofEpochMilli(2)),
+ Row.of("Dave", "Berlin",
Instant.ofEpochMilli(4)),
+ Row.of("Eve", "Rome",
Instant.ofEpochMilli(5)))
+ .build())
.setupTableSink(
SinkTestStep.newBuilder("sink")
.addSchema(TIMED_MULTI_BASE_SINK_SCHEMA)
.consumedValues(
"+I[Bob, Bob, 1 score in city
London, 1970-01-01T00:00:00Z]",
- "+I[Bob, Bob, 2 score in city
London, 1970-01-01T00:00:00.002Z]",
- "+I[Bob, Bob, 3 score in city
London, 1970-01-01T00:00:00.003Z]",
- "+I[Bob, Bob, 4 score in city
London, 1970-01-01T00:00:00.004Z]",
- "+I[Bob, Bob, 5 score in city
London, 1970-01-01T00:00:00.005Z]",
- "+I[Bob, Bob, 6 score in city
London, 1970-01-01T00:00:00.006Z]",
- "+I[Alice, Alice, no city found
for score 1, 1970-01-01T00:00:01.001Z]")
+ "+I[Charly, Charly, 3 score in
city Paris, 1970-01-01T00:00:00.002Z]",
+ "+I[Dave, Dave, 4 score in city
Berlin, 1970-01-01T00:00:00.004Z]",
+ "+I[Alice, Alice, no city found
for score 2, 1970-01-01T00:00:01.003Z]",
+ "+I[Frank, Frank, no city found
for score 5, 1970-01-01T00:00:01.006Z]")
.build())
.runSql(
"INSERT INTO sink SELECT * FROM f("
@@ -2000,6 +2023,34 @@ public class ProcessTableFunctionTestPrograms {
+ "on_time => DESCRIPTOR(ts))")
.build();
+ public static final TableTestProgram PROCESS_CLEAR_ALL_TIMERS_KEEPS_STATE =
+ TableTestProgram.of(
+ "process-clear-all-timers-keeps-state",
+ "clearing all timers does not clear the state")
+ .setupTemporarySystemFunction("f",
ClearAllTimersKeepsStateFunction.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("Bob", 2,
Instant.ofEpochMilli(2)),
+ Row.of("Bob", 3,
Instant.ofEpochMilli(3)))
+ .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[Bob, count 2,
1970-01-01T00:00:00.002Z]",
+ "+I[Bob, count 3,
1970-01-01T00:00:00.003Z]",
+ "+I[Alice, timeout with count 1,
1970-01-01T00:00:01.001Z]")
+ .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 4e247a92a8a..1e5dad159c3 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
@@ -100,16 +100,6 @@ public class ProcessTableFunctionTestUtils {
"CREATE VIEW city AS SELECT * FROM "
+ "(VALUES ('Bob', 'London'), ('Alice', 'Berlin'),
('Charly', 'Paris')) AS T(name, city)";
- public static final SourceTestStep TIMED_CITY_SOURCE =
- SourceTestStep.newBuilder("city")
- .addSchema(
- "name STRING",
- "city STRING",
- "ts TIMESTAMP_LTZ(3)",
- "WATERMARK FOR ts AS ts - INTERVAL '0.001' SECOND")
- .producedValues(Row.of("Bob", "London",
Instant.ofEpochMilli(0)))
- .build();
-
public static final String UPDATING_VALUES =
"CREATE VIEW t AS SELECT name, COUNT(*) FROM "
+ "(VALUES ('Bob', 12), ('Alice', 42), ('Bob', 14)) AS
T(name, score) "
@@ -1254,6 +1244,27 @@ public class ProcessTableFunctionTestUtils {
}
}
+ /** Testing function. */
+ public static class ClearAllTimersKeepsStateFunction extends
AppendProcessTableFunctionBase {
+ public void eval(
+ Context ctx,
+ @StateHint Tuple1<Integer> count,
+ @ArgumentHint({SET_SEMANTIC_TABLE, REQUIRE_ON_TIME}) Row r) {
+ final TimeContext<Instant> timeCtx =
ctx.timeContext(Instant.class);
+ count.f0 = count.f0 == null ? 1 : count.f0 + 1;
+ if (count.f0 == 1) {
+ timeCtx.registerOnTime("timeout",
timeCtx.time().plusMillis(1000));
+ } else if (count.f0 == 2) {
+ ctx.clearAllTimers();
+ }
+ collect(Row.of("count " + count.f0));
+ }
+
+ 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
