This is an automated email from the ASF dual-hosted git repository.
twalthr 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 61ababd42fa [FLINK-40708][table] Add semantic test for ValueView state
in onTimer
61ababd42fa is described below
commit 61ababd42fa0d0b1189b20d5c86bbeb1a5811fee
Author: Timo Walther <[email protected]>
AuthorDate: Fri Oct 2 17:12:02 2026 +0200
[FLINK-40708][table] Add semantic test for ValueView state in onTimer
This closes #29364.
---
.../stream/ProcessTableFunctionSemanticTests.java | 1 +
.../stream/ProcessTableFunctionTestPrograms.java | 27 ++++++++++++++++++
.../exec/stream/ProcessTableFunctionTestUtils.java | 32 ++++++++++++++++++++++
3 files changed, 60 insertions(+)
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 bb893881a74..728ae11cb12 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
@@ -84,6 +84,7 @@ public class ProcessTableFunctionSemanticTests extends
SemanticTestBase {
ProcessTableFunctionTestPrograms.PROCESS_OPTIONAL_PARTITION_BY_TIME,
ProcessTableFunctionTestPrograms.PROCESS_OPTIONAL_ON_TIME,
ProcessTableFunctionTestPrograms.PROCESS_POJO_STATE_TIME,
+
ProcessTableFunctionTestPrograms.PROCESS_EAGER_AND_VALUE_VIEW_STATE_TIME,
ProcessTableFunctionTestPrograms.PROCESS_CHAINED_TIME,
ProcessTableFunctionTestPrograms.PROCESS_CHAINED_TIME_TABLE_API,
ProcessTableFunctionTestPrograms.PROCESS_INVALID_ROW_SEMANTIC_TABLE_TIMERS,
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 72a051edab3..e846ef52ac6 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
@@ -28,6 +28,7 @@ import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctio
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ComplexValueViewFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ContextFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.DescriptorFunction;
+import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.EagerAndValueViewStateTimeFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.EmptyArgFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ImplicitCastingFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.IntervalDayArgFunction;
@@ -1398,6 +1399,32 @@ 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_EAGER_AND_VALUE_VIEW_STATE_TIME =
+ TableTestProgram.of(
+ "process-eager-and-value-view-state-time",
+ "eager value state and value view accessed from
both eval() and onTimer()")
+ .setupTemporarySystemFunction("f",
EagerAndValueViewStateTimeFunction.class)
+ .setupTableSource(TIMED_SOURCE)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(KEYED_TIMED_BASE_SINK_SCHEMA)
+ .consumedValues(
+ "+I[Bob, {Score(s='null', i=null),
null, +I[Bob, 1, 1970-01-01T00:00:00Z]}, 1970-01-01T00:00:00Z]",
+ "+I[Bob, {Registering timer t for
2 at time 0 watermark null}, 1970-01-01T00:00:00Z]",
+ "+I[Alice, {Score(s='null',
i=null), null, +I[Alice, 1, 1970-01-01T00:00:00.001Z]},
1970-01-01T00:00:00.001Z]",
+ "+I[Alice, {Registering timer t
for 3 at time 1 watermark -1}, 1970-01-01T00:00:00.001Z]",
+ "+I[Bob, {Score(s='null', i=1), 1,
+I[Bob, 2, 1970-01-01T00:00:00.002Z]}, 1970-01-01T00:00:00.002Z]",
+ "+I[Bob, {Score(s='null', i=2), 2,
+I[Bob, 3, 1970-01-01T00:00:00.003Z]}, 1970-01-01T00:00:00.003Z]",
+ "+I[Bob, {Timer t fired at time 2
watermark 2}, 1970-01-01T00:00:00.002Z]",
+ "+I[Bob, {Score(s='null', i=30),
103, +I[Bob, 4, 1970-01-01T00:00:00.004Z]}, 1970-01-01T00:00:00.004Z]",
+ "+I[Alice, {Timer t fired at time
3 watermark 3}, 1970-01-01T00:00:00.003Z]",
+ "+I[Bob, {Score(s='null', i=31),
104, +I[Bob, 5, 1970-01-01T00:00:00.005Z]}, 1970-01-01T00:00:00.005Z]",
+ "+I[Bob, {Score(s='null', i=32),
105, +I[Bob, 6, 1970-01-01T00:00:00.006Z]}, 1970-01-01T00:00:00.006Z]")
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM f(r => TABLE t
PARTITION BY name, on_time => DESCRIPTOR(ts))")
+ .build();
+
public static final TableTestProgram PROCESS_CHAINED_TIME =
TableTestProgram.of(
"process-chained-time",
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 330cce0346d..7808afc4d58 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
@@ -780,6 +780,38 @@ public class ProcessTableFunctionTestUtils {
}
}
+ /**
+ * Testing function that combines eager value state (a POJO) and a {@link
ValueView} and
+ * accesses both from eval() and onTimer().
+ */
+ public static class EagerAndValueViewStateTimeFunction extends
AppendProcessTableFunctionBase {
+ public void eval(
+ Context ctx,
+ @StateHint Score eager,
+ @StateHint ValueView<Integer> view,
+ @ArgumentHint({SET_SEMANTIC_TABLE, REQUIRE_ON_TIME}) Row r) {
+ final TimeContext<Long> timeCtx = ctx.timeContext(Long.class);
+ collectObjects(eager, view.getValue(), r);
+ if (eager.i == null) {
+ eager.i = 1;
+ collectCreateTimer(timeCtx, "t", timeCtx.time() + 2);
+ } else {
+ eager.i += 1;
+ }
+ final Integer count = view.getValue();
+ view.setValue(count == null ? 1 : count + 1);
+ }
+
+ public void onTimer(OnTimerContext ctx, Score eager,
ValueView<Integer> view) {
+ collectOnTimerEvent(ctx);
+ // Mutate both eager value state and the value view; the changes
must be persisted and
+ // visible to the next eval() call. Both are guaranteed to be
non-null here because the
+ // timer is only registered in the first eval() call, which also
initializes them.
+ eager.i *= 10;
+ view.setValue(view.getValue() + 100);
+ }
+ }
+
/** Testing function. */
public static class ChainedSendingFunction extends
AppendProcessTableFunctionBase {
public void eval(
