[FLINK-17751][table-planner-blink] Fix proctime defined in ddl can no…

…t work with over window in Table api This closes apache#12342
shuttie · May 27, 2020 · 0e67f18 · 0e67f18
1 parent 5a9fe5d
commit 0e67f18
Show file tree

Hide file tree

Showing 3 changed files with 20 additions and 3 deletions.
diff --git a/...mon/src/main/java/org/apache/flink/table/types/utils/LegacyTypeInfoDataTypeConverter.java b/...mon/src/main/java/org/apache/flink/table/types/utils/LegacyTypeInfoDataTypeConverter.java
@@ -281,7 +281,7 @@ private static DataType convertToTimeAttributeType(TimeIndicatorTypeInfo timeInd
 
 	private static boolean canConvertToTimeAttributeTypeInfo(DataType dataType) {
 		return hasRoot(dataType.getLogicalType(), LogicalTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE) &&
-			dataTypeTypeInfoMap.containsKey(dataType) && // checks precision and conversion
+			dataTypeTypeInfoMap.containsKey(dataType.nullable()) && // checks precision and conversion and ignore nullable
 			((TimestampType) dataType.getLogicalType()).getKind() != TimestampKind.REGULAR;
 	}
 

diff --git a/...k/src/test/resources/org/apache/flink/table/planner/plan/stream/table/TableSourceTest.xml b/...k/src/test/resources/org/apache/flink/table/planner/plan/stream/table/TableSourceTest.xml
@@ -79,6 +79,24 @@ Calc(select=[PROCTIME_MATERIALIZE(ptime) AS ptime, name, val, id])
 +- WatermarkAssigner(rowtime=[rtime], watermark=[rtime])
    +- Calc(select=[id, rtime, val, name, PROCTIME() AS ptime])
       +- TableSourceScan(table=[[default_catalog, default_database, T]], fields=[id, rtime, val, name])
+]]>
+    </Resource>
+  </TestCase>
+  <TestCase name="testProcTimeTableSourceOverWindow">
+    <Resource name="planBefore">
+      <![CDATA[
+LogicalFilter(condition=[>($2, 100)])
++- LogicalProject(id=[$0], name=[$2], valSum=[AS(SUM($1) OVER (PARTITION BY $0 ORDER BY PROCTIME() NULLS FIRST RANGE BETWEEN 7200000 PRECEDING AND CURRENT ROW), _UTF-16LE'valSum')])
+   +- LogicalTableScan(table=[[default_catalog, default_database, procTimeT]])
+]]>
+    </Resource>
+    <Resource name="planAfter">
+      <![CDATA[
+Calc(select=[id, name, w0$o0 AS valSum], where=[>(w0$o0, 100)])
++- OverAggregate(partitionBy=[id], orderBy=[$3 ASC], window=[ RANG BETWEEN 7200000 PRECEDING AND CURRENT ROW], select=[id, val, name, $3, SUM(val) AS w0$o0])
+   +- Exchange(distribution=[hash[id]])
+      +- Calc(select=[id, val, name, PROCTIME() AS $3])
+         +- TableSourceScan(table=[[default_catalog, default_database, procTimeT]], fields=[id, val, name])
 ]]>
     </Resource>
   </TestCase>

diff --git a/...ink/src/test/scala/org/apache/flink/table/planner/plan/stream/table/TableSourceTest.scala b/...ink/src/test/scala/org/apache/flink/table/planner/plan/stream/table/TableSourceTest.scala
@@ -25,7 +25,7 @@ import org.junit.{Ignore, Test}
 
 class TableSourceTest extends TableTestBase {
 
-  val util = streamTestUtil()
+  private val util = streamTestUtil()
 
   @Test
   def testTableSourceWithTimestampRowTimeField(): Unit = {
@@ -95,7 +95,6 @@ class TableSourceTest extends TableTestBase {
     util.verifyPlan(t)
   }
 
-  @Ignore("remove ignore once FLINK-17751 is fixed")
   @Test
   def testProcTimeTableSourceOverWindow(): Unit = {
     val ddl =