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 c29ddcb7f0 [format][python] Coalesce map blob metadata reads (#10121)
c29ddcb7f0 is described below
commit c29ddcb7f0a81a45a2c870b539623f263444d21a
Author: XiaoHongbo <[email protected]>
AuthorDate: Wed Sep 23 21:58:30 2026 +0800
[format][python] Coalesce map blob metadata reads (#10121)
---
.../format/blob/MapBlobElementSerializer.java | 22 +++++---
.../paimon/format/blob/BlobFileFormatTest.java | 60 ++++++++++++++++++++++
.../pypaimon/read/reader/format_blob_reader.py | 8 +--
paimon-python/pypaimon/tests/blob_test.py | 43 ++++++++++++++++
paimon-python/pypaimon/tests/native_commit_test.py | 17 ++++++
paimon-python/pypaimon/write/native_commit.py | 13 +++--
6 files changed, 148 insertions(+), 15 deletions(-)
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 9d45b02ad9..0640acbc76 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
@@ -39,9 +39,13 @@ import org.apache.paimon.utils.DeltaVarintCompressor;
import org.apache.paimon.utils.IOUtils;
import org.apache.paimon.utils.Preconditions;
+import org.apache.paimon.shade.guava30.com.google.common.io.ByteStreams;
+
import javax.annotation.Nullable;
+import java.io.BufferedInputStream;
import java.io.IOException;
+import java.io.InputStream;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.HashSet;
@@ -320,13 +324,15 @@ final class MapBlobElementSerializer implements
BlobElementSerializer {
long valueIndexStart = indexLengthsPosition - valueIndexLength;
long keyIndexStart = valueIndexStart - keyIndexLength;
- byte[] keyIndexBytes = new byte[keyIndexLength];
in.seek(keyIndexStart);
- IOUtils.readFully(in, keyIndexBytes);
+ InputStream indexes =
+ new BufferedInputStream(
+ ByteStreams.limit(in, (long) keyIndexLength +
valueIndexLength));
+ byte[] keyIndexBytes = new byte[keyIndexLength];
+ IOUtils.readFully(indexes, keyIndexBytes);
byte[] valueIndexBytes = new byte[valueIndexLength];
- in.seek(valueIndexStart);
- IOUtils.readFully(in, valueIndexBytes);
+ IOUtils.readFully(indexes, valueIndexBytes);
long[] keyLengths;
try {
@@ -347,7 +353,9 @@ final class MapBlobElementSerializer implements
BlobElementSerializer {
// 2. deserialize keys
Object[] keys = new Object[entryCount];
- long keyOffset = dataStart;
+ in.seek(dataStart);
+ // Limit read-ahead to keys so descriptor reads never fetch
BLOB values.
+ InputStream keyData = new
BufferedInputStream(ByteStreams.limit(in, keyDataLength));
for (int i = 0; i < entryCount; i++) {
long keyLength = keyLengths[i];
Object key;
@@ -355,14 +363,12 @@ final class MapBlobElementSerializer implements
BlobElementSerializer {
key = null;
} else {
byte[] keyBytes = new byte[(int) keyLength];
- in.seek(keyOffset);
- IOUtils.readFully(in, keyBytes);
+ IOUtils.readFully(keyData, keyBytes);
try {
key = keySerializer.deserialize(keyBytes);
} catch (RuntimeException e) {
throw new IllegalArgumentException("Invalid MAP<X,
BLOB> key.", e);
}
- keyOffset += keyLength;
}
keys[i] = key;
}
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 4431f0758a..afcf6ca98a 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
@@ -667,6 +667,63 @@ public class BlobFileFormatTest {
.hasMessage("Invalid MAP<X, BLOB> payload: duplicate key.");
}
+ @Test
+ public void testMapDescriptorReadsCoalesceMetadata() throws IOException {
+ for (int entryCount : new int[] {32, 4097}) {
+ TrackingLocalFileIO trackingIO = new TrackingLocalFileIO();
+ RowType rowType = RowType.of(DataTypes.MAP(DataTypes.INT(),
DataTypes.BLOB()));
+ Map<Object, Object> entries = new LinkedHashMap<>();
+ entries.put(null, null);
+ for (int i = 0; i < entryCount; i++) {
+ entries.put(i, new BlobData(i == 0 ? new byte[0] : new byte[]
{1, 2, 3}));
+ }
+ Path mapFile = new Path(parent, UUID.randomUUID().toString());
+ BlobFileFormat format =
+ new BlobFileFormat(true,
BlobFormatWriter.DEFAULT_COPY_BUFFER_SIZE);
+ try (PositionOutputStream out =
trackingIO.newOutputStream(mapFile, false)) {
+ FormatWriter writer =
format.createWriterFactory(rowType).create(out, null);
+ writer.addElement(GenericRow.of(new GenericMap(entries)));
+ writer.close();
+ }
+
+ FormatReaderContext context =
+ new FormatReaderContext(
+ trackingIO, mapFile,
trackingIO.getFileSize(mapFile), null, null);
+ List<InternalRow> rows = new ArrayList<>();
+ try (FileRecordReader<InternalRow> reader =
+ format.createReaderFactory(null, rowType,
null).createReader(context)) {
+ reader.forEachRemaining(rows::add);
+ }
+ GenericMap result = (GenericMap) rows.get(0).getMap(0);
+ assertThat(result.size()).isEqualTo(entryCount + 1);
+ assertThat(result.get(null)).isNull();
+ long valueStart = 4 + 9 + (long) entryCount * Integer.BYTES;
+ long valueEnd = valueStart + (entryCount - 1L) * 3;
+ for (int i = 0; i < entryCount; i++) {
+ Blob blob = (Blob) result.get(i);
+ assertThat(blob).isInstanceOf(BlobRef.class);
+ assertThat(blob.toDescriptor().offset())
+ .isEqualTo(valueStart + Math.max(0, i - 1L) * 3);
+ assertThat(blob.toDescriptor().length()).isEqualTo(i == 0 ? 0
: 3);
+ }
+ List<long[]> ranges = trackingIO.lastInputStream.readRanges;
+ if (entryCount == 32) {
+ // File footer/index plus map header, lengths, combined
indexes and keys.
+ assertThat(ranges).hasSize(6);
+ } else {
+ // Metadata larger than the buffer is read in bounded chunks,
not per key.
+ assertThat(ranges.size()).isLessThan(20);
+ }
+ for (long[] range : ranges) {
+ assertThat(range[0] >= valueEnd || range[0] + range[1] <=
valueStart)
+ .as(
+ "metadata range [%s, %s) must not read values",
+ range[0], range[0] + range[1])
+ .isTrue();
+ }
+ }
+ }
+
@Test
public void testMapBlobSupportedKeyTypes() throws IOException {
DataType[] keyTypes =
@@ -1185,6 +1242,7 @@ public class BlobFileFormatTest {
private int closeCount;
private int readCount;
private int seekCount;
+ private final List<long[]> readRanges = new ArrayList<>();
private TrackingSeekableInputStream(SeekableInputStream delegate) {
this.delegate = delegate;
@@ -1212,8 +1270,10 @@ public class BlobFileFormatTest {
@Override
public int read(byte[] bytes, int offset, int length) throws
IOException {
+ long position = delegate.getPos();
int read = delegate.read(bytes, offset, length);
if (read > 0) {
+ readRanges.add(new long[] {position, read});
readCount += read;
}
return read;
diff --git a/paimon-python/pypaimon/read/reader/format_blob_reader.py
b/paimon-python/pypaimon/read/reader/format_blob_reader.py
index a469102c72..e39e08b7b2 100644
--- a/paimon-python/pypaimon/read/reader/format_blob_reader.py
+++ b/paimon-python/pypaimon/read/reader/format_blob_reader.py
@@ -676,11 +676,13 @@ class BlobRecordIterator:
value_index_start = index_lengths_position - value_index_length
key_index_start = value_index_start - key_index_length
stream.seek(key_index_start)
- key_index_bytes = self._read_fully_from(stream, key_index_length)
+ # The two indexes are adjacent; read both without touching BLOB
values.
+ index_bytes = self._read_fully_from(
+ stream, key_index_length + value_index_length)
+ key_index_bytes = index_bytes[:key_index_length]
if len(key_index_bytes) != key_index_length:
raise IOError("Invalid MAP<X, BLOB> payload: cannot read key
index")
- stream.seek(value_index_start)
- value_index_bytes = self._read_fully_from(stream,
value_index_length)
+ value_index_bytes = index_bytes[key_index_length:]
if len(value_index_bytes) != value_index_length:
raise IOError("Invalid MAP<X, BLOB> payload: cannot read value
index")
diff --git a/paimon-python/pypaimon/tests/blob_test.py
b/paimon-python/pypaimon/tests/blob_test.py
index 7e25b720a3..21f2dc14e7 100644
--- a/paimon-python/pypaimon/tests/blob_test.py
+++ b/paimon-python/pypaimon/tests/blob_test.py
@@ -5029,6 +5029,49 @@ class BlobEndToEndTest(unittest.TestCase):
self.assertEqual(calls[0][1], 4)
parallel_reader.close()
+ def
test_map_blob_descriptor_coalesces_indexes_without_reading_values(self):
+ field = DataField(
+ 0, "blob_map", MapType(True, AtomicType("STRING"),
AtomicType("BLOB")))
+ value_data = b"payload" * 1024
+ key_lengths = [1, 1, 1]
+ value_lengths = [len(value_data), -1, 0]
+ payload = self._map_blob_payload(
+ b"abc", value_data, key_lengths, value_lengths)
+ prefix = b"preceding row"
+ value_start = len(prefix) + BlobRecordIterator.MAP_HEADER_SIZE + 3
+ index_start = value_start + len(value_data)
+ index_length = (len(DeltaVarintCompressor.compress(key_lengths))
+ + len(DeltaVarintCompressor.compress(value_lengths)))
+ reads = []
+
+ class TrackingStream(io.BytesIO):
+ def read(self, size=-1):
+ start = self.tell()
+ reads.append((start, size))
+ if start < index_start and start + size > value_start:
+ raise AssertionError("Descriptor read touched BLOB value
data")
+ return super().read(size)
+
+ with TrackingStream(prefix + payload) as stream:
+ iterator = BlobRecordIterator(
+ None, "test.blob", [], [], field,
+ input_stream=stream, blob_as_descriptor=True)
+ result = iterator._read_blob_map(len(prefix), len(payload))
+ self.assertEqual(list(result), ['a', 'b', 'c'])
+ descriptor = result['a'].to_descriptor()
+ self.assertEqual((descriptor.offset, descriptor.length),
+ (value_start, len(value_data)))
+ self.assertIsNone(result['b'])
+ self.assertEqual(result['c'].to_descriptor().length, 0)
+
+ self.assertEqual(reads, [
+ (len(prefix), BlobRecordIterator.MAP_HEADER_SIZE),
+ (len(prefix) + len(payload) -
BlobRecordIterator.MAP_INDEX_LENGTHS_SIZE,
+ BlobRecordIterator.MAP_INDEX_LENGTHS_SIZE),
+ (index_start, index_length),
+ (len(prefix) + BlobRecordIterator.MAP_HEADER_SIZE, 3),
+ ])
+
def test_map_blob_consumer_descriptors_and_flush(self):
from pypaimon.write.blob_format_writer import BlobFormatWriter
diff --git a/paimon-python/pypaimon/tests/native_commit_test.py
b/paimon-python/pypaimon/tests/native_commit_test.py
index 67c9656c39..e4936ae447 100644
--- a/paimon-python/pypaimon/tests/native_commit_test.py
+++ b/paimon-python/pypaimon/tests/native_commit_test.py
@@ -492,6 +492,23 @@ def
test_incompatible_publication_environment_is_not_reconstructed(tmp_path, kin
resolve.assert_not_called()
[email protected]('missing_type,missing_method', [
+ ('Table', 'from_resolved_schema'),
+ ('CommitMessage', 'deserialize'),
+ ('StreamWriteBuilder', 'with_commit_user'),
+ ('BatchWriteBuilder', '_with_commit_user'),
+ ('BatchWriteBuilder', 'with_overwrite'),
+])
+def test_incomplete_runtime_falls_back_without_reconstructing_table(
+ tmp_path, missing_type, missing_method):
+ table = _table(tmp_path)
+ with patch('pypaimon.write.native_commit.native_method_available',
+ side_effect=lambda cls, method: (cls, method) != (missing_type,
missing_method)), \
+
patch('pypaimon.write.native_commit._resolved_schema_file_io_options') as
resolve:
+ assert create_native_commit(table, 'job') is None
+ resolve.assert_not_called()
+
+
def test_missing_runtime_falls_back_without_reconstructing_table(tmp_path):
table = _table(tmp_path)
with patch('pypaimon.write.native_commit.native_commit_available',
return_value=False), \
diff --git a/paimon-python/pypaimon/write/native_commit.py
b/paimon-python/pypaimon/write/native_commit.py
index 86fc008942..97d5316582 100644
--- a/paimon-python/pypaimon/write/native_commit.py
+++ b/paimon-python/pypaimon/write/native_commit.py
@@ -19,14 +19,19 @@
from pypaimon.common.json_util import JSON
from pypaimon.read.native_plan import (
- _option_value_to_string, _resolved_schema_file_io_options)
+ _option_value_to_string, _resolved_schema_file_io_options,
native_method_available)
from pypaimon.write.commit_message_serializer import serialize_commit_message
def native_commit_available() -> bool:
- """Whether the optional Rust runtime is installed."""
- from importlib.util import find_spec
- return find_spec('pypaimon_rust') is not None
+ """Whether the Rust runtime provides the required commit APIs."""
+ return all(native_method_available(type_name, method) for type_name,
method in (
+ ('Table', 'from_resolved_schema'),
+ ('CommitMessage', 'deserialize'),
+ ('StreamWriteBuilder', 'with_commit_user'),
+ ('BatchWriteBuilder', '_with_commit_user'),
+ ('BatchWriteBuilder', 'with_overwrite'),
+ ))
def native_messages_supported(table, messages) -> bool:
