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 be34a5a8b0 [python] Fix missing compression for manifest files (#8376)
be34a5a8b0 is described below
commit be34a5a8b05dc75ffb88b78e49ed8ff084cbd711
Author: XiaoHongbo <[email protected]>
AuthorDate: Mon Jun 29 16:42:44 2026 +0800
[python] Fix missing compression for manifest files (#8376)
---
.../pypaimon/common/options/core_options.py | 10 +++++
paimon-python/pypaimon/manifest/__init__.py | 22 ++++++++++
.../pypaimon/manifest/index_manifest_file.py | 6 ++-
.../pypaimon/manifest/manifest_file_manager.py | 13 ++++--
.../pypaimon/manifest/manifest_list_manager.py | 6 ++-
.../tests/manifest/manifest_manager_test.py | 47 +++++++++++++++++++++-
.../tests/manifest/manifest_schema_test.py | 1 +
7 files changed, 98 insertions(+), 7 deletions(-)
diff --git a/paimon-python/pypaimon/common/options/core_options.py
b/paimon-python/pypaimon/common/options/core_options.py
index c601555464..56e323c997 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -186,6 +186,13 @@ class CoreOptions:
.with_description("The parallelism for scanning manifest files.")
)
+ MANIFEST_COMPRESSION: ConfigOption[str] = (
+ ConfigOptions.key("manifest.compression")
+ .string_type()
+ .default_value("zstd")
+ .with_description("Default file compression for manifest.")
+ )
+
MANIFEST_TARGET_FILE_SIZE: ConfigOption[MemorySize] = (
ConfigOptions.key("manifest.target-file-size")
.memory_type()
@@ -838,6 +845,9 @@ class CoreOptions:
def scan_manifest_parallelism(self, default=None):
return self.options.get(CoreOptions.SCAN_MANIFEST_PARALLELISM, default)
+ def manifest_compression(self, default=None):
+ return self.options.get(CoreOptions.MANIFEST_COMPRESSION, default)
+
def manifest_target_size(self, default=None):
if default is not None and not isinstance(default, MemorySize):
default = MemorySize.of_bytes(default) if isinstance(default, int)
else MemorySize.parse(default)
diff --git a/paimon-python/pypaimon/manifest/__init__.py
b/paimon-python/pypaimon/manifest/__init__.py
index 8cb828ad1d..88064e3a67 100644
--- a/paimon-python/pypaimon/manifest/__init__.py
+++ b/paimon-python/pypaimon/manifest/__init__.py
@@ -23,3 +23,25 @@ if sys.version_info[:2] == (3, 6):
from pypaimon.manifest import fastavro_py36_compat # noqa: F401
except ImportError:
pass
+
+
+_SUPPORTED_MANIFEST_COMPRESSIONS = {
+ 'null', 'deflate', 'snappy', 'zstd', 'bzip2', 'xz',
+}
+
+
+_DEFAULT_MANIFEST_CODEC = 'zstandard'
+
+
+def avro_codec(compression):
+ """Map Paimon manifest.compression config value to fastavro codec name."""
+ if not isinstance(compression, str):
+ return _DEFAULT_MANIFEST_CODEC
+ lower = compression.lower()
+ if lower not in _SUPPORTED_MANIFEST_COMPRESSIONS:
+ raise ValueError(
+ f"Unsupported manifest compression '{compression}'. "
+ f"Supported: {sorted(_SUPPORTED_MANIFEST_COMPRESSIONS)}")
+ if lower == 'zstd':
+ return 'zstandard'
+ return lower
diff --git a/paimon-python/pypaimon/manifest/index_manifest_file.py
b/paimon-python/pypaimon/manifest/index_manifest_file.py
index a7d82d4b12..62ca9ea16d 100644
--- a/paimon-python/pypaimon/manifest/index_manifest_file.py
+++ b/paimon-python/pypaimon/manifest/index_manifest_file.py
@@ -85,12 +85,14 @@ class IndexManifestFile:
def __init__(self, table):
from pypaimon.table.file_store_table import FileStoreTable
+ from pypaimon.manifest import avro_codec
self.table: FileStoreTable = table
manifest_path = table.table_path.rstrip('/')
self.manifest_path = f"{manifest_path}/manifest"
self.file_io = table.file_io
self.partition_keys_fields = self.table.partition_keys_fields
+ self._codec = avro_codec(table.options.manifest_compression())
def read(self, index_manifest_name: str) -> List[IndexManifestEntry]:
index_manifest_path = f"{self.manifest_path}/{index_manifest_name}"
@@ -247,7 +249,9 @@ class IndexManifestFile:
records = [self._to_avro_record(e) for e in entries]
try:
buffer = BytesIO()
- fastavro.writer(buffer, INDEX_MANIFEST_ENTRY_SCHEMA, records)
+ fastavro.writer(
+ buffer, INDEX_MANIFEST_ENTRY_SCHEMA, records,
+ codec=self._codec)
with self.file_io.new_output_stream(path) as output_stream:
output_stream.write(buffer.getvalue())
except Exception as e:
diff --git a/paimon-python/pypaimon/manifest/manifest_file_manager.py
b/paimon-python/pypaimon/manifest/manifest_file_manager.py
index 196b7bedf5..3bd6dbb226 100644
--- a/paimon-python/pypaimon/manifest/manifest_file_manager.py
+++ b/paimon-python/pypaimon/manifest/manifest_file_manager.py
@@ -49,6 +49,8 @@ class ManifestFileManager:
self.partition_keys_fields = self.table.partition_keys_fields
self.primary_keys_fields = self.table.primary_keys_fields
self.trimmed_primary_keys_fields =
self.table.trimmed_primary_keys_fields
+ from pypaimon.manifest import avro_codec
+ self._codec = avro_codec(table.options.manifest_compression())
def read_entries_parallel(self, manifest_files: List[ManifestFileMeta],
manifest_entry_filter=None,
drop_stats=True, max_workers=8,
@@ -211,7 +213,9 @@ class ManifestFileManager:
def write(self, file_name, entries: List[ManifestEntry]):
buf = BytesIO()
- fastavro.writer(buf, MANIFEST_ENTRY_SCHEMA,
self._to_avro_records(entries))
+ fastavro.writer(
+ buf, MANIFEST_ENTRY_SCHEMA, self._to_avro_records(entries),
+ codec=self._codec)
self._flush(file_name, buf.getvalue())
def rolling_write(self, entries: List[ManifestEntry],
@@ -227,7 +231,8 @@ class ManifestFileManager:
written_files = []
chunk_start = 0
buf = BytesIO()
- writer = Writer(buf, MANIFEST_ENTRY_SCHEMA,
sync_interval=sync_interval)
+ writer = Writer(buf, MANIFEST_ENTRY_SCHEMA,
+ sync_interval=sync_interval, codec=self._codec)
try:
for i, entry in enumerate(entries):
writer.write(self._to_avro_record(entry))
@@ -241,7 +246,9 @@ class ManifestFileManager:
file_name, entries[chunk_start:i + 1],
len(avro_bytes)))
chunk_start = i + 1
buf = BytesIO()
- writer = Writer(buf, MANIFEST_ENTRY_SCHEMA,
sync_interval=sync_interval)
+ writer = Writer(
+ buf, MANIFEST_ENTRY_SCHEMA,
+ sync_interval=sync_interval, codec=self._codec)
if chunk_start < len(entries):
writer.flush()
diff --git a/paimon-python/pypaimon/manifest/manifest_list_manager.py
b/paimon-python/pypaimon/manifest/manifest_list_manager.py
index 273dfbe363..3a0e606ef5 100644
--- a/paimon-python/pypaimon/manifest/manifest_list_manager.py
+++ b/paimon-python/pypaimon/manifest/manifest_list_manager.py
@@ -32,11 +32,13 @@ class ManifestListManager:
def __init__(self, table):
from pypaimon.table.file_store_table import FileStoreTable
+ from pypaimon.manifest import avro_codec
self.table: FileStoreTable = table
manifest_path = table.table_path.rstrip('/')
self.manifest_path = f"{manifest_path}/manifest"
self.file_io = self.table.file_io
+ self._codec = avro_codec(table.options.manifest_compression())
def read_all(self, snapshot: Optional[Snapshot]) -> List[ManifestFileMeta]:
"""Read base + delta manifest lists for full file state."""
@@ -124,7 +126,9 @@ class ManifestListManager:
list_path = f"{self.manifest_path}/{file_name}"
try:
buffer = BytesIO()
- fastavro.writer(buffer, MANIFEST_FILE_META_SCHEMA, avro_records)
+ fastavro.writer(
+ buffer, MANIFEST_FILE_META_SCHEMA, avro_records,
+ codec=self._codec)
avro_bytes = buffer.getvalue()
with self.file_io.new_output_stream(list_path) as output_stream:
output_stream.write(avro_bytes)
diff --git a/paimon-python/pypaimon/tests/manifest/manifest_manager_test.py
b/paimon-python/pypaimon/tests/manifest/manifest_manager_test.py
index adde81c34e..ad4754e057 100644
--- a/paimon-python/pypaimon/tests/manifest/manifest_manager_test.py
+++ b/paimon-python/pypaimon/tests/manifest/manifest_manager_test.py
@@ -346,7 +346,7 @@ class ManifestFileManagerTest(_ManifestManagerSetup):
wb = table.new_batch_write_builder()
w = wb.new_write()
- for pt in range(200):
+ for pt in range(800):
rows = [{'pt': f'p{pt}', 'pk': i, 'val': f'v{i}'} for i in
range(5)]
w.write_arrow(pa.Table.from_pylist(rows, schema=pa_schema))
wb.new_commit().commit(w.prepare_commit())
@@ -372,7 +372,7 @@ class ManifestFileManagerTest(_ManifestManagerSetup):
entries = mfm.read(meta.file_name)
self.assertEqual(len(entries), meta.num_added_files +
meta.num_deleted_files)
all_entries.extend(entries)
- self.assertEqual(len(all_entries), 200)
+ self.assertEqual(len(all_entries), 800)
def test_rolling_write_with_skewed_entries(self):
target_size = 16 * 1024
@@ -405,6 +405,49 @@ class ManifestFileManagerTest(_ManifestManagerSetup):
total_entries = sum(m.num_added_files + m.num_deleted_files for m in
metas)
self.assertEqual(total_entries, 300)
+ def test_manifest_compression(self):
+ pa_schema = pa.schema([('pk', pa.int32()), ('val', pa.string())])
+ schema_compressed = Schema.from_pyarrow_schema(
+ pa_schema, primary_keys=['pk'], options={'bucket': '1'})
+ schema_uncompressed = Schema.from_pyarrow_schema(
+ pa_schema, primary_keys=['pk'],
+ options={'bucket': '1', 'manifest.compression': 'null'})
+
+ self.catalog.create_table('default.compressed', schema_compressed,
False)
+ self.catalog.create_table('default.uncompressed', schema_uncompressed,
False)
+
+ rows = [{'pk': i, 'val': f'value-{i}' * 20} for i in range(100)]
+ arrow_data = pa.Table.from_pylist(rows, schema=pa_schema)
+
+ for name in ('default.compressed', 'default.uncompressed'):
+ table = self.catalog.get_table(name)
+ wb = table.new_batch_write_builder()
+ w = wb.new_write()
+ w.write_arrow(arrow_data)
+ wb.new_commit().commit(w.prepare_commit())
+ w.close()
+
+ t_comp = self.catalog.get_table('default.compressed')
+ t_uncomp = self.catalog.get_table('default.uncompressed')
+
+ metas_comp = ManifestListManager(t_comp).read_all(
+ t_comp.snapshot_manager().get_latest_snapshot())
+ metas_uncomp = ManifestListManager(t_uncomp).read_all(
+ t_uncomp.snapshot_manager().get_latest_snapshot())
+
+ size_comp = sum(m.file_size for m in metas_comp)
+ size_uncomp = sum(m.file_size for m in metas_uncomp)
+ self.assertLess(
+ size_comp, size_uncomp,
+ f"Compressed manifest ({size_comp}B) should be smaller "
+ f"than uncompressed ({size_uncomp}B)")
+
+ mfm = ManifestFileManager(t_comp)
+ entries = []
+ for m in metas_comp:
+ entries.extend(mfm.read(m.file_name))
+ self.assertGreater(len(entries), 0)
+
class ManifestListManagerTest(_ManifestManagerSetup):
"""Tests for ManifestListManager."""
diff --git a/paimon-python/pypaimon/tests/manifest/manifest_schema_test.py
b/paimon-python/pypaimon/tests/manifest/manifest_schema_test.py
index b806ad3fa9..f483d84ccc 100644
--- a/paimon-python/pypaimon/tests/manifest/manifest_schema_test.py
+++ b/paimon-python/pypaimon/tests/manifest/manifest_schema_test.py
@@ -221,6 +221,7 @@ class ManifestSchemaTest(unittest.TestCase):
table.table_path = table_path
table.file_io = file_io
table.partition_keys_fields = []
+ table.options.manifest_compression.return_value = 'zstd'
manager = ManifestListManager(table)
metas = manager.read(manifest_list_name)
