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 9d45b02ad964..0640acbc7661 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.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,7 +353,9 @@ 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; @@ -355,14 +363,12 @@ public Object read(long payloadPosition, long payloadLength) { 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 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 4431f0758a5c..afcf6ca98a08 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 @@ private void assertDuplicateMapBlobKeyRejected( .hasMessage("Invalid MAP 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 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 rows = new ArrayList<>(); + try (FileRecordReader 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 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 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; diff --git a/paimon-python/pypaimon/read/reader/format_blob_reader.py b/paimon-python/pypaimon/read/reader/format_blob_reader.py index a469102c7219..e39e08b7b229 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 @@ 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 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 payload: cannot read value index") diff --git a/paimon-python/pypaimon/tests/blob_test.py b/paimon-python/pypaimon/tests/blob_test.py index 7e25b720a371..21f2dc14e780 100644 --- a/paimon-python/pypaimon/tests/blob_test.py +++ b/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 diff --git a/paimon-python/pypaimon/tests/native_commit_test.py b/paimon-python/pypaimon/tests/native_commit_test.py index 67c9656c3962..e4936ae44786 100644 --- a/paimon-python/pypaimon/tests/native_commit_test.py +++ b/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), \ diff --git a/paimon-python/pypaimon/write/native_commit.py b/paimon-python/pypaimon/write/native_commit.py index 86fc00894218..97d531658209 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: