[FLINK-38824][table] Fix incorrect default values for primitive types

This closes #27363.

Author: dylanhz

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGenUtils.scala

@@ -301,7 +301,9 @@ object CodeGenUtils {
     // ordered by type root definition
     case CHAR | VARCHAR => s"$BINARY_STRING.EMPTY_UTF8"
     case BOOLEAN => "false"
-    case TINYINT | SMALLINT | INTEGER | DATE | TIME_WITHOUT_TIME_ZONE | INTERVAL_YEAR_MONTH => "-1"
+    case TINYINT => "((byte) -1)"
+    case SMALLINT => "((short) -1)"
+    case INTEGER | DATE | TIME_WITHOUT_TIME_ZONE | INTERVAL_YEAR_MONTH => "-1"
     case BIGINT | INTERVAL_DAY_TIME => "-1L"
     case FLOAT => "-1.0f"
     case DOUBLE => "-1.0d"

flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/codegen/CodeGenUtilsTest.scala

@@ -91,8 +91,8 @@ object CodeGenUtilsTest {
     java.util.stream.Stream.of(
       // Basic primitive types
       Arguments.of(new BooleanType(), "false"),
-      Arguments.of(new TinyIntType(), "-1"),
-      Arguments.of(new SmallIntType(), "-1"),
+      Arguments.of(new TinyIntType(), "((byte) -1)"),
+      Arguments.of(new SmallIntType(), "((short) -1)"),
       Arguments.of(new IntType(), "-1"),
       Arguments.of(new BigIntType(), "-1L"),
       Arguments.of(new FloatType(), "-1.0f"),
@@ -141,12 +141,12 @@ object CodeGenUtilsTest {
         DistinctType
           .newBuilder(objectIdentifier, new SmallIntType())
           .build(),
-        "-1"),
+        "((short) -1)"),
       Arguments.of(
         DistinctType
           .newBuilder(objectIdentifier, new TinyIntType())
           .build(),
-        "-1"),
+        "((byte) -1)"),
       Arguments.of(
         DistinctType
           .newBuilder(objectIdentifier, new BigIntType())

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

@@ -926,4 +926,26 @@ class CalcITCase extends StreamingTestBase {
     assertThat(arr.apply(1).get("nested2").asInstanceOf[Array[String]])
       .isEqualTo(Array("Test2", "False"))
   }
+
+  @Test
+  def testPrimitiveDefaultValues(): Unit = {
+    val sql =
+      """
+        |SELECT
+        |  a[1]
+        |  ,b[1]
+        |FROM (
+        |  VALUES (CAST(ARRAY[1,2] AS ARRAY<SMALLINT>), CAST(ARRAY[2,3] AS ARRAY<TINYINT>))
+        |) t(a, b)
+        |""".stripMargin
+
+    val result = tEnv
+      .executeSql(sql)
+      .collect()
+      .asScala
+      .toList
+      .map(_.toString)
+    val expected = List("1,2")
+    assertThat(result).isEqualTo(expected)
+  }
 }