This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a commit to branch release-2.3
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-2.3 by this push:
new 3822b7f8db4 [FLINK-40082][table] Watermark generated method
`extractTimestamp` might invoke code throwing exceptions
3822b7f8db4 is described below
commit 3822b7f8db403a5b8b8b69222776644c0c5e51e5
Author: Sergey Nuyanzin <[email protected]>
AuthorDate: Wed Jul 8 07:53:10 2026 +0200
[FLINK-40082][table] Watermark generated method `extractTimestamp` might
invoke code throwing exceptions
---
.../codegen/WatermarkGeneratorCodeGenerator.scala | 2 +-
.../plan/nodes/exec/stream/MiscSemanticTests.java | 3 +-
.../exec/stream/WatermarkAssignerRestoreTest.java | 5 +-
.../exec/stream/WatermarkAssignerTestPrograms.java | 32 +++
.../plan/watermark-assigner-pushdown-decode.json | 265 +++++++++++++++++++++
5 files changed, 302 insertions(+), 5 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/WatermarkGeneratorCodeGenerator.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/WatermarkGeneratorCodeGenerator.scala
index f63f270fecf..ec90657a489 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/WatermarkGeneratorCodeGenerator.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/WatermarkGeneratorCodeGenerator.scala
@@ -115,7 +115,7 @@ object WatermarkGeneratorCodeGenerator {
}
@Override
- public long extractTimestamp($ROW_DATA row) {
+ public long extractTimestamp($ROW_DATA row) throws Exception {
$generatedTimestampCode
}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/MiscSemanticTests.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/MiscSemanticTests.java
index 7467e7cf45e..25957d8115f 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/MiscSemanticTests.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/MiscSemanticTests.java
@@ -31,6 +31,7 @@ public class MiscSemanticTests extends SemanticTestBase {
public List<TableTestProgram> programs() {
return List.of(
WindowRankTestPrograms.WINDOW_RANK_HOP_TVF_NAMED_MIN_TOP_1,
- CalcTestPrograms.CURRENT_WATERMARK);
+ CalcTestPrograms.CURRENT_WATERMARK,
+
WatermarkAssignerTestPrograms.WATERMARK_ASSIGNER_PUSHDOWN_DECODE);
}
}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/WatermarkAssignerRestoreTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/WatermarkAssignerRestoreTest.java
index e64656f9fe7..0a684e4c133 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/WatermarkAssignerRestoreTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/WatermarkAssignerRestoreTest.java
@@ -21,11 +21,10 @@ package
org.apache.flink.table.planner.plan.nodes.exec.stream;
import
org.apache.flink.table.planner.plan.nodes.exec.testutils.RestoreTestBase;
import org.apache.flink.table.test.program.TableTestProgram;
-import java.util.Arrays;
import java.util.List;
/** Restore tests for {@link StreamExecWatermarkAssigner}. */
-public class WatermarkAssignerRestoreTest extends RestoreTestBase {
+class WatermarkAssignerRestoreTest extends RestoreTestBase {
public WatermarkAssignerRestoreTest() {
super(StreamExecWatermarkAssigner.class);
@@ -33,7 +32,7 @@ public class WatermarkAssignerRestoreTest extends
RestoreTestBase {
@Override
public List<TableTestProgram> programs() {
- return Arrays.asList(
+ return List.of(
WatermarkAssignerTestPrograms.WATERMARK_ASSIGNER_BASIC_FILTER,
WatermarkAssignerTestPrograms.WATERMARK_ASSIGNER_PUSHDOWN_METADATA,
WatermarkAssignerTestPrograms.WATERMARK_ASSIGNER_PUSHDOWN_COMPUTED);
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/WatermarkAssignerTestPrograms.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/WatermarkAssignerTestPrograms.java
index fe33b7bb680..15907bb2f09 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/WatermarkAssignerTestPrograms.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/WatermarkAssignerTestPrograms.java
@@ -24,6 +24,8 @@ import org.apache.flink.table.test.program.TableTestProgram;
import org.apache.flink.table.utils.DateTimeUtils;
import org.apache.flink.types.Row;
+import java.nio.charset.StandardCharsets;
+
/** {@link TableTestProgram} definitions for testing {@link
StreamExecWindowRank}. */
public class WatermarkAssignerTestPrograms {
@@ -178,4 +180,34 @@ public class WatermarkAssignerTestPrograms {
.runSql(
"INSERT INTO sink_t SELECT i, s, ts,
CURRENT_WATERMARK(ts) as w FROM source_t")
.build();
+
+ static final TableTestProgram WATERMARK_ASSIGNER_PUSHDOWN_DECODE =
+ TableTestProgram.of(
+ "watermark-assigner-pushdown-decode",
+ "validates watermark assigner with a DECODE
computed column (FLINK-40082)")
+ .setupTableSource(
+ SourceTestStep.newBuilder("source_t")
+ .addSchema(
+ "a INT",
+ "b BYTES",
+ "ts AS TO_TIMESTAMP(DECODE(b,
'UTF-8'))",
+ "WATERMARK FOR ts AS ts - INTERVAL
'1' SECOND")
+ .producedBeforeRestore(
+ Row.of(
+ 1,
+ "2020-04-15 08:00:00"
+
.getBytes(StandardCharsets.UTF_8)),
+ Row.of(
+ 2,
+ "2020-04-15 08:00:01"
+
.getBytes(StandardCharsets.UTF_8)))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema("a INT", "ts TIMESTAMP(3)")
+ .consumedBeforeRestore(
+ "+I[1, 2020-04-15T08:00]", "+I[2,
2020-04-15T08:00:01]")
+ .build())
+ .runSql("INSERT INTO sink_t SELECT a, ts FROM source_t")
+ .build();
}
diff --git
a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-watermark-assigner_1/watermark-assigner-pushdown-decode/plan/watermark-assigner-pushdown-decode.json
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-watermark-assigner_1/watermark-assigner-pushdown-decode/plan/watermark-assigner-pushdown-decode.json
new file mode 100644
index 00000000000..30873b1ad86
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-watermark-assigner_1/watermark-assigner-pushdown-decode/plan/watermark-assigner-pushdown-decode.json
@@ -0,0 +1,265 @@
+{
+ "flinkVersion" : "2.3",
+ "nodes" : [ {
+ "id" : 11,
+ "type" : "stream-exec-table-source-scan_2",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`source_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "a",
+ "dataType" : "INT"
+ }, {
+ "name" : "b",
+ "dataType" : "VARBINARY(2147483647)"
+ }, {
+ "name" : "ts",
+ "kind" : "COMPUTED",
+ "expression" : {
+ "rexNode" : {
+ "kind" : "CALL",
+ "internalName" : "$TO_TIMESTAMP$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "internalName" : "$DECODE$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "VARBINARY(2147483647)"
+ }, {
+ "kind" : "LITERAL",
+ "value" : "UTF-8",
+ "type" : "CHAR(5) NOT NULL"
+ } ],
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "type" : "TIMESTAMP(3)"
+ },
+ "serializableString" : "TO_TIMESTAMP(DECODE(`b`, 'UTF-8'))"
+ }
+ } ],
+ "watermarkSpecs" : [ {
+ "rowtimeAttribute" : "ts",
+ "expression" : {
+ "rexNode" : {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$-$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "TIMESTAMP(3)"
+ }, {
+ "kind" : "LITERAL",
+ "value" : "1000",
+ "type" : "INTERVAL SECOND(6) NOT NULL"
+ } ],
+ "type" : "TIMESTAMP(3)"
+ },
+ "serializableString" : "`ts` - INTERVAL '1' SECOND"
+ }
+ } ]
+ }
+ }
+ },
+ "abilities" : [ {
+ "type" : "WatermarkPushDown",
+ "watermarkExpr" : {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$-$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "internalName" : "$TO_TIMESTAMP$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "internalName" : "$DECODE$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "VARBINARY(2147483647)"
+ }, {
+ "kind" : "LITERAL",
+ "value" : "UTF-8",
+ "type" : "CHAR(5) NOT NULL"
+ } ],
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "type" : "TIMESTAMP(3)"
+ }, {
+ "kind" : "LITERAL",
+ "value" : "1000",
+ "type" : "INTERVAL SECOND(6) NOT NULL"
+ } ],
+ "type" : "TIMESTAMP(3)"
+ },
+ "rowtimeExpr" : {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$REINTERPRET$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "internalName" : "$TO_TIMESTAMP$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "internalName" : "$DECODE$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "VARBINARY(2147483647)"
+ }, {
+ "kind" : "LITERAL",
+ "value" : "UTF-8",
+ "type" : "CHAR(5) NOT NULL"
+ } ],
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "type" : "TIMESTAMP(3)"
+ } ],
+ "type" : {
+ "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
+ "precision" : 3,
+ "kind" : "ROWTIME"
+ }
+ },
+ "idleTimeoutMillis" : -1,
+ "producedType" : "ROW<`a` INT, `b` VARBINARY(2147483647)> NOT NULL",
+ "watermarkParams" : {
+ "emitStrategy" : "ON_EVENT",
+ "alignGroupName" : null,
+ "alignMaxDrift" : "PT0S",
+ "alignUpdateInterval" : "PT1S",
+ "sourceIdleTimeout" : -1
+ }
+ } ]
+ },
+ "outputType" : "ROW<`a` INT, `b` VARBINARY(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, source_t, watermark=[-(TO_TIMESTAMP(DECODE(b,
_UTF-16LE'UTF-8')), 1000:INTERVAL SECOND)], watermarkEmitStrategy=[on-event]]],
fields=[a, b])"
+ }, {
+ "id" : 12,
+ "type" : "stream-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$REINTERPRET$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "internalName" : "$TO_TIMESTAMP$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "internalName" : "$DECODE$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "VARBINARY(2147483647)"
+ }, {
+ "kind" : "LITERAL",
+ "value" : "UTF-8",
+ "type" : "CHAR(5) NOT NULL"
+ } ],
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "type" : "TIMESTAMP(3)"
+ } ],
+ "type" : {
+ "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
+ "precision" : 3,
+ "kind" : "ROWTIME"
+ }
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : {
+ "type" : "ROW",
+ "fields" : [ {
+ "name" : "a",
+ "fieldType" : "INT"
+ }, {
+ "name" : "ts",
+ "fieldType" : {
+ "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
+ "precision" : 3,
+ "kind" : "ROWTIME"
+ }
+ } ]
+ },
+ "description" : "Calc(select=[a, Reinterpret(TO_TIMESTAMP(DECODE(b,
'UTF-8'))) AS ts])"
+ }, {
+ "id" : 13,
+ "type" : "stream-exec-sink_2",
+ "configuration" : {
+ "table.exec.sink.keyed-shuffle" : "AUTO",
+ "table.exec.sink.not-null-enforcer" : "ERROR",
+ "table.exec.sink.rowtime-inserter" : "ENABLED",
+ "table.exec.sink.type-length-enforcer" : "IGNORE",
+ "table.exec.sink.upsert-materialize" : "AUTO"
+ },
+ "dynamicTableSink" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`sink_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "a",
+ "dataType" : "INT"
+ }, {
+ "name" : "ts",
+ "dataType" : "TIMESTAMP(3)"
+ } ]
+ }
+ }
+ }
+ },
+ "inputChangelogMode" : [ "INSERT" ],
+ "upsertMaterializeStrategy" : "VALUE",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : {
+ "type" : "ROW",
+ "fields" : [ {
+ "name" : "a",
+ "fieldType" : "INT"
+ }, {
+ "name" : "ts",
+ "fieldType" : {
+ "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
+ "precision" : 3,
+ "kind" : "ROWTIME"
+ }
+ } ]
+ },
+ "description" : "Sink(table=[default_catalog.default_database.sink_t],
fields=[a, ts])"
+ } ],
+ "edges" : [ {
+ "source" : 11,
+ "target" : 12,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 12,
+ "target" : 13,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
