From 560b3638979fef79b2e3cb9f5a84386fa8ef49a0 Mon Sep 17 00:00:00 2001
From: Jark Wu <jark@apache.org>
Date: Thu, 29 Apr 2021 20:22:35 +0800
Subject: [PATCH 1/2] [FLINK-22313][table-planner-blink] Redundant CAST in plan
 when selecting window start and window end in window agg

---
 .../org/apache/calcite/sql/SqlGroupedWindowFunction.java    | 6 +-----
 .../flink/table/planner/expressions/PlannerWindowEnd.java   | 2 +-
 .../flink/table/planner/expressions/PlannerWindowStart.java | 2 +-
 .../table/planner/plan/stream/sql/agg/GroupWindowTest.xml   | 4 ++--
 4 files changed, 5 insertions(+), 9 deletions(-)

diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/calcite/sql/SqlGroupedWindowFunction.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/calcite/sql/SqlGroupedWindowFunction.java
index fb620da0f7b4c..60fa7489329b0 100644
--- a/flink-table/flink-table-planner-blink/src/main/java/org/apache/calcite/sql/SqlGroupedWindowFunction.java
+++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/calcite/sql/SqlGroupedWindowFunction.java
@@ -184,11 +184,7 @@ private static class WindowStartEndReturnTypeInference implements SqlReturnTypeI
         public WindowStartEndReturnTypeInference() {}
 
         public RelDataType inferReturnType(SqlOperatorBinding opBinding) {
-            if (opBinding.getOperandType(0).getSqlTypeName().equals(SqlTypeName.TIMESTAMP)) {
-                return opBinding.getOperandType(0);
-            } else {
-                return explicit.inferReturnType(opBinding);
-            }
+            return explicit.inferReturnType(opBinding);
         }
     }
 }
diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/PlannerWindowEnd.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/PlannerWindowEnd.java
index 59a4437530f0f..d3010db037ccb 100644
--- a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/PlannerWindowEnd.java
+++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/PlannerWindowEnd.java
@@ -36,7 +36,7 @@ public PlannerWindowEnd(@JsonProperty(FIELD_NAME_REFERENCE) PlannerWindowReferen
 
     @Override
     public LogicalType getResultType() {
-        return new TimestampType(3);
+        return new TimestampType(false, 3);
     }
 
     @Override
diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/PlannerWindowStart.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/PlannerWindowStart.java
index 9dd894f02bd8e..1e8819343a807 100644
--- a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/PlannerWindowStart.java
+++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/PlannerWindowStart.java
@@ -37,7 +37,7 @@ public PlannerWindowStart(
 
     @Override
     public LogicalType getResultType() {
-        return new TimestampType(3);
+        return new TimestampType(false, 3);
     }
 
     @Override
diff --git a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/agg/GroupWindowTest.xml b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/agg/GroupWindowTest.xml
index a51280204ee5c..0f330752ac661 100644
--- a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/agg/GroupWindowTest.xml
+++ b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/agg/GroupWindowTest.xml
@@ -126,7 +126,7 @@ LogicalProject(EXPR$0=[$1], wAvg=[$2], EXPR$2=[HOP_START($0)], EXPR$3=[HOP_END($
     </Resource>
     <Resource name="optimized exec plan">
       <![CDATA[
-Calc(select=[EXPR$0, wAvg, CAST(w$start) AS EXPR$2, CAST(w$end) AS EXPR$3])
+Calc(select=[EXPR$0, wAvg, w$start AS EXPR$2, w$end AS EXPR$3])
 +- GroupWindowAggregate(window=[SlidingGroupWindow('w$, proctime, 3600000, 900000)], properties=[w$start, w$end, w$proctime], select=[COUNT(*) AS EXPR$0, weightedAvg(c, a) AS wAvg, start('w$) AS w$start, end('w$) AS w$end, proctime('w$) AS w$proctime])
    +- Exchange(distribution=[single])
       +- Calc(select=[proctime, c, CAST(a) AS a])
@@ -388,7 +388,7 @@ LogicalProject(EXPR$0=[$1], wAvg=[$2], EXPR$2=[SESSION_START($0)], EXPR$3=[SESSI
     </Resource>
     <Resource name="optimized exec plan">
       <![CDATA[
-Calc(select=[EXPR$0, wAvg, CAST(w$start) AS EXPR$2, CAST(w$end) AS EXPR$3])
+Calc(select=[EXPR$0, wAvg, w$start AS EXPR$2, w$end AS EXPR$3])
 +- GroupWindowAggregate(window=[SessionGroupWindow('w$, proctime, 900000)], properties=[w$start, w$end, w$proctime], select=[COUNT(*) AS EXPR$0, weightedAvg(c, a) AS wAvg, start('w$) AS w$start, end('w$) AS w$end, proctime('w$) AS w$proctime])
    +- Exchange(distribution=[single])
       +- Calc(select=[proctime, c, CAST(a) AS a])

From 4fbba845c4423843678c023251defed5ac4c89db Mon Sep 17 00:00:00 2001
From: Jark Wu <jark@apache.org>
Date: Mon, 10 May 2021 15:09:31 +0800
Subject: [PATCH 2/2] fixup

---
 .../testEventTimeTumbleWindow.out             | 16 ++++++-------
 .../testProcTimeTumbleWindow.out              | 23 ++++---------------
 .../plan/stream/sql/LegacyTableSourceTest.xml |  2 +-
 .../plan/stream/sql/MatchRecognizeTest.xml    |  2 +-
 4 files changed, 15 insertions(+), 28 deletions(-)

diff --git a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupWindowAggregateJsonPlanTest_jsonplan/testEventTimeTumbleWindow.out b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupWindowAggregateJsonPlanTest_jsonplan/testEventTimeTumbleWindow.out
index 7a57f9a53fc27..73bbf05191e19 100644
--- a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupWindowAggregateJsonPlanTest_jsonplan/testEventTimeTumbleWindow.out
+++ b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupWindowAggregateJsonPlanTest_jsonplan/testEventTimeTumbleWindow.out
@@ -385,14 +385,14 @@
       }, {
         "w$start" : {
           "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
-          "nullable" : true,
+          "nullable" : false,
           "precision" : 3,
           "kind" : "REGULAR"
         }
       }, {
         "w$end" : {
           "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
-          "nullable" : true,
+          "nullable" : false,
           "precision" : 3,
           "kind" : "REGULAR"
         }
@@ -427,7 +427,7 @@
       "inputIndex" : 5,
       "type" : {
         "typeName" : "TIMESTAMP",
-        "nullable" : true,
+        "nullable" : false,
         "precision" : 3
       }
     }, {
@@ -435,7 +435,7 @@
       "inputIndex" : 6,
       "type" : {
         "typeName" : "TIMESTAMP",
-        "nullable" : true,
+        "nullable" : false,
         "precision" : 3
       }
     }, {
@@ -485,14 +485,14 @@
       }, {
         "window_start" : {
           "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
-          "nullable" : true,
+          "nullable" : false,
           "precision" : 3,
           "kind" : "REGULAR"
         }
       }, {
         "window_end" : {
           "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
-          "nullable" : true,
+          "nullable" : false,
           "precision" : 3,
           "kind" : "REGULAR"
         }
@@ -550,14 +550,14 @@
       }, {
         "window_start" : {
           "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
-          "nullable" : true,
+          "nullable" : false,
           "precision" : 3,
           "kind" : "REGULAR"
         }
       }, {
         "window_end" : {
           "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
-          "nullable" : true,
+          "nullable" : false,
           "precision" : 3,
           "kind" : "REGULAR"
         }
diff --git a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupWindowAggregateJsonPlanTest_jsonplan/testProcTimeTumbleWindow.out b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupWindowAggregateJsonPlanTest_jsonplan/testProcTimeTumbleWindow.out
index c3b015218035e..95d02161d1835 100644
--- a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupWindowAggregateJsonPlanTest_jsonplan/testProcTimeTumbleWindow.out
+++ b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupWindowAggregateJsonPlanTest_jsonplan/testProcTimeTumbleWindow.out
@@ -366,14 +366,14 @@
       }, {
         "w$start" : {
           "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
-          "nullable" : true,
+          "nullable" : false,
           "precision" : 3,
           "kind" : "REGULAR"
         }
       }, {
         "w$end" : {
           "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
-          "nullable" : true,
+          "nullable" : false,
           "precision" : 3,
           "kind" : "REGULAR"
         }
@@ -397,21 +397,8 @@
         "nullable" : true
       }
     }, {
-      "kind" : "REX_CALL",
-      "operator" : {
-        "name" : "CAST",
-        "kind" : "CAST",
-        "syntax" : "SPECIAL"
-      },
-      "operands" : [ {
-        "kind" : "INPUT_REF",
-        "inputIndex" : 3,
-        "type" : {
-          "typeName" : "TIMESTAMP",
-          "nullable" : true,
-          "precision" : 3
-        }
-      } ],
+      "kind" : "INPUT_REF",
+      "inputIndex" : 3,
       "type" : {
         "typeName" : "TIMESTAMP",
         "nullable" : false,
@@ -450,7 +437,7 @@
         "EXPR$2" : "BIGINT NOT NULL"
       } ]
     },
-    "description" : "Calc(select=[b, CAST(w$end) AS window_end, EXPR$2])"
+    "description" : "Calc(select=[b, w$end AS window_end, EXPR$2])"
   }, {
     "class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecSink",
     "dynamicTableSink" : {
diff --git a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/LegacyTableSourceTest.xml b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/LegacyTableSourceTest.xml
index 108a93c1a592b..af09500a05152 100644
--- a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/LegacyTableSourceTest.xml
+++ b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/LegacyTableSourceTest.xml
@@ -227,7 +227,7 @@ LogicalProject(name=[$0], EXPR$1=[TUMBLE_END($1)], EXPR$2=[$2])
     </Resource>
     <Resource name="optimized exec plan">
       <![CDATA[
-Calc(select=[name, CAST(w$end) AS EXPR$1, EXPR$2])
+Calc(select=[name, w$end AS EXPR$1, EXPR$2])
 +- GroupWindowAggregate(groupBy=[name], window=[TumblingGroupWindow('w$, proctime, 600000)], properties=[w$start, w$end, w$proctime], select=[name, AVG(val) AS EXPR$2, start('w$) AS w$start, end('w$) AS w$end, proctime('w$) AS w$proctime])
    +- Exchange(distribution=[hash[name]])
       +- Calc(select=[val, name, proctime], where=[(val > 100)])
diff --git a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/MatchRecognizeTest.xml b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/MatchRecognizeTest.xml
index e02a25e065485..e5b5b62be7a11 100644
--- a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/MatchRecognizeTest.xml
+++ b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/MatchRecognizeTest.xml
@@ -52,7 +52,7 @@ LogicalProject(symbol=[$0], price=[$2], rowTime=[TUMBLE_ROWTIME($1)], startTime=
     </Resource>
     <Resource name="optimized rel plan">
       <![CDATA[
-Calc(select=[symbol, price, CAST(w$rowtime) AS rowTime, CAST(w$start) AS startTime])
+Calc(select=[symbol, price, CAST(w$rowtime) AS rowTime, w$start AS startTime])
 +- GroupWindowAggregate(groupBy=[symbol], window=[TumblingGroupWindow('w$, matchRowtime, 3000)], properties=[w$start, w$end, w$rowtime, w$proctime], select=[symbol, SUM(price) AS price, start('w$) AS w$start, end('w$) AS w$end, rowtime('w$) AS w$rowtime, proctime('w$) AS w$proctime])
    +- Exchange(distribution=[hash[symbol]])
       +- Calc(select=[symbol, matchRowtime, price])