[FLINK-19796][table] Explicit casting should be made if the type of a…

flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/sql/FlinkSqlOperatorTable.java

@@ -1054,8 +1054,7 @@ public List<SqlGroupedWindowFunction> getAuxiliaryFunctions() {
     public static final SqlAggFunction SINGLE_VALUE = SqlStdOperatorTable.SINGLE_VALUE;
 
     // ARRAY OPERATORS
-    public static final SqlOperator ARRAY_VALUE_CONSTRUCTOR =
-            SqlStdOperatorTable.ARRAY_VALUE_CONSTRUCTOR;
+    public static final SqlOperator ARRAY_VALUE_CONSTRUCTOR = new SqlArrayFunction();
     public static final SqlOperator ELEMENT = SqlStdOperatorTable.ELEMENT;
 
     // MAP OPERATORS

flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/sql/SqlArrayFunction.java

@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.functions.sql;
+
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlCallBinding;
+import org.apache.calcite.sql.SqlFunction;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlOperatorBinding;
+import org.apache.calcite.sql.fun.SqlArrayValueConstructor;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
+import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.calcite.sql.type.SqlTypeUtil;
+
+import java.util.List;
+
+/**
+ * {@link SqlFunction} for <code>ARRAY</code>, which makes explicit casting if the element type not
+ * equals the derived component type.
+ */
+public class SqlArrayFunction extends SqlArrayValueConstructor {
+    @Override
+    public RelDataType inferReturnType(SqlOperatorBinding opBinding) {
+        RelDataType type =
+                getComponentType(opBinding.getTypeFactory(), opBinding.collectOperandTypes());
+        if (null == type) {
+            return null;
+        }
+
+        // explicit cast elements to component type if they are not same
+        if (opBinding instanceof SqlCallBinding) {
+            SqlCall call = ((SqlCallBinding) opBinding).getCall();
+            List<RelDataType> operandTypes = opBinding.collectOperandTypes();
+            List<SqlNode> operands = call.getOperandList();
+            int idx = -1;
+            for (RelDataType opType : operandTypes) {
+                idx += 1;
+                if (opType.equalsSansFieldNames(type)) {
+                    continue;
+                }
+                call.setOperand(idx, castTo(operands.get(idx), type));
+            }
+        }
+
+        return SqlTypeUtil.createArrayType(opBinding.getTypeFactory(), type, false);
+    }
+
+    private SqlNode castTo(SqlNode node, RelDataType type) {
+        return SqlStdOperatorTable.CAST.createCall(
+                SqlParserPos.ZERO,
+                node,
+                SqlTypeUtil.convertTypeToSpec(type).withNullable(type.isNullable()));
+    }
+}

flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/CalcTest.xml

@@ -22,13 +22,13 @@ limitations under the License.
     </Resource>
     <Resource name="ast">
       <![CDATA[
-LogicalProject(EXPR$0=[ARRAY(_UTF-16LE'Hi', _UTF-16LE'Hello', $2)])
+LogicalProject(EXPR$0=[ARRAY(_UTF-16LE'Hi':VARCHAR(2147483647) CHARACTER SET "UTF-16LE", _UTF-16LE'Hello':VARCHAR(2147483647) CHARACTER SET "UTF-16LE", $2)])
 +- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]])
 ]]>
     </Resource>
     <Resource name="optimized exec plan">
       <![CDATA[
-Calc(select=[ARRAY(_UTF-16LE'Hi', _UTF-16LE'Hello', c) AS EXPR$0])
+Calc(select=[ARRAY(_UTF-16LE'Hi':VARCHAR(2147483647) CHARACTER SET "UTF-16LE", _UTF-16LE'Hello':VARCHAR(2147483647) CHARACTER SET "UTF-16LE", c) AS EXPR$0])
 +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
 ]]>
     </Resource>
@@ -409,4 +409,21 @@ Calc(select=[ROW(1, _UTF-16LE'Hi', a) AS EXPR$0])
 ]]>
     </Resource>
   </TestCase>
+  <TestCase name="testDecimalArrayWithDifferentPrecision">
+	<Resource name="sql">
+	  <![CDATA[SELECT ARRAY[0.12, 0.5, 0.99]]]>
+	</Resource>
+	<Resource name="ast">
+	  <![CDATA[
+LogicalProject(EXPR$0=[ARRAY(0.12:DECIMAL(3, 2), 0.5:DECIMAL(3, 2), 0.99:DECIMAL(3, 2))])
++- LogicalValues(tuples=[[{ 0 }]])
+]]>
+	</Resource>
+	<Resource name="optimized exec plan">
+	  <![CDATA[
+Calc(select=[ARRAY(0.12:DECIMAL(3, 2), 0.5:DECIMAL(3, 2), 0.99:DECIMAL(3, 2)) AS EXPR$0])
++- Values(tuples=[[{ 0 }]], values=[ZERO])
+]]>
+	</Resource>
+  </TestCase>
 </Root>

flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/CalcTest.xml

@@ -22,13 +22,13 @@ limitations under the License.
     </Resource>
     <Resource name="ast">
       <![CDATA[
-LogicalProject(EXPR$0=[ARRAY(_UTF-16LE'Hi', _UTF-16LE'Hello', $2)])
+LogicalProject(EXPR$0=[ARRAY(_UTF-16LE'Hi':VARCHAR(2147483647) CHARACTER SET "UTF-16LE", _UTF-16LE'Hello':VARCHAR(2147483647) CHARACTER SET "UTF-16LE", $2)])
 +- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]])
 ]]>
     </Resource>
     <Resource name="optimized exec plan">
       <![CDATA[
-Calc(select=[ARRAY(_UTF-16LE'Hi', _UTF-16LE'Hello', c) AS EXPR$0])
+Calc(select=[ARRAY(_UTF-16LE'Hi':VARCHAR(2147483647) CHARACTER SET "UTF-16LE", _UTF-16LE'Hello':VARCHAR(2147483647) CHARACTER SET "UTF-16LE", c) AS EXPR$0])
 +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
 ]]>
     </Resource>
@@ -410,4 +410,21 @@ Calc(select=[ROW(1, _UTF-16LE'Hi', a) AS EXPR$0])
 ]]>
     </Resource>
   </TestCase>
+  <TestCase name="testDecimalArrayWithDifferentPrecision">
+    <Resource name="sql">
+	  <![CDATA[SELECT ARRAY[0.12, 0.5, 0.99]]]>
+	</Resource>
+	<Resource name="ast">
+	  <![CDATA[
+LogicalProject(EXPR$0=[ARRAY(0.12:DECIMAL(3, 2), 0.5:DECIMAL(3, 2), 0.99:DECIMAL(3, 2))])
++- LogicalValues(tuples=[[{ 0 }]])
+]]>
+	</Resource>
+	<Resource name="optimized exec plan">
+	  <![CDATA[
+Calc(select=[ARRAY(0.12:DECIMAL(3, 2), 0.5:DECIMAL(3, 2), 0.99:DECIMAL(3, 2)) AS EXPR$0])
++- Values(tuples=[[{ 0 }]])
+]]>
+	</Resource>
+  </TestCase>
 </Root>

flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/CalcTest.scala

@@ -171,4 +171,9 @@ class CalcTest extends TableTestBase {
   def testOrWithIsNullInIf(): Unit = {
     util.verifyExecPlan("SELECT IF(c = '' OR c IS NULL, 'a', 'b') FROM MyTable")
   }
+
+  @Test
+  def testDecimalArrayWithDifferentPrecision(): Unit = {
+    util.verifyExecPlan("SELECT ARRAY[0.12, 0.5, 0.99]")
+  }
 }

flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/CalcTest.scala

@@ -170,4 +170,9 @@ class CalcTest extends TableTestBase {
   def testOrWithIsNullInIf(): Unit = {
     util.verifyExecPlan("SELECT IF(c = '' OR c IS NULL, 'a', 'b') FROM MyTable")
   }
+
+  @Test
+  def testDecimalArrayWithDifferentPrecision(): Unit = {
+    util.verifyExecPlan("SELECT ARRAY[0.12, 0.5, 0.99]")
+  }
 }

flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/CalcITCase.scala

@@ -407,4 +407,17 @@ class CalcITCase extends StreamingTestBase {
       List("1,HI,1111,true,111","2,HELLO,2222,false,222", "3,HELLO WORLD,3333,true,333")
     assertEquals(expected.sorted, sink.getAppendResults.sorted)
   }
+
+  @Test
+  def testDecimalArrayWithDifferentPrecision(): Unit = {
+    val sqlQuery = "SELECT ARRAY[0.12, 0.5, 0.99]"
+
+    val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
+    val sink = new TestingAppendSink
+    result.addSink(sink)
+    env.execute()
+
+    val expected = List("[0.12, 0.50, 0.99]")
+    assertEquals(expected.sorted, sink.getAppendResults.sorted)
+  }
 }