This is an automated email from the ASF dual-hosted git repository.
lincoln-lil 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 a417739e30e [FLINK-40153][table] Fix resolving inherited metadata
columns used by watermarks in LIKE clause
a417739e30e is described below
commit a417739e30e1fa2f40554619c9b35543f2b4d69e
Author: lincoln lee <[email protected]>
AuthorDate: Fri Jul 17 14:20:20 2026 +0800
[FLINK-40153][table] Fix resolving inherited metadata columns used by
watermarks in LIKE clause
This closes #28750.
---
.../operations/converters/MergeTableLikeUtil.java | 6 ++++
.../converters/MergeTableLikeUtilTest.java | 33 ++++++++++++++++++++
.../planner/plan/stream/sql/TableScanTest.xml | 22 +++++++++++++
.../planner/plan/stream/sql/TableScanTest.scala | 36 ++++++++++++++++++++++
4 files changed, 97 insertions(+)
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/MergeTableLikeUtil.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/MergeTableLikeUtil.java
index a5d722282fd..93676ee49e0 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/MergeTableLikeUtil.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/MergeTableLikeUtil.java
@@ -282,6 +282,12 @@ public class MergeTableLikeUtil {
} else if (sourceColumn instanceof UnresolvedMetadataColumn) {
if (mergingStrategies.get(FeatureOption.METADATA)
!= MergingStrategy.EXCLUDING) {
+ UnresolvedMetadataColumn metadataColumn =
+ (UnresolvedMetadataColumn) sourceColumn;
+ LogicalType columnType =
getLogicalType(dataTypeFactory, metadataColumn);
+ metadataFieldNamesToTypes.put(
+ sourceColumn.getName(),
+
typeFactory.createFieldTypeFromLogicalType(columnType));
columns.put(sourceColumn.getName(), sourceColumn);
}
}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/converters/MergeTableLikeUtilTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/converters/MergeTableLikeUtilTest.java
index bdd39709ebd..b7e8ad06743 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/converters/MergeTableLikeUtilTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/converters/MergeTableLikeUtilTest.java
@@ -507,6 +507,39 @@ class MergeTableLikeUtilTest {
assertThat(mergedSchema).isEqualTo(expectedSchema);
}
+ @Test
+ void mergeWatermarkReferencingInheritedMetadataColumn() {
+ Schema sourceSchema =
+ Schema.newBuilder()
+ .column("timestamp", DataTypes.TIMESTAMP(3))
+ .columnByMetadata("opts", DataTypes.TIMESTAMP(3), true)
+ .watermark("timestamp", "timestamp")
+ .build();
+
+ List<SqlWatermark> derivedWatermarkSpecs =
+ List.of(
+ new SqlWatermark(
+ SqlParserPos.ZERO,
+ identifier("opts"),
+ boundedStrategy("opts", "5")));
+
+ Map<FeatureOption, MergingStrategy> mergingStrategies =
getDefaultMergingStrategies();
+ mergingStrategies.put(FeatureOption.WATERMARKS,
MergingStrategy.EXCLUDING);
+
+ Schema mergedSchema =
+ util.mergeTables(
+ mergingStrategies, sourceSchema, List.of(),
derivedWatermarkSpecs, null);
+
+ Schema expectedSchema =
+ Schema.newBuilder()
+ .column("timestamp", DataTypes.TIMESTAMP(3))
+ .columnByMetadata("opts", DataTypes.TIMESTAMP(3), true)
+ .watermark("opts", "`opts` - INTERVAL '5' SECOND")
+ .build();
+
+ assertThat(mergedSchema).isEqualTo(expectedSchema);
+ }
+
@Test
void mergeIncludingWatermarksFailsOnDuplicate() {
Schema sourceSchema =
diff --git
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/TableScanTest.xml
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/TableScanTest.xml
index adf7dd0d612..9a1909432cd 100644
---
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/TableScanTest.xml
+++
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/TableScanTest.xml
@@ -102,6 +102,28 @@ Calc(select=[a, b, c], changelogMode=[I,UA,D])
+- Calc(select=[id, a, +(a, 1) AS b, c, TO_TIMESTAMP(c) AS ts],
changelogMode=[I,UA,D])
+- DropUpdateBefore(changelogMode=[I,UA,D])
+- TableSourceScan(table=[[default_catalog,
default_database, src]], fields=[id, a, c], changelogMode=[I,UB,UA,D])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testCreateTableLikeWithWatermarkReferencingMetadataColumn">
+ <Resource name="sql">
+ <![CDATA[INSERT INTO SinkTable SELECT `ts`, `opts` FROM
LikeSourceWithWatermark]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalSink(table=[default_catalog.default_database.SinkTable], fields=[ts,
opts])
++- LogicalProject(ts=[$0], opts=[$1])
+ +- LogicalWatermarkAssigner(rowtime=[opts], watermark=[-($1, 5000:INTERVAL
SECOND)])
+ +- LogicalProject(ts=[$0], opts=[CAST($1):TIMESTAMP(3) *ROWTIME*])
+ +- LogicalTableScan(table=[[default_catalog, default_database,
LikeSourceWithWatermark, metadata=[opts]]])
+]]>
+ </Resource>
+ <Resource name="optimized exec plan">
+ <![CDATA[
+Sink(table=[default_catalog.default_database.SinkTable], fields=[ts, opts])
++- WatermarkAssigner(rowtime=[opts], watermark=[(opts - 5000:INTERVAL SECOND)])
+ +- Calc(select=[ts, CAST(opts AS TIMESTAMP(3) *ROWTIME*) AS opts])
+ +- TableSourceScan(table=[[default_catalog, default_database,
LikeSourceWithWatermark, metadata=[opts]]], fields=[ts, opts])
]]>
</Resource>
</TestCase>
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/TableScanTest.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/TableScanTest.scala
index ea905662a42..60f39645d10 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/TableScanTest.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/TableScanTest.scala
@@ -146,6 +146,42 @@ class TableScanTest extends TableTestBase {
util.verifyExecPlan("SELECT `b`, `other_metadata` FROM MetadataTable")
}
+ @Test
+ def testCreateTableLikeWithWatermarkReferencingMetadataColumn(): Unit = {
+ util.tableEnv.executeSql("""
+ |CREATE TABLE LikeSourceTable (
+ | `ts` TIMESTAMP(3),
+ | `opts` TIMESTAMP(3) METADATA VIRTUAL,
+ | WATERMARK FOR `ts` AS `ts`
+ |) WITH (
+ | 'connector' = 'values',
+ | 'readable-metadata' = 'opts:TIMESTAMP(3)'
+ |)
+ |""".stripMargin)
+
+ util.tableEnv.executeSql("""
+ |CREATE TABLE LikeSourceWithWatermark (
+ | WATERMARK FOR `opts`
+ | AS `opts` - INTERVAL '5' SECOND
+ |) LIKE LikeSourceTable (
+ | EXCLUDING WATERMARKS
+ |)
+ |""".stripMargin)
+
+ util.tableEnv.executeSql("""
+ |CREATE TABLE SinkTable (
+ | `ts` TIMESTAMP(3),
+ | `opts` TIMESTAMP(3)
+ |) WITH (
+ | 'connector' = 'values'
+ |)
+ |""".stripMargin)
+
+ util.verifyExecPlanInsert(
+ "INSERT INTO SinkTable " +
+ "SELECT `ts`, `opts` FROM LikeSourceWithWatermark")
+ }
+
@Test
def testDDLWithWatermarkComputedColumn(): Unit = {
// Create table with field as atom expression.
