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 6632646cfd [python] Avoid redundant manifest size lookups (#10310)
6632646cfd is described below
commit 6632646cfdd565691c41392e125daccdbecbfd84
Author: XiaoHongbo <[email protected]>
AuthorDate: Mon Oct 5 21:20:53 2026 +0800
[python] Avoid redundant manifest size lookups (#10310)
---
.../pypaimon/filesystem/caching_file_io.py | 15 ++++-
.../pypaimon/manifest/manifest_file_manager.py | 15 ++++-
.../pypaimon/manifest/manifest_list_manager.py | 27 +++++---
.../pypaimon/manifest/manifest_sidecar.py | 5 +-
.../tests/manifest/manifest_manager_test.py | 77 ++++++++++++++++++++++
.../tests/manifest/manifest_sidecar_test.py | 15 ++++-
.../pypaimon/tests/reader_append_only_test.py | 4 +-
.../pypaimon/write/commit/commit_scanner.py | 2 +-
.../write/commit/overwrite_changes_provider.py | 3 +-
paimon-python/pypaimon/write/file_store_commit.py | 16 +++--
10 files changed, 150 insertions(+), 29 deletions(-)
diff --git a/paimon-python/pypaimon/filesystem/caching_file_io.py
b/paimon-python/pypaimon/filesystem/caching_file_io.py
index 3fad543d8d..1ddbb76ee0 100644
--- a/paimon-python/pypaimon/filesystem/caching_file_io.py
+++ b/paimon-python/pypaimon/filesystem/caching_file_io.py
@@ -195,11 +195,11 @@ class LocalDiskCacheManager:
class CachingInputStream:
"""Wraps a remote stream with block-level caching."""
- def __init__(self, file_io, file_path: str, cache):
+ def __init__(self, file_io, file_path: str, cache, file_size=None):
self._file_io = file_io
self._stream = None
self._file_path = file_path
- self._file_size = -1
+ self._file_size = file_size if file_size is not None and file_size > 0
else -1
self._cache = cache
self._pos = 0
self._io_lock = threading.Lock()
@@ -397,10 +397,13 @@ class CachingFileIO(FileIO):
return self._delegate.properties
def new_input_stream(self, path: str):
+ return self._new_input_stream_with_known_size(path, None)
+
+ def _new_input_stream_with_known_size(self, path: str, file_size):
file_type = FileType.classify(path)
if self._cache is None or file_type not in self._whitelist or
FileType.is_mutable(path):
return self._delegate.new_input_stream(path)
- return CachingInputStream(self._delegate, path, self._cache)
+ return CachingInputStream(self._delegate, path, self._cache, file_size)
def new_output_stream(self, path: str):
return self._delegate.new_output_stream(path)
@@ -471,3 +474,9 @@ class CachingFileIO(FileIO):
def close(self):
self._delegate.close()
+
+
+def _open_input_stream_with_known_size(file_io, path, file_size):
+ if isinstance(file_io, CachingFileIO) and file_size is not None and
file_size > 0:
+ return file_io._new_input_stream_with_known_size(path, file_size)
+ return file_io.new_input_stream(path)
diff --git a/paimon-python/pypaimon/manifest/manifest_file_manager.py
b/paimon-python/pypaimon/manifest/manifest_file_manager.py
index 08fac1bf88..27ae7b32d8 100644
--- a/paimon-python/pypaimon/manifest/manifest_file_manager.py
+++ b/paimon-python/pypaimon/manifest/manifest_file_manager.py
@@ -31,6 +31,8 @@ except ImportError: # pragma: no cover - supported fastavro
versions provide th
from datetime import datetime
+from pypaimon.filesystem.caching_file_io import (
+ CachingInputStream, _open_input_stream_with_known_size)
from pypaimon.manifest.manifest_sidecar import (
Query, read_sidecar, read_selected_bytes,
)
@@ -152,7 +154,8 @@ class ManifestFileManager:
early_entry_filter=early_entry_filter,
early_record_filter=early_record_filter,
partition_filter=partition_filter,
- selected_blocks=selected)
+ selected_blocks=selected,
+ file_size=manifest_file.file_size)
def _entry_identifier(e: ManifestEntry) -> tuple:
return (
@@ -196,6 +199,7 @@ class ManifestFileManager:
early_record_filter: Optional[Callable[[dict], bool]] = None,
partition_filter=None,
selected_blocks=None,
+ file_size: Optional[int] = None,
) -> List[ManifestEntry]:
"""
early_entry_filter: ``(bucket, total_buckets) -> bool``, skip before
deserializing _FILE.
@@ -210,10 +214,15 @@ class ManifestFileManager:
entries = []
if selected_blocks is not None:
- avro_bytes = read_selected_bytes(self.file_io, manifest_file_path,
selected_blocks)
+ avro_bytes = read_selected_bytes(
+ self.file_io, manifest_file_path, selected_blocks, file_size)
else:
- with self.file_io.new_input_stream(manifest_file_path) as
input_stream:
+ with _open_input_stream_with_known_size(
+ self.file_io, manifest_file_path, file_size) as
input_stream:
avro_bytes = input_stream.read()
+ if (isinstance(input_stream, CachingInputStream) and file_size
is not None
+ and file_size > 0 and len(avro_bytes) != file_size):
+ raise EOFError('Truncated manifest file
{}'.format(manifest_file_path))
buffer = BytesIO(avro_bytes)
records = _read_manifest_records(
buffer, early_entry_filter, partition_filter,
diff --git a/paimon-python/pypaimon/manifest/manifest_list_manager.py
b/paimon-python/pypaimon/manifest/manifest_list_manager.py
index 70ec8f2ae6..7a8e269d00 100644
--- a/paimon-python/pypaimon/manifest/manifest_list_manager.py
+++ b/paimon-python/pypaimon/manifest/manifest_list_manager.py
@@ -19,6 +19,9 @@ from io import BytesIO
from typing import List, Optional
import fastavro
+
+from pypaimon.filesystem.caching_file_io import (
+ CachingInputStream, _open_input_stream_with_known_size)
from pypaimon.manifest.schema.manifest_file_meta import (
MANIFEST_FILE_META_SCHEMA, ManifestFileMeta)
from pypaimon.manifest.schema.simple_stats import SimpleStats
@@ -45,35 +48,40 @@ class ManifestListManager:
if snapshot is None:
return []
manifest_files = []
- base_manifests = self.read(snapshot.base_manifest_list)
+ base_manifests = self.read(snapshot.base_manifest_list,
snapshot.base_manifest_list_size)
manifest_files.extend(base_manifests)
- delta_manifests = self.read(snapshot.delta_manifest_list)
+ delta_manifests = self.read(snapshot.delta_manifest_list,
snapshot.delta_manifest_list_size)
manifest_files.extend(delta_manifests)
return manifest_files
def read_base(self, snapshot: Snapshot) -> List[ManifestFileMeta]:
"""Read only the base manifest list for the given snapshot."""
- return self.read(snapshot.base_manifest_list)
+ return self.read(snapshot.base_manifest_list,
snapshot.base_manifest_list_size)
def read_delta(self, snapshot: Snapshot) -> List[ManifestFileMeta]:
- return self.read(snapshot.delta_manifest_list)
+ return self.read(snapshot.delta_manifest_list,
snapshot.delta_manifest_list_size)
def read_changelog(self, snapshot: Snapshot) -> List[ManifestFileMeta]:
"""Read changelog manifest files from snapshot, or empty list if
none."""
if snapshot.changelog_manifest_list is None:
return []
- return self.read(snapshot.changelog_manifest_list)
+ return self.read(snapshot.changelog_manifest_list,
snapshot.changelog_manifest_list_size)
- def read(self, manifest_list_name: str) -> List[ManifestFileMeta]:
- return self._read_from_storage(manifest_list_name)
+ def read(self, manifest_list_name: str, file_size: Optional[int] = None)
-> List[ManifestFileMeta]:
+ return self._read_from_storage(manifest_list_name, file_size)
- def _read_from_storage(self, manifest_list_name: str) ->
List[ManifestFileMeta]:
+ def _read_from_storage(self, manifest_list_name: str,
+ file_size: Optional[int] = None) ->
List[ManifestFileMeta]:
"""Read manifest list from storage."""
manifest_files = []
manifest_list_path = f"{self.manifest_path}/{manifest_list_name}"
- with self.file_io.new_input_stream(manifest_list_path) as input_stream:
+ with _open_input_stream_with_known_size(
+ self.file_io, manifest_list_path, file_size) as input_stream:
avro_bytes = input_stream.read()
+ if (isinstance(input_stream, CachingInputStream) and file_size is
not None
+ and file_size > 0 and len(avro_bytes) != file_size):
+ raise EOFError('Truncated manifest list
{}'.format(manifest_list_path))
buffer = BytesIO(avro_bytes)
reader = fastavro.reader(buffer)
for record in reader:
@@ -147,3 +155,4 @@ class ManifestListManager:
except Exception as e:
self.file_io.delete_quietly(list_path)
raise RuntimeError(f"Failed to write manifest list file: {e}")
from e
+ return len(avro_bytes)
diff --git a/paimon-python/pypaimon/manifest/manifest_sidecar.py
b/paimon-python/pypaimon/manifest/manifest_sidecar.py
index 8c1397333c..e6c1704430 100644
--- a/paimon-python/pypaimon/manifest/manifest_sidecar.py
+++ b/paimon-python/pypaimon/manifest/manifest_sidecar.py
@@ -28,6 +28,7 @@ from typing import Tuple
from pyarrow import ArrowCancelled
+from pypaimon.filesystem.caching_file_io import
_open_input_stream_with_known_size
from pypaimon.utils.range import Range
from pypaimon.table.row.generic_row import GenericRowSerializer,
GenericRowDeserializer
@@ -505,14 +506,14 @@ def read_sidecar(file_io, manifest_path, manifest, query,
partition_filter=None,
return None
-def read_selected_bytes(file_io, manifest_path, selected):
+def read_selected_bytes(file_io, manifest_path, selected, file_size=None):
"""Read complete selected blocks with seek; adjacent blocks share one
contiguous span.
The concatenated original header and blocks form a valid Avro OCF. Partial
entries
must not be stored in a cache keyed by the complete manifest.
"""
data = bytearray(selected.header)
- with file_io.new_input_stream(manifest_path) as stream:
+ with _open_input_stream_with_known_size(file_io, manifest_path, file_size)
as stream:
block_position = 0
while block_position < len(selected.blocks):
block = selected.blocks[block_position]
diff --git a/paimon-python/pypaimon/tests/manifest/manifest_manager_test.py
b/paimon-python/pypaimon/tests/manifest/manifest_manager_test.py
index 7389ea02da..043e405907 100644
--- a/paimon-python/pypaimon/tests/manifest/manifest_manager_test.py
+++ b/paimon-python/pypaimon/tests/manifest/manifest_manager_test.py
@@ -45,6 +45,7 @@ from pypaimon.manifest.schema.simple_stats import SimpleStats
from pypaimon.schema.data_types import AtomicType, DataField
from pypaimon.schema.schema import Schema
from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.write.commit.commit_scanner import CommitScanner
_EMPTY_ROW = GenericRow([], [])
_EMPTY_STATS = SimpleStats(min_values=_EMPTY_ROW, max_values=_EMPTY_ROW,
null_counts=[])
@@ -725,6 +726,82 @@ class ManifestListManagerTest(_ManifestManagerSetup):
self.assertEqual(len(result), 1)
self.assertEqual(result[0].file_name, "manifest-base.avro")
+ def test_scan_uses_manifest_sizes_without_file_status(self):
+ for disk in (False, True):
+ with self.subTest(disk=disk), tempfile.TemporaryDirectory() as
warehouse:
+ options = {'warehouse': warehouse, 'local-cache.enabled':
'true',
+ 'local-cache.block-size': '64 b'}
+ if disk:
+ options['local-cache.dir'] = os.path.join(warehouse,
'cache')
+
+ catalog = FileSystemCatalog(Options(options))
+ catalog.create_database('default', False)
+ schema = Schema.from_pyarrow_schema(pa.schema([('v',
pa.int32())]))
+ catalog.create_table('default.t', schema, False)
+ table = catalog.get_table('default.t')
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(pa.table({'v': pa.array([1, 2, 3],
type=pa.int32())}))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ for name, size in ((snapshot.base_manifest_list,
+ snapshot.base_manifest_list_size),
+ (snapshot.delta_manifest_list,
+ snapshot.delta_manifest_list_size)):
+ self.assertGreater(size, 0)
+ self.assertEqual(size, table.file_io.get_file_size(
+ '{}/manifest/{}'.format(table.table_path, name)))
+
+ # A fresh catalog has no in-process file-size cache.
+ fresh_catalog = FileSystemCatalog(Options(options))
+ fresh_table = fresh_catalog.get_table('default.t')
+ delegate = fresh_table.file_io._delegate
+ with mock.patch.object(delegate, 'get_file_size',
+ wraps=delegate.get_file_size) as
get_size:
+ splits =
fresh_table.new_read_builder().new_scan().plan().splits()
+ self.assertTrue(splits)
+ self.assertEqual(
+
fresh_table.new_read_builder().new_read().to_arrow(splits).column(
+ 'v').to_pylist(), [1, 2, 3])
+ self.assertFalse([call for call in get_size.call_args_list
+ if '/manifest/' in call[0][0]])
+
+ # Conflict retries read raw delta manifests through
CommitScanner.
+ retry_catalog = FileSystemCatalog(Options(options))
+ retry_table = retry_catalog.get_table('default.t')
+ retry_delegate = retry_table.file_io._delegate
+ scanner = CommitScanner(retry_table,
ManifestListManager(retry_table))
+ with mock.patch.object(retry_delegate, 'get_file_size',
+ wraps=retry_delegate.get_file_size) as
get_size:
+ entries =
scanner.read_incremental_raw_entries_from_changed_partitions(
+ snapshot, [])
+ self.assertTrue(entries)
+ self.assertFalse([call for call in get_size.call_args_list
+ if '/manifest/' in call[0][0]])
+
+ # Old snapshots without list sizes retain the status lookup.
+ legacy = replace(snapshot, base_manifest_list_size=None,
+ delta_manifest_list_size=None)
+ legacy_catalog = FileSystemCatalog(Options(options))
+ legacy_table = legacy_catalog.get_table('default.t')
+ delegate = legacy_table.file_io._delegate
+ with mock.patch.object(delegate, 'get_file_size',
+ wraps=delegate.get_file_size) as
get_size:
+
self.assertTrue(ManifestListManager(legacy_table).read_all(legacy))
+ self.assertEqual(len([call for call in get_size.call_args_list
+ if '/manifest/' in call[0][0]]), 2)
+
+ wrong_size_catalog = FileSystemCatalog(Options(options))
+ wrong_size_table = wrong_size_catalog.get_table('default.t')
+ with self.assertRaisesRegex(EOFError, 'Truncated manifest
list'):
+ ManifestListManager(wrong_size_table).read(
+ snapshot.base_manifest_list,
snapshot.base_manifest_list_size + 1)
+
@unittest.skipIf(
sys.version_info < (3, 9),
'PyPI backports.zstd only supports Python 3.9–3.13',
diff --git a/paimon-python/pypaimon/tests/manifest/manifest_sidecar_test.py
b/paimon-python/pypaimon/tests/manifest/manifest_sidecar_test.py
index 405c4c92b1..868b1518cd 100644
--- a/paimon-python/pypaimon/tests/manifest/manifest_sidecar_test.py
+++ b/paimon-python/pypaimon/tests/manifest/manifest_sidecar_test.py
@@ -32,7 +32,7 @@ from unittest.mock import Mock, call, patch
from pypaimon.common.options.core_options import CoreOptions
from pypaimon.common.options.options import Options
-from pypaimon.filesystem.caching_file_io import CachingFileIO
+from pypaimon.filesystem.caching_file_io import CachingFileIO,
LocalMemoryCacheManager
from pypaimon.globalindex.global_index_result import GlobalIndexResult
from pypaimon.manifest import manifest_sidecar
from pypaimon.manifest.manifest_sidecar import (
@@ -131,6 +131,19 @@ class FailingIndexInput(BytesIO):
class ManifestSidecarReadTest(unittest.TestCase):
+ def test_selected_manifest_blocks_use_known_size(self):
+ data = b'headfoo'
+ delegate = SimpleNamespace(
+ new_input_stream=Mock(side_effect=lambda path: BytesIO(data)),
+ get_file_size=Mock(side_effect=AssertionError('source stat')))
+ file_io = CachingFileIO(delegate, LocalMemoryCacheManager(1024,
block_size=4))
+ selection = Selection(b'head', (Block(4, 3, 0, 1),))
+
+ self.assertEqual(
+ read_selected_bytes(file_io, '/manifest/manifest-1', selection,
len(data)),
+ data)
+ delegate.get_file_size.assert_not_called()
+
def test_local_cache_shares_sidecar_bytes_across_queries_and_readers(self):
data, meta = golden(), golden_meta()
meta.extra_files = ['custom' + SUFFIX]
diff --git a/paimon-python/pypaimon/tests/reader_append_only_test.py
b/paimon-python/pypaimon/tests/reader_append_only_test.py
index 032b7fdc94..dccc884b63 100644
--- a/paimon-python/pypaimon/tests/reader_append_only_test.py
+++ b/paimon-python/pypaimon/tests/reader_append_only_test.py
@@ -1125,7 +1125,7 @@ class AoReaderTest(unittest.TestCase):
manifest_entry_filter=None,
drop_stats=True, early_entry_filter=None,
early_record_filter=None, partition_filter=None,
- selected_blocks=None):
+ selected_blocks=None, file_size=None):
# avro_total = every entry in the manifest (no manifest-file
pruning
# here: single file, is_in spans its partition stats).
path = f"{self_mgr.manifest_path}/{manifest_file_name}"
@@ -1136,7 +1136,7 @@ class AoReaderTest(unittest.TestCase):
self_mgr, manifest_file_name,
manifest_entry_filter, drop_stats,
early_entry_filter, early_record_filter, partition_filter,
- selected_blocks=selected_blocks)
+ selected_blocks=selected_blocks, file_size=file_size)
def counting_dfm_init(self_dfm, *args, **kwargs):
entry_counts['constructed'] += 1
diff --git a/paimon-python/pypaimon/write/commit/commit_scanner.py
b/paimon-python/pypaimon/write/commit/commit_scanner.py
index 77a7c9c74b..b08bafcb81 100644
--- a/paimon-python/pypaimon/write/commit/commit_scanner.py
+++ b/paimon-python/pypaimon/write/commit/commit_scanner.py
@@ -121,7 +121,7 @@ class CommitScanner:
mfm = ManifestFileManager(self.table)
entries = []
for mf in delta_manifests:
- for entry in mfm.read(mf.file_name):
+ for entry in mfm.read(mf.file_name, file_size=mf.file_size):
if partition_filter is not None and not
partition_filter.test(entry.partition):
continue
entries.append(entry)
diff --git a/paimon-python/pypaimon/write/commit/overwrite_changes_provider.py
b/paimon-python/pypaimon/write/commit/overwrite_changes_provider.py
index 94241aaed7..31e969ab7b 100644
--- a/paimon-python/pypaimon/write/commit/overwrite_changes_provider.py
+++ b/paimon-python/pypaimon/write/commit/overwrite_changes_provider.py
@@ -111,7 +111,8 @@ class OverwriteChangesProvider:
# snapshots are applied instead of being discarded by FileScanner.
entries = []
for manifest_file in delta_manifests:
- for entry in manifest_file_manager.read(manifest_file.file_name):
+ for entry in manifest_file_manager.read(
+ manifest_file.file_name,
file_size=manifest_file.file_size):
if (self.partition_filter is not None
and not self.partition_filter.test(entry.partition)):
continue
diff --git a/paimon-python/pypaimon/write/file_store_commit.py
b/paimon-python/pypaimon/write/file_store_commit.py
index e46346e5eb..5259be2461 100644
--- a/paimon-python/pypaimon/write/file_store_commit.py
+++ b/paimon-python/pypaimon/write/file_store_commit.py
@@ -682,7 +682,8 @@ class FileStoreCommit:
changelog_record_count = None
try:
new_manifest_file_metas =
self._write_manifest_files(commit_entries, new_manifest_file)
- self.manifest_list_manager.write(delta_manifest_list,
new_manifest_file_metas)
+ delta_manifest_list_size = self.manifest_list_manager.write(
+ delta_manifest_list, new_manifest_file_metas)
# Write changelog manifest if changelog entries exist
if changelog_entries:
@@ -690,11 +691,8 @@ class FileStoreCommit:
changelog_manifest_file_metas = self._write_manifest_files(
changelog_entries, changelog_manifest_file)
changelog_manifest_list_name =
f"manifest-list-{unique_id}-changelog"
- self.manifest_list_manager.write(
+ changelog_manifest_list_size =
self.manifest_list_manager.write(
changelog_manifest_list_name,
changelog_manifest_file_metas)
- manifest_path = self.manifest_list_manager.manifest_path
- changelog_manifest_list_size =
self.table.file_io.get_file_size(
- f"{manifest_path}/{changelog_manifest_list_name}")
# kind==0 means ADD; pypaimon producers only support additions
currently
changelog_record_count = sum(
entry.file.row_count for entry in changelog_entries if
entry.kind == 0)
@@ -709,7 +707,8 @@ class FileStoreCommit:
if previous_record_count:
total_record_count += previous_record_count
- self.manifest_list_manager.write(base_manifest_list,
existing_manifests)
+ base_manifest_list_size = self.manifest_list_manager.write(
+ base_manifest_list, existing_manifests)
delta_record_count = 0
for entry in commit_entries:
@@ -733,7 +732,9 @@ class FileStoreCommit:
id=new_snapshot_id,
schema_id=self.table.table_schema.id,
base_manifest_list=base_manifest_list,
+ base_manifest_list_size=base_manifest_list_size,
delta_manifest_list=delta_manifest_list,
+ delta_manifest_list_size=delta_manifest_list_size,
changelog_manifest_list=changelog_manifest_list_name,
changelog_manifest_list_size=changelog_manifest_list_size,
changelog_record_count=changelog_record_count,
@@ -900,7 +901,8 @@ class FileStoreCommit:
for manifest in self.manifest_list_manager.read_delta(
snapshot):
entries.extend(self.manifest_file_manager.read(
- manifest.file_name, drop_stats=False))
+ manifest.file_name, drop_stats=False,
+ file_size=manifest.file_size))
path_factory = self.table.path_factory()
for entry in entries:
file = entry.file
