fix failed cases

XuQianJin-Stars · XuQianJin-Stars · commit 6b505ee7cdec · 2025-12-24T21:21:41.000+08:00
diff --git a/fluss-common/src/main/java/org/apache/fluss/utils/ArrowUtils.java b/fluss-common/src/main/java/org/apache/fluss/utils/ArrowUtils.java
@@ -452,7 +452,14 @@ private static Field toArrowField(String fieldName, DataType logicalType) {
             }
         } else if (logicalType instanceof MapType) {
             MapType mapType = (MapType) logicalType;
-            Field keyField = toArrowField("key", mapType.getKeyType());
+            DataType keyType = mapType.getKeyType();
+            if (keyType.isNullable()) {
+                throw new IllegalArgumentException(
+                        String.format(
+                                "Map key type must be non-nullable for Arrow conversion, but got: %s",
+                                keyType.asSummaryString()));
+            }
+            Field keyField = toArrowField("key", keyType);
             Field valueField = toArrowField("value", mapType.getValueType());
             FieldType structFieldType = new FieldType(false, ArrowType.Struct.INSTANCE, null);
             List<Field> structChildren = new ArrayList<>();
diff --git a/fluss-common/src/main/java/org/apache/fluss/utils/json/DataTypeJsonSerde.java b/fluss-common/src/main/java/org/apache/fluss/utils/json/DataTypeJsonSerde.java
@@ -285,7 +285,8 @@ private static DataType deserializeMap(JsonNode dataTypeNode) {
         final DataType keyType = DataTypeJsonSerde.INSTANCE.deserialize(keyNode);
         final JsonNode valueNode = dataTypeNode.get(FIELD_NAME_VALUE_TYPE);
         final DataType valueType = DataTypeJsonSerde.INSTANCE.deserialize(valueNode);
-        return new MapType(keyType, valueType);
+        final DataType nonNullableKeyType = keyType.isNullable() ? keyType.copy(false) : keyType;
+        return new MapType(nonNullableKeyType, valueType);
     }
 
     private static DataType deserializeRow(JsonNode dataTypeNode) {
diff --git a/fluss-common/src/test/java/org/apache/fluss/utils/json/DataTypeJsonSerdeTest.java b/fluss-common/src/test/java/org/apache/fluss/utils/json/DataTypeJsonSerdeTest.java
@@ -76,7 +76,7 @@ protected DataType[] createObjects() {
                         new LocalZonedTimestampType(),
                         new LocalZonedTimestampType(3),
                         new ArrayType(new IntType(false)),
-                        new MapType(new BigIntType(), new IntType(false)),
+                        new MapType(new BigIntType(false), new IntType(false)),
                         RowType.of(new BigIntType(), new IntType(false), new StringType()));
 
         final List<DataType> allTypes = new ArrayList<>();
@@ -137,8 +137,8 @@ protected String[] expectedJsons() {
             "{\"type\":\"TIMESTAMP_WITH_LOCAL_TIME_ZONE\",\"nullable\":false,\"precision\":3}",
             "{\"type\":\"ARRAY\",\"element_type\":{\"type\":\"INTEGER\",\"nullable\":false}}",
             "{\"type\":\"ARRAY\",\"nullable\":false,\"element_type\":{\"type\":\"INTEGER\",\"nullable\":false}}",
-            "{\"type\":\"MAP\",\"key_type\":{\"type\":\"BIGINT\"},\"value_type\":{\"type\":\"INTEGER\",\"nullable\":false}}",
-            "{\"type\":\"MAP\",\"nullable\":false,\"key_type\":{\"type\":\"BIGINT\"},\"value_type\":{\"type\":\"INTEGER\",\"nullable\":false}}",
+            "{\"type\":\"MAP\",\"key_type\":{\"type\":\"BIGINT\",\"nullable\":false},\"value_type\":{\"type\":\"INTEGER\",\"nullable\":false}}",
+            "{\"type\":\"MAP\",\"nullable\":false,\"key_type\":{\"type\":\"BIGINT\",\"nullable\":false},\"value_type\":{\"type\":\"INTEGER\",\"nullable\":false}}",
             "{\"type\":\"ROW\",\"fields\":[{\"name\":\"f0\",\"field_type\":{\"type\":\"BIGINT\"}},{\"name\":\"f1\",\"field_type\":{\"type\":\"INTEGER\",\"nullable\":false}},{\"name\":\"f2\",\"field_type\":{\"type\":\"STRING\"}}]}",
             "{\"type\":\"ROW\",\"nullable\":false,\"fields\":[{\"name\":\"f0\",\"field_type\":{\"type\":\"BIGINT\"}},{\"name\":\"f1\",\"field_type\":{\"type\":\"INTEGER\",\"nullable\":false}},{\"name\":\"f2\",\"field_type\":{\"type\":\"STRING\"}}]}",
         };
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlinkComplexTypeITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlinkComplexTypeITCase.java
@@ -323,7 +323,7 @@ void testMapTypesInLogTable() throws Exception {
                                 + "MAP['arr1', ARRAY[1, 2, 3]], "
                                 + "MAP[1, ROW(100, 'row1')]), "
                                 + "(2, "
-                                + "MAP[3, CAST(NULL AS STRING)], "
+                                + "MAP[3, 'three'], "
                                 + "MAP['k3', MAP[40, 'v4']], "
                                 + "MAP['arr3', ARRAY[6]], "
                                 + "MAP[3, ROW(300, 'row3')]), "
@@ -338,7 +338,7 @@ void testMapTypesInLogTable() throws Exception {
         List<String> expectedRows =
                 Arrays.asList(
                         "+I[1, {1=one}, {k1={10=v1}}, {arr1=[1, 2, 3]}, {1=+I[100, row1]}]",
-                        "+I[2, {3=null}, {k3={40=v4}}, {arr3=[6]}, {3=+I[300, row3]}]",
+                        "+I[2, {3=three}, {k3={40=v4}}, {arr3=[6]}, {3=+I[300, row3]}]",
                         "+I[3, null, {k4={50=v5}}, {arr4=[7, 8, 9]}, {4=+I[400, row4]}]");
         assertResultsIgnoreOrder(rowIter, expectedRows, true);
     }
@@ -376,6 +376,36 @@ void testMapTypesInPartitionedLogTable() throws Exception {
         assertResultsIgnoreOrder(rowIter, expectedRows, true);
     }
 
+    @Test
+    void testMapWithNullValue() throws Exception {
+        tEnv.executeSql(
+                "create table map_null_test ("
+                        + "id int, "
+                        + "simple_map map<int, string>, "
+                        + "nested_map map<string, map<int, string>>, "
+                        + "map_with_array_value map<string, array<int>>"
+                        + ") with ('bucket.num' = '3')");
+
+        tEnv.executeSql(
+                        "INSERT INTO map_null_test VALUES "
+                                + "(1, "
+                                + "MAP[1, CAST(NULL AS STRING)], "
+                                + "MAP['k1', MAP[10, CAST(NULL AS STRING)]], "
+                                + "MAP['arr1', CAST(NULL AS ARRAY<INT>)]), "
+                                + "(2, "
+                                + "MAP[2, 'two', 3, CAST(NULL AS STRING)], "
+                                + "MAP['k2', CAST(NULL AS MAP<INT, STRING>)], "
+                                + "MAP['arr2', ARRAY[1, 2], 'arr3', CAST(NULL AS ARRAY<INT>)])")
+                .await();
+
+        CloseableIterator<Row> rowIter = tEnv.executeSql("select * from map_null_test").collect();
+        List<String> expectedRows =
+                Arrays.asList(
+                        "+I[1, {1=null}, {k1={10=null}}, {arr1=null}]",
+                        "+I[2, {2=two, 3=null}, {k2=null}, {arr2=[1, 2], arr3=null}]");
+        assertResultsIgnoreOrder(rowIter, expectedRows, true);
+    }
+
     @Test
     void testExceptionsForArrayTypeUsage() {
         assertThatThrownBy(

Original file line number	Diff line number	Diff line change
`@@ -285,7 +285,8 @@ private static DataType deserializeMap(JsonNode dataTypeNode) {`
`285`	`285`	`final DataType keyType = DataTypeJsonSerde.INSTANCE.deserialize(keyNode);`
`286`	`286`	`final JsonNode valueNode = dataTypeNode.get(FIELD_NAME_VALUE_TYPE);`
`287`	`287`	`final DataType valueType = DataTypeJsonSerde.INSTANCE.deserialize(valueNode);`
`288`		`- return new MapType(keyType, valueType);`
	`288`	`+ final DataType nonNullableKeyType = keyType.isNullable() ? keyType.copy(false) : keyType;`
	`289`	`+ return new MapType(nonNullableKeyType, valueType);`
`289`	`290`	`}`
`290`	`291`
`291`	`292`	`private static DataType deserializeRow(JsonNode dataTypeNode) {`