This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new b02d8f0707 [spark] Reject null map keys before conversion (#10587)
b02d8f0707 is described below
commit b02d8f0707461452120de61f60f7b13fc986aaf7
Author: LsomeYeah <[email protected]>
AuthorDate: Fri Oct 16 08:50:07 2026 +0800
[spark] Reject null map keys before conversion (#10587)
---
.../org/apache/paimon/spark/DataConverter.java | 18 ++++----
.../apache/paimon/spark/SparkInternalRowTest.java | 18 ++++++--
.../org/apache/paimon/spark/SparkReadITCase.java | 49 ++++++++++++++++++++++
3 files changed, 73 insertions(+), 12 deletions(-)
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/DataConverter.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/DataConverter.java
index d34a9c7da7..b7ae9849ea 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/DataConverter.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/DataConverter.java
@@ -37,6 +37,7 @@ import org.apache.paimon.types.MapType;
import org.apache.paimon.types.MultisetType;
import org.apache.paimon.types.RowType;
import org.apache.paimon.types.VectorType;
+import org.apache.paimon.utils.Preconditions;
import org.apache.spark.sql.catalyst.util.ArrayBasedMapData;
import org.apache.spark.sql.catalyst.util.ArrayData;
@@ -154,18 +155,17 @@ public class DataConverter {
throw new UnsupportedOperationException("Unsupported type: " +
mapType);
}
- if (valueType.getTypeRoot() ==
org.apache.paimon.types.DataTypeRoot.BLOB) {
- InternalArray keys = map.keyArray();
- for (int i = 0; i < keys.size(); i++) {
- if (keys.isNullAt(i)) {
- throw new IllegalArgumentException(
- "Spark MAP<X, BLOB> does not support null keys.");
- }
- }
+ InternalArray keys = map.keyArray();
+ String nullKeyErrorMessage =
+ valueType.getTypeRoot() ==
org.apache.paimon.types.DataTypeRoot.BLOB
+ ? "Spark MAP<X, BLOB> does not support null keys."
+ : "Spark map data does not support null keys.";
+ for (int i = 0; i < keys.size(); i++) {
+ Preconditions.checkArgument(!keys.isNullAt(i),
nullKeyErrorMessage);
}
return new ArrayBasedMapData(
- fromPaimonArrayElementType(map.keyArray(), keyType),
+ fromPaimonArrayElementType(keys, keyType),
fromPaimonArrayElementType(map.valueArray(), valueType,
blobAsDescriptor));
}
}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkInternalRowTest.java
b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkInternalRowTest.java
index bde723cdd0..74225ffe0b 100644
---
a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkInternalRowTest.java
+++
b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkInternalRowTest.java
@@ -246,17 +246,29 @@ public class SparkInternalRowTest {
@Test
public void testMapBlobRejectsNullKey() {
- Map<Object, Object> map = new LinkedHashMap<>();
- map.put(null, Blob.fromData(new byte[] {1}));
+ Map<Object, Object> blobMap = new LinkedHashMap<>();
+ blobMap.put(null, Blob.fromData(new byte[] {1}));
assertThatThrownBy(
() ->
DataConverter.fromPaimon(
- new GenericMap(map),
+ new GenericMap(blobMap),
DataTypes.MAP(DataTypes.STRING(),
DataTypes.BLOB()),
false))
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Spark MAP<X, BLOB> does not support null keys.");
+
+ Map<Object, Object> intMap = new LinkedHashMap<>();
+ intMap.put(null, 1);
+
+ assertThatThrownBy(
+ () ->
+ DataConverter.fromPaimon(
+ new GenericMap(intMap),
+ DataTypes.MAP(DataTypes.INT(),
DataTypes.INT()),
+ false))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Spark map data does not support null keys.");
}
@Test
diff --git
a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkReadITCase.java
b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkReadITCase.java
index fbc2f660d4..a1c01a4be5 100644
---
a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkReadITCase.java
+++
b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkReadITCase.java
@@ -18,14 +18,23 @@
package org.apache.paimon.spark;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogFactory;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.GenericMap;
+import org.apache.paimon.data.GenericRow;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.FileStoreTableFactory;
import org.apache.paimon.types.ArrayType;
import org.apache.paimon.types.BigIntType;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.DoubleType;
import org.apache.paimon.types.VarCharType;
@@ -40,11 +49,13 @@ import org.junit.jupiter.api.Test;
import java.io.File;
import java.util.Arrays;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
+import static org.apache.paimon.options.CatalogOptions.WAREHOUSE;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -61,6 +72,44 @@ public class SparkReadITCase extends SparkReadTestBase {
innerTestNestedType(spark.table("t2"));
}
+ @Test
+ public void testReadOrcMapWithNullKey() throws Exception {
+ Schema schema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("m", DataTypes.MAP(DataTypes.INT(),
DataTypes.INT()))
+ .option("file.format", "orc")
+ .build();
+
+ Options options = new Options();
+ options.set(WAREHOUSE, warehousePath.toString());
+ try (Catalog catalog =
CatalogFactory.createCatalog(CatalogContext.create(options))) {
+ catalog.createTable(Identifier.create("default", "invalid_map"),
schema, false);
+ catalog.createTable(Identifier.create("default", "valid_map"),
schema, false);
+ }
+
+ Map<Integer, Integer> invalidMap = new LinkedHashMap<>();
+ invalidMap.put(null, 17);
+ invalidMap.put(2, 20);
+ writeTable("invalid_map", GenericRow.of(1, new
GenericMap(invalidMap)));
+
+ Map<Integer, Integer> validMap = new LinkedHashMap<>();
+ validMap.put(1, 17);
+ validMap.put(2, null);
+ writeTable("valid_map", GenericRow.of(2, new GenericMap(validMap)));
+
+ assertThat(spark.sql("SELECT id FROM
invalid_map").collectAsList().toString())
+ .isEqualTo("[[1]]");
+
+ Row validRow = spark.sql("SELECT id, m FROM
valid_map").collectAsList().get(0);
+ assertThat(validRow.getInt(0)).isEqualTo(2);
+ assertThat(validRow.<Integer, Integer>getJavaMap(1))
+ .containsExactlyInAnyOrderEntriesOf(validMap);
+
+ assertThatThrownBy(() -> spark.sql("SELECT id, m FROM
invalid_map").collectAsList())
+ .hasStackTraceContaining("Spark map data does not support null
keys.");
+ }
+
@Test
public void testFilterPushDown() {
innerTestSimpleTypeFilterPushDown(spark.table("t1"));
