[FLINK-21733][table-planner-blink] WatermarkAssigner incorrectly reco…

…mputing the rowtime index which may cause ArrayIndexOutOfBoundsException (apache#15150)
deepfield · Mar 12, 2021 · c2d9e69 · c2d9e69
1 parent 615d5d1
commit c2d9e69
Show file tree

Hide file tree

Showing 3 changed files with 42 additions and 5 deletions.
diff --git a/.../src/main/scala/org/apache/flink/table/planner/plan/nodes/calcite/WatermarkAssigner.scala b/.../src/main/scala/org/apache/flink/table/planner/plan/nodes/calcite/WatermarkAssigner.scala
@@ -65,11 +65,7 @@ abstract class WatermarkAssigner(
  }
 
  override def copy(traitSet: RelTraitSet, inputs: util.List[RelNode]): RelNode = {
- val rowtimeFieldName = inputRel.getRowType.getFieldNames.get(rowtimeFieldIndex)
- val newInputRel = inputs.get(0)
- // the input fields maybe reordered, re-computed the rowtime index
- val newIndex = newInputRel.getRowType.getFieldNames.indexOf(rowtimeFieldName)
- copy(traitSet, newInputRel, newIndex, watermarkExpr)
+ copy(traitSet, inputs.get(0), rowtimeFieldIndex, watermarkExpr)
  }
 
  /**

diff --git a/...src/test/resources/org/apache/flink/table/planner/plan/stream/sql/SourceWatermarkTest.xml b/...src/test/resources/org/apache/flink/table/planner/plan/stream/sql/SourceWatermarkTest.xml
@@ -127,6 +127,24 @@ LogicalProject(a=[$0], b=[$1])
  <![CDATA[
 Calc(select=[a, b], where=[(b > 10)])
 +- TableSourceScan(table=[[default_catalog, default_database, VirtualTable, watermark=[-(+($2, 5000:INTERVAL SECOND), 5000:INTERVAL SECOND)]]], fields=[a, b, c])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testProjectTransposeWatermarkAssigner">
+ <Resource name="sql">
+ <![CDATA[SELECT a, b, ts FROM t1]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(a=[$0], b=[$1], ts=[$5])
++- LogicalWatermarkAssigner(rowtime=[ts], watermark=[-($5, 10000:INTERVAL SECOND)])
+ +- LogicalProject(a=[$0], b=[$1], c=[$2], d=[$3], t=[$4], ts=[$4])
+ +- LogicalTableScan(table=[[default_catalog, default_database, t1]])
+]]>
+ </Resource>
+ <Resource name="optimized exec plan">
+ <![CDATA[
+TableSourceScan(table=[[default_catalog, default_database, t1, project=[a, b, t], watermark=[-($2, 10000:INTERVAL SECOND)]]], fields=[a, b, t])
 ]]>
  </Resource>
  </TestCase>

diff --git a/...k/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/SourceWatermarkTest.scala b/...k/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/SourceWatermarkTest.scala
@@ -137,4 +137,27 @@ class SourceWatermarkTest extends TableTestBase {
  def testWatermarkWithMetadata(): Unit = {
  util.verifyExecPlan("SELECT a, b FROM MyTable")
  }
+
+ @Test
+ def testProjectTransposeWatermarkAssigner(): Unit = {
+ val sourceDDL =
+ s"""
+ |CREATE TEMPORARY TABLE `t1` (
+ | `a` VARCHAR,
+ | `b` VARCHAR,
+ | `c` VARCHAR,
+ | `d` INT,
+ | `t` TIMESTAMP(3),
+ | `ts` AS `t`,
+ | WATERMARK FOR `ts` AS `ts` - INTERVAL '10' SECOND
+ |) WITH (
+ | 'connector' = 'values',
+ | 'enable-watermark-push-down' = 'true',
+ | 'bounded' = 'false',
+ | 'disable-lookup' = 'true'
+ |)
+ """.stripMargin
+ util.tableEnv.executeSql(sourceDDL)
+ util.verifyExecPlan("SELECT a, b, ts FROM t1")
+ }
 }