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 eed76ed708 [format][python] Reject duplicate MAP BLOB keys (#9107)
eed76ed708 is described below
commit eed76ed7085db83cafef51f4c0ce2b02e36069cb
Author: QuakeWang <[email protected]>
AuthorDate: Mon Aug 10 16:44:13 2026 +0800
[format][python] Reject duplicate MAP BLOB keys (#9107)
---
.../test/java/org/apache/paimon/JavaPyE2ETest.java | 3 -
.../format/blob/MapBlobElementSerializer.java | 48 ++++++++++----
.../paimon/format/blob/BlobFileFormatTest.java | 76 +++++++++++-----------
.../pypaimon/read/reader/format_blob_reader.py | 2 +
paimon-python/pypaimon/tests/blob_table_test.py | 6 +-
paimon-python/pypaimon/tests/blob_test.py | 45 +++++++++++--
.../pypaimon/tests/e2e/java_py_read_write_test.py | 1 -
paimon-python/pypaimon/write/blob_format_writer.py | 14 ++--
8 files changed, 128 insertions(+), 67 deletions(-)
diff --git a/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
b/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
index 69de8c06f9..b6801b4a7b 100644
--- a/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
@@ -1494,9 +1494,6 @@ public class JavaPyE2ETest {
Map<Object, Object> timePayloads = new LinkedHashMap<>();
timePayloads.put(45_296_789, new
BlobData("java-time".getBytes(StandardCharsets.UTF_8)));
Map<Object, Object> binaryPayloads = new LinkedHashMap<>();
- binaryPayloads.put(
- new byte[] {0, (byte) 0xff, 1, 2},
- new
BlobData("java-binary-first".getBytes(StandardCharsets.UTF_8)));
binaryPayloads.put(
new byte[] {0, (byte) 0xff, 1, 2},
new BlobData("java-binary".getBytes(StandardCharsets.UTF_8)));
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/blob/MapBlobElementSerializer.java
b/paimon-format/src/main/java/org/apache/paimon/format/blob/MapBlobElementSerializer.java
index b9fb6a74b9..9d45b02ad9 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/blob/MapBlobElementSerializer.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/blob/MapBlobElementSerializer.java
@@ -44,8 +44,10 @@ import javax.annotation.Nullable;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
+import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.Map;
+import java.util.Set;
import static org.apache.paimon.utils.StreamUtils.intToLittleEndian;
import static org.apache.paimon.utils.StreamUtils.longToLittleEndian;
@@ -176,25 +178,41 @@ final class MapBlobElementSerializer implements
BlobElementSerializer {
"MAP<X, BLOB> key/value array size does not match map
size.");
}
- long recordPosition = startRecord();
- // 1. Write meta
- write(MAGIC_NUMBER_BYTES);
- write(new byte[] {VERSION});
- write(intToLittleEndian(map.size()));
-
- // 2. Write key array
+ // 1. Serialize and validate the key array before writing the
record.
long[] keyLengths = new long[map.size()];
+ byte[][] keyBytes = new byte[map.size()][];
+ Set<ByteBuffer> seenKeys = new HashSet<>();
for (int i = 0; i < map.size(); i++) {
if (keys.isNullAt(i)) {
keyLengths[i] = NULL_KEY_LENGTH;
+ if (!seenKeys.add(null)) {
+ throw new IllegalArgumentException("MAP<X, BLOB> keys
must be unique.");
+ }
} else {
- byte[] keyBytes =
keySerializer.serialize(keyGetter.getElementOrNull(keys, i));
- keyLengths[i] = keyBytes.length;
- write(keyBytes);
+ byte[] serializedKey =
+
keySerializer.serialize(keyGetter.getElementOrNull(keys, i));
+ keyLengths[i] = serializedKey.length;
+ keyBytes[i] = serializedKey;
+ if (!seenKeys.add(ByteBuffer.wrap(serializedKey))) {
+ throw new IllegalArgumentException("MAP<X, BLOB> keys
must be unique.");
+ }
}
}
- // 3. Write values(blobs) array, same as ArrayBlobElementWriter
+ long recordPosition = startRecord();
+ // 2. Write meta
+ write(MAGIC_NUMBER_BYTES);
+ write(new byte[] {VERSION});
+ write(intToLittleEndian(map.size()));
+
+ // 3. Write key array
+ for (byte[] serializedKey : keyBytes) {
+ if (serializedKey != null) {
+ write(serializedKey);
+ }
+ }
+
+ // 4. Write values(blobs) array, same as ArrayBlobElementWriter
long[] valueLengths = new long[map.size()];
boolean flush = false;
for (int i = 0; i < map.size(); i++) {
@@ -364,7 +382,13 @@ final class MapBlobElementSerializer implements
BlobElementSerializer {
}
map.put(keys[i], value);
}
- return binaryKey ? GenericMap.fromBinaryKeyMap(map) : new
GenericMap(map);
+ GenericMap result =
+ binaryKey ? GenericMap.fromBinaryKeyMap(map) : new
GenericMap(map);
+ if (result.size() != entryCount) {
+ throw new IllegalArgumentException(
+ "Invalid MAP<X, BLOB> payload: duplicate key.");
+ }
+ return result;
} catch (IOException e) {
throw new RuntimeException(e);
}
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFileFormatTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFileFormatTest.java
index bd11580bb0..a9a2d2f99c 100644
---
a/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFileFormatTest.java
+++
b/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFileFormatTest.java
@@ -442,47 +442,54 @@ public class BlobFileFormatTest {
}
@Test
- public void testDuplicateMapBlobKeyLastWinsInline() throws IOException {
- assertDuplicateMapBlobKeyLastWins(
+ public void testRejectDuplicateBinaryMapBlobKeyOnWrite() throws
IOException {
+ RowType rowType = RowType.of(DataTypes.MAP(DataTypes.BYTES(),
DataTypes.BLOB()));
+ Map<Object, Object> entries = new LinkedHashMap<>();
+ entries.put(new byte[] {1}, new BlobData("first".getBytes()));
+ entries.put(new byte[] {1}, new BlobData("second".getBytes()));
+
+ try (PositionOutputStream out = fileIO.newOutputStream(file, false)) {
+ FormatWriter writer =
+ new BlobFileFormat(false,
BlobFormatWriter.DEFAULT_COPY_BUFFER_SIZE)
+ .createWriterFactory(rowType)
+ .create(out, null);
+ assertThatThrownBy(() -> writer.addElement(GenericRow.of(new
GenericMap(entries))))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("MAP<X, BLOB> keys must be unique.");
+ writer.close();
+ }
+ }
+
+ @Test
+ public void testRejectDuplicateMapBlobKeyInline() throws IOException {
+ assertDuplicateMapBlobKeyRejected(
false,
DataTypes.STRING(),
BinaryString.fromString("a"),
BinaryString.fromString("b"),
- BinaryString.fromString("a"),
(byte) 'a');
}
@Test
- public void testDuplicateMapBlobKeyLastWinsAsDescriptor() throws
IOException {
- assertDuplicateMapBlobKeyLastWins(
+ public void testRejectDuplicateMapBlobKeyAsDescriptor() throws IOException
{
+ assertDuplicateMapBlobKeyRejected(
true,
DataTypes.STRING(),
BinaryString.fromString("a"),
BinaryString.fromString("b"),
- BinaryString.fromString("a"),
(byte) 'a');
}
@Test
- public void testDuplicateBinaryMapBlobKeyLastWinsInline() throws
IOException {
- assertDuplicateMapBlobKeyLastWins(
- false,
- DataTypes.BINARY(1),
- new byte[] {1},
- new byte[] {2},
- new byte[] {1},
- (byte) 1);
+ public void testRejectDuplicateBinaryMapBlobKeyInline() throws IOException
{
+ assertDuplicateMapBlobKeyRejected(
+ false, DataTypes.BINARY(1), new byte[] {1}, new byte[] {2},
(byte) 1);
}
@Test
- public void testDuplicateBinaryMapBlobKeyLastWinsAsDescriptor() throws
IOException {
- assertDuplicateMapBlobKeyLastWins(
- true,
- DataTypes.BINARY(1),
- new byte[] {1},
- new byte[] {2},
- new byte[] {1},
- (byte) 1);
+ public void testRejectDuplicateBinaryMapBlobKeyAsDescriptor() throws
IOException {
+ assertDuplicateMapBlobKeyRejected(
+ true, DataTypes.BINARY(1), new byte[] {1}, new byte[] {2},
(byte) 1);
}
@Test
@@ -564,12 +571,11 @@ public class BlobFileFormatTest {
assertThat(genericMap.keyArray().getBinary(0)).isEqualTo(new byte[]
{1});
}
- private void assertDuplicateMapBlobKeyLastWins(
+ private void assertDuplicateMapBlobKeyRejected(
boolean blobAsDescriptor,
DataType keyType,
Object firstKey,
Object secondKey,
- Object lookupKey,
byte duplicateKeyByte)
throws IOException {
Map<Object, Object> entries = new LinkedHashMap<>();
@@ -599,19 +605,15 @@ public class BlobFileFormatTest {
FormatReaderFactory readerFactory = format.createReaderFactory(null,
rowType, null);
FormatReaderContext context =
new FormatReaderContext(fileIO, file,
fileIO.getFileSize(file));
- List<InternalRow> rows = new ArrayList<>();
- try (FileRecordReader<InternalRow> reader =
readerFactory.createReader(context)) {
- reader.forEachRemaining(rows::add);
- }
-
- assertThat(rows).hasSize(1);
- GenericMap result = (GenericMap) rows.get(0).getMap(0);
- assertThat(result.size()).isOne();
- assertThat(result.contains(lookupKey)).isTrue();
- assertMapBlob(result.get(lookupKey), blobAsDescriptor,
"second".getBytes());
- if (lookupKey instanceof byte[]) {
- assertThat(result.keyArray().getBinary(0)).isEqualTo(lookupKey);
- }
+ assertThatThrownBy(
+ () -> {
+ try (FileRecordReader<InternalRow> reader =
+ readerFactory.createReader(context)) {
+ reader.forEachRemaining(ignored -> {});
+ }
+ })
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid MAP<X, BLOB> payload: duplicate key.");
}
@Test
diff --git a/paimon-python/pypaimon/read/reader/format_blob_reader.py
b/paimon-python/pypaimon/read/reader/format_blob_reader.py
index 897a81f67a..1e403f53a0 100644
--- a/paimon-python/pypaimon/read/reader/format_blob_reader.py
+++ b/paimon-python/pypaimon/read/reader/format_blob_reader.py
@@ -659,6 +659,8 @@ class BlobRecordIterator:
value_offset += value_length
value_data_offset += value_length
result[key] = value
+ if len(result) != entry_count:
+ raise ValueError("Invalid MAP<X, BLOB> payload: duplicate
key.")
return result
finally:
if close_stream:
diff --git a/paimon-python/pypaimon/tests/blob_table_test.py
b/paimon-python/pypaimon/tests/blob_table_test.py
index 1559f0efd7..4b7e99a937 100755
--- a/paimon-python/pypaimon/tests/blob_table_test.py
+++ b/paimon-python/pypaimon/tests/blob_table_test.py
@@ -1471,7 +1471,7 @@ class DedicatedFormatWriterTest(unittest.TestCase):
[],
[('last', b'blob-4')],
[('descriptor', source_descriptor.serialize())],
- [('duplicate', b'first'), ('duplicate', b'last')],
+ [('first', b'first'), ('last', b'last')],
],
type=map_blob_type,
),
@@ -1497,7 +1497,7 @@ class DedicatedFormatWriterTest(unittest.TestCase):
3: {},
4: {'last': b'blob-4'},
5: {'descriptor': descriptor_body},
- 6: {'duplicate': b'last'},
+ 6: {'first': b'first', 'last': b'last'},
},
)
@@ -1576,7 +1576,7 @@ class DedicatedFormatWriterTest(unittest.TestCase):
3: {},
4: {'last': b'blob-4'},
5: {'descriptor': descriptor_body},
- 6: {'duplicate': b'last'},
+ 6: {'first': b'first', 'last': b'last'},
},
)
diff --git a/paimon-python/pypaimon/tests/blob_test.py
b/paimon-python/pypaimon/tests/blob_test.py
index 19aacce9e0..e5a14677cc 100644
--- a/paimon-python/pypaimon/tests/blob_test.py
+++ b/paimon-python/pypaimon/tests/blob_test.py
@@ -3126,7 +3126,28 @@ class BlobEndToEndTest(unittest.TestCase):
placeholder_reader.read_arrow_batch()
placeholder_reader.close()
- def test_duplicate_map_blob_key_last_wins(self):
+ def test_reject_duplicate_map_blob_key_on_write(self):
+ from pypaimon.write.blob_format_writer import BlobFormatWriter
+
+ fields = [DataField(
+ 0,
+ "blob_map",
+ MapType(True, AtomicType("STRING"), AtomicType("BLOB")),
+ )]
+ output = io.BytesIO()
+ writer = BlobFormatWriter(output)
+ with self.assertRaisesRegex(ValueError, "MAP<X, BLOB> keys must be
unique"):
+ writer.add_element(GenericRow(
+ [[
+ ("duplicate", BlobData(b"first")),
+ ("duplicate", BlobData(b"second")),
+ ]],
+ fields,
+ RowKind.INSERT,
+ ))
+ self.assertEqual(output.getvalue(), b"")
+
+ def test_reject_duplicate_map_blob_key_payload(self):
from pypaimon.write.blob_format_writer import BlobFormatWriter
file_io = LocalFileIO(self.temp_dir, Options({}))
@@ -3140,7 +3161,7 @@ class BlobEndToEndTest(unittest.TestCase):
writer.add_element(GenericRow(
[[
("duplicate", BlobData(b"first")),
- ("duplicate", BlobData(b"second")),
+ ("duplicatE", BlobData(b"second")),
("tail", BlobData(b"third")),
]],
fields,
@@ -3149,6 +3170,17 @@ class BlobEndToEndTest(unittest.TestCase):
record_length = writer.lengths[0]
writer.close()
+ with open(blob_file_path, 'r+b') as blob_file:
+ second_key_position = (
+ BlobRecordIterator.MAGIC_NUMBER_SIZE
+ + BlobRecordIterator.MAP_HEADER_SIZE
+ + len("duplicate")
+ )
+ bytes_data = bytearray(blob_file.read())
+ bytes_data[second_key_position + len("duplicatE") - 1] = ord("e")
+ blob_file.seek(0)
+ blob_file.write(bytes_data)
+
for blob_as_descriptor in (False, True):
iterator = BlobRecordIterator(
file_io,
@@ -3158,10 +3190,11 @@ class BlobEndToEndTest(unittest.TestCase):
fields[0],
blob_as_descriptor=blob_as_descriptor,
)
- result = next(iterator).values[0]
- self.assertEqual(list(result), ["duplicate", "tail"])
- self.assertEqual(result["duplicate"].to_data(), b"second")
- self.assertEqual(result["tail"].to_data(), b"third")
+ with self.subTest(blob_as_descriptor=blob_as_descriptor):
+ with self.assertRaisesRegex(
+ ValueError, "Invalid MAP<X, BLOB> payload: duplicate key"
+ ):
+ next(iterator)
def test_map_blob_key_types_and_rejections(self):
from pypaimon.common.map_blob_key_serializer import
create_map_blob_key_serializer
diff --git a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
index fbf619a6f7..d14e1f7459 100644
--- a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
+++ b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
@@ -1663,7 +1663,6 @@ class JavaPyReadWriteTest(unittest.TestCase):
'binary_payloads': pa.array(
[
[
- (bytes([0, 255, 1, 2]), b'python-binary-first'),
(bytes([0, 255, 1, 2]), b'python-binary'),
],
None,
diff --git a/paimon-python/pypaimon/write/blob_format_writer.py
b/paimon-python/pypaimon/write/blob_format_writer.py
index b68600c129..8a7969b271 100644
--- a/paimon-python/pypaimon/write/blob_format_writer.py
+++ b/paimon-python/pypaimon/write/blob_format_writer.py
@@ -215,13 +215,17 @@ class BlobFormatWriter:
key_serializer = create_map_blob_key_serializer(key_type)
key_bytes = []
+ seen_keys = set()
for key, _ in entries:
if key is None:
- key_bytes.append(None)
- continue
- serialized = key_serializer.serialize(key)
- if len(serialized) > 0x7fffffff:
- raise ValueError(f"MAP<X, BLOB> key is too large:
{len(serialized)}")
+ serialized = None
+ else:
+ serialized = key_serializer.serialize(key)
+ if len(serialized) > 0x7fffffff:
+ raise ValueError(f"MAP<X, BLOB> key is too large:
{len(serialized)}")
+ if serialized in seen_keys:
+ raise ValueError("MAP<X, BLOB> keys must be unique.")
+ seen_keys.add(serialized)
key_bytes.append(serialized)
for _, blob_value in entries:
