[format][python] Coalesce map blob metadata reads

paimon-format/src/main/java/org/apache/paimon/format/blob/MapBlobElementSerializer.java

@@ -39,9 +39,13 @@
 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 @@ public Object read(long payloadPosition, long payloadLength) {
                 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,22 +353,22 @@ public Object read(long payloadPosition, long payloadLength) {
 
                 // 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;
                     if (keyLength == NULL_KEY_LENGTH) {
                         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;
                 }

paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFileFormatTest.java

@@ -667,6 +667,63 @@ private void assertDuplicateMapBlobKeyRejected(
                 .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 @@ private static class TrackingSeekableInputStream extends SeekableInputStream {
         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 int read() throws IOException {
 
         @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;

paimon-python/pypaimon/read/reader/format_blob_reader.py

@@ -676,11 +676,13 @@ def _read_blob_map(self, position: int, length: int):
             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")
 

paimon-python/pypaimon/tests/blob_test.py

@@ -5029,6 +5029,49 @@ def read_blobs_concurrent(blobs, parallelism):
         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
 

paimon-python/pypaimon/tests/native_commit_test.py

@@ -492,6 +492,23 @@ class CustomEnvironment(CatalogEnvironment):
         resolve.assert_not_called()
 
 
+@pytest.mark.parametrize('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), \

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: