[FLINK-8215] [table] Support implicit type widening for array/map constructors in SQL

This closes #5148.

flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala

@@ -1312,10 +1312,10 @@ abstract class CodeGenerator(
   }
 
   private[flink] def generateNullableOutputBoxing(
-      element: GeneratedExpression,
+      expr: GeneratedExpression,
       typeInfo: TypeInformation[_])
     : GeneratedExpression = {
-    val boxedExpr = generateOutputFieldBoxing(element)
+    val boxedExpr = generateOutputFieldBoxing(generateCast(nullCheck, expr, typeInfo))
     val boxedTypeTerm = boxedTypeTermForTypeInfo(typeInfo)
     val exprOrNull: String = if (nullCheck) {
       s"${boxedExpr.nullTerm} ? null : ($boxedTypeTerm) ${boxedExpr.resultTerm}"

flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/ArrayTypeTest.scala

@@ -99,6 +99,9 @@ class ArrayTypeTest extends ArrayTypeTestBase {
       "Array(Array(1, 2, 3), Array(3, 2, 1))",
       "ARRAY[ARRAY[1, 2, 3], ARRAY[3, 2, 1]]",
       "[[1, 2, 3], [3, 2, 1]]")
+
+    // implicit type cast only works on SQL APIs.
+    testSqlApi("ARRAY[CAST(1 AS DOUBLE), CAST(2 AS FLOAT)]", "[1.0, 2.0]")
   }
 
   @Test

flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/MapTypeTest.scala

@@ -77,6 +77,9 @@ class MapTypeTest extends MapTypeTestBase {
       "map(2.0002p, 2.0003p)",
       "MAP[CAST(2.0002 AS DECIMAL), CAST(2.0003 AS DECIMAL)]",
       "{2.0002=2.0003}")
+
+    // implicit conversion
+    testSqlApi("MAP['k1', CAST(1 AS DOUBLE), 'k2', CAST(2 AS FLOAT)]", "{k1=1.0, k2=2.0}")
   }
 
   @Test

flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/validation/ArrayTypeValidationTest.scala

@@ -25,6 +25,16 @@ import org.junit.Test
 
 class ArrayTypeValidationTest extends ArrayTypeTestBase {
 
+  @Test(expected = classOf[ValidationException])
+  def testImplicitTypeCastTableApi(): Unit = {
+    testTableApi(array(1.0, 2.0f), "FAIL", "FAIL")
+  }
+
+  @Test(expected = classOf[ValidationException])
+  def testImplicitTypeCastArraySql(): Unit = {
+    testSqlApi("ARRAY['string', 12]", "FAIL")
+  }
+
   @Test(expected = classOf[ValidationException])
   def testObviousInvalidIndexTableApi(): Unit = {
     testTableApi('f2.at(0), "FAIL", "FAIL")

flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/validation/MapTypeValidationTest.scala

@@ -44,4 +44,14 @@ class MapTypeValidationTest extends MapTypeTestBase {
   def testEmptyMap(): Unit = {
     testAllApis("FAIL", "map()", "MAP[]", "FAIL")
   }
+
+  @Test(expected = classOf[ValidationException])
+  def testUnsupportedMapImplicitTypeCastTableApi(): Unit = {
+    testTableApi(map("k1", 1.0, "k2", 2.0f), "map('k1', 1.0, 'k2', 2.0f)", "FAIL")
+  }
+
+  @Test(expected = classOf[ValidationException])
+  def testUnsupportedMapImplicitTypeCastSql(): Unit = {
+    testSqlApi("MAP['k1', 'string', 'k2', 12]", "FAIL")
+  }
 }