diff --git a/paimon-python/dev/requirements.txt b/paimon-python/dev/requirements.txt index 0bd9f01b767d..341542724833 100644 --- a/paimon-python/dev/requirements.txt +++ b/paimon-python/dev/requirements.txt @@ -19,6 +19,7 @@ cachetools>=4.2,<6; python_version=="3.6" cachetools>=5,<6; python_version>"3.6" dataclasses>=0.8; python_version < "3.7" fastavro>=1.4,<2 +isal>=1.8,<2; python_version >= "3.9" and (platform_machine == "x86_64" or platform_machine == "AMD64" or platform_machine == "aarch64" or platform_machine == "arm64") fsspec>=2021.10,<2026; python_version<"3.8" fsspec>=2023,<2026; python_version>="3.8" packaging>=21,<26 diff --git a/paimon-python/pypaimon/tests/blob_test.py b/paimon-python/pypaimon/tests/blob_test.py index 86c7b5daa026..33fc5cbdd905 100644 --- a/paimon-python/pypaimon/tests/blob_test.py +++ b/paimon-python/pypaimon/tests/blob_test.py @@ -21,7 +21,9 @@ import struct import tempfile import unittest +import zlib from pathlib import Path +from unittest.mock import patch import pyarrow as pa @@ -1599,6 +1601,37 @@ def test_null_blob_write(self): self.assertEqual(writer.lengths, [-1]) self.assertEqual(writer.position, 0) + @staticmethod + def _write_blob_record_with_crc_backend(backend): + from pypaimon.write import blob_format_writer + + output = io.BytesIO() + payload = b'blob-crc-payload' * 1024 + with patch.object(blob_format_writer, 'crc_backend', backend): + writer = blob_format_writer.BlobFormatWriter(output) + writer.add_blob('blob_field', BlobData(payload)) + return output.getvalue(), payload + + def test_blob_crc_fallback_matches_zlib(self): + from pypaimon.write.blob_format_writer import BlobFormatWriter + + record, payload = self._write_blob_record_with_crc_backend(zlib) + expected_crc = zlib.crc32( + struct.pack(' int: - crc32 = zlib.crc32(data, crc32) + crc32 = crc_backend.crc32(data, crc32) self.output_stream.write(data) self.position += len(data) return crc32 diff --git a/paimon-python/pypaimon/write/table_write.py b/paimon-python/pypaimon/write/table_write.py index 629383409f94..59846bd59ea4 100644 --- a/paimon-python/pypaimon/write/table_write.py +++ b/paimon-python/pypaimon/write/table_write.py @@ -56,8 +56,18 @@ def write_arrow_batch(self, data: pa.RecordBatch): partition_bucket_groups[(tuple(partitions[i]), buckets[i])].append(i) for (partition, bucket), row_indices in partition_bucket_groups.items(): - indices_array = pa.array(row_indices, type=pa.int64()) - sub_table = pa.compute.take(data, indices_array) + if len(row_indices) == data.num_rows: + # Every input row belongs to the same partition/bucket. Passing the + # original batch through avoids copying large BLOB values through + # Arrow take before the dedicated BLOB writer consumes them. + sub_table = data + elif row_indices[-1] - row_indices[0] + 1 == len(row_indices): + # Contiguous groups can share the original Arrow buffers instead of + # gathering their rows into newly allocated buffers with take. + sub_table = data.slice(row_indices[0], len(row_indices)) + else: + indices_array = pa.array(row_indices, type=pa.int64()) + sub_table = pa.compute.take(data, indices_array) self.file_store_write.write(partition, bucket, sub_table) def write_row(self, row):