[FLINK-22313][table-planner-blink] Redundant CAST in plan when selecting window start and window end in window agg

Author: wuchong

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);
         }
     }
 }

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

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

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])