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 d208b698f7 [python] Skip non-overlapping manifest entries before
DataFileMeta construction (#8387)
d208b698f7 is described below
commit d208b698f794816693876ce2e81fc0a3937a4a80
Author: XiaoHongbo <[email protected]>
AuthorDate: Tue Jun 30 17:20:32 2026 +0800
[python] Skip non-overlapping manifest entries before DataFileMeta
construction (#8387)
---
.../pypaimon/manifest/manifest_file_manager.py | 23 +--
.../pypaimon/read/scanner/file_scanner.py | 48 +++++-
.../pypaimon/tests/test_early_row_range_filter.py | 185 +++++++++++++++++++++
3 files changed, 243 insertions(+), 13 deletions(-)
diff --git a/paimon-python/pypaimon/manifest/manifest_file_manager.py
b/paimon-python/pypaimon/manifest/manifest_file_manager.py
index 3bd6dbb226..3e0ae9a866 100644
--- a/paimon-python/pypaimon/manifest/manifest_file_manager.py
+++ b/paimon-python/pypaimon/manifest/manifest_file_manager.py
@@ -54,12 +54,14 @@ class ManifestFileManager:
def read_entries_parallel(self, manifest_files: List[ManifestFileMeta],
manifest_entry_filter=None,
drop_stats=True, max_workers=8,
- early_entry_filter: Optional[Callable[[int,
int], bool]] = None
+ early_entry_filter: Optional[Callable[[int,
int], bool]] = None,
+ early_record_filter: Optional[Callable[[dict],
bool]] = None
) -> List[ManifestEntry]:
def _process_single_manifest(manifest_file: ManifestFileMeta) ->
List[ManifestEntry]:
return self.read(manifest_file.file_name, manifest_entry_filter,
drop_stats,
- early_entry_filter=early_entry_filter)
+ early_entry_filter=early_entry_filter,
+ early_record_filter=early_record_filter)
def _entry_identifier(e: ManifestEntry) -> tuple:
return (
@@ -90,17 +92,14 @@ class ManifestFileManager:
return final_entries
def read(self, manifest_file_name: str, manifest_entry_filter=None,
drop_stats=True,
- early_entry_filter: Optional[Callable[[int, int], bool]] = None
+ early_entry_filter: Optional[Callable[[int, int], bool]] = None,
+ early_record_filter: Optional[Callable[[dict], bool]] = None
) -> List[ManifestEntry]:
"""
- early_entry_filter: optional ``(bucket, total_buckets) -> bool``
- called immediately after the avro record is parsed. Mirrors
- Java ``BucketFilter`` applied at the InternalRow stage in
- ``ManifestEntryCache``: when it returns False, the entry's
- ``_FILE`` block / partition / stats are never deserialized.
- Caller is responsible for soundness (any non-pruning rule must
- return True). The full ``manifest_entry_filter`` still runs on
- the survivors.
+ early_entry_filter: ``(bucket, total_buckets) -> bool``, skip before
deserializing _FILE.
+ early_record_filter: ``(fastavro record dict) -> bool``, skip before
constructing
+ DataFileMeta. Separate from early_entry_filter because it operates
on the full
+ record (can inspect _FILE sub-fields) rather than just
bucket/total_buckets.
"""
manifest_file_path = f"{self.manifest_path}/{manifest_file_name}"
@@ -120,6 +119,8 @@ class ManifestFileManager:
else:
if not early_entry_filter(bucket, total_buckets):
continue
+ if early_record_filter is not None and not
early_record_filter(record):
+ continue
file_dict = dict(record['_FILE'])
key_dict = dict(file_dict['_KEY_STATS'])
key_stats = SimpleStats(
diff --git a/paimon-python/pypaimon/read/scanner/file_scanner.py
b/paimon-python/pypaimon/read/scanner/file_scanner.py
index 3facb2f7d5..fe46abd67b 100755
--- a/paimon-python/pypaimon/read/scanner/file_scanner.py
+++ b/paimon-python/pypaimon/read/scanner/file_scanner.py
@@ -101,6 +101,42 @@ def _row_ranges_from_predicate(predicate:
Optional[Predicate]) -> Optional[List]
return visit(predicate)
+def _build_early_row_range_filter(row_ranges):
+ """Skip entries whose row-id range doesn't intersect ``row_ranges``.
+
+ Runs on the raw fastavro record (OrderedDict) before the expensive
+ Python object construction (BinaryRow, GenericRow, SimpleStats,
+ DataFileMeta). fastavro has already parsed the Avro bytes; this
+ filter avoids the construction cost, not I/O or parsing.
+
+ Safe for DELETE entries because ADD and DELETE for the same file
+ share the same ``_FIRST_ROW_ID``.
+ """
+ if row_ranges is None or not row_ranges:
+ return None
+
+ from pypaimon.utils.range import Range
+
+ def _filter(record):
+ file_dict = record.get('_FILE')
+ if file_dict is None:
+ return True
+ first_row_id = file_dict.get('_FIRST_ROW_ID')
+ if first_row_id is None:
+ return True
+ row_count = file_dict.get('_ROW_COUNT')
+ if row_count is None:
+ return True
+ file_start = int(first_row_id)
+ file_end = file_start + int(row_count) - 1
+ for r in row_ranges:
+ if Range.intersect(file_start, file_end, r.from_, r.to):
+ return True
+ return False
+
+ return _filter
+
+
def _filter_manifest_files_by_row_ranges(
manifest_files: List[ManifestFileMeta],
row_ranges: List) -> List[ManifestFileMeta]:
@@ -341,8 +377,9 @@ class FileScanner:
if row_ranges is not None:
manifest_files =
_filter_manifest_files_by_row_ranges(manifest_files, row_ranges)
- entries = self.read_manifest_entries(manifest_files)
+ entries = self.read_manifest_entries(manifest_files,
row_ranges=row_ranges)
+ # Redundant when early_record_filter ran; kept for explain mode and as
safety net.
if row_ranges is not None:
entries = _filter_manifest_entries_by_row_ranges(entries,
row_ranges)
@@ -391,7 +428,8 @@ class FileScanner:
except Exception:
return None
- def read_manifest_entries(self, manifest_files: List[ManifestFileMeta]) ->
List[ManifestEntry]:
+ def read_manifest_entries(self, manifest_files: List[ManifestFileMeta],
+ row_ranges=None) -> List[ManifestEntry]:
max_workers =
self.table.options.scan_manifest_parallelism(os.cpu_count() or 8)
if self.scan_stats is not None:
self.scan_stats.manifest_files_total += len(manifest_files)
@@ -407,11 +445,17 @@ class FileScanner:
self.scan_stats.manifest_files_after_partition +=
len(manifest_files)
# Force single-threaded so we can mutate stats without locking.
max_workers = 1
+ # Disable early_record_filter when scan_stats is active (explain mode)
+ # so that all entries flow through _filter_manifest_entry for accurate
+ # funnel counting.
+ early_row_filter = None if self.scan_stats is not None \
+ else _build_early_row_range_filter(row_ranges)
return self.manifest_file_manager.read_entries_parallel(
manifest_files,
self._filter_manifest_entry,
max_workers=max_workers,
early_entry_filter=self._build_early_bucket_filter(),
+ early_record_filter=early_row_filter,
)
def _build_early_bucket_filter(self):
diff --git a/paimon-python/pypaimon/tests/test_early_row_range_filter.py
b/paimon-python/pypaimon/tests/test_early_row_range_filter.py
new file mode 100644
index 0000000000..2253518a9f
--- /dev/null
+++ b/paimon-python/pypaimon/tests/test_early_row_range_filter.py
@@ -0,0 +1,185 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import os
+import shutil
+import tempfile
+import unittest
+from unittest.mock import patch
+
+import pyarrow as pa
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.common.predicate import Predicate
+from pypaimon.manifest.schema.data_file_meta import DataFileMeta
+
+
+class TestManifestReadRowRangePerformance(unittest.TestCase):
+ """Reproduce: scan planning constructs ALL manifest entries even when
+ row_ranges only match a few files."""
+
+ NUM_COMMITS = 20
+ ROWS_PER_COMMIT = 10
+
+ @classmethod
+ def setUpClass(cls):
+ cls.tempdir = tempfile.mkdtemp()
+ cls.warehouse = os.path.join(cls.tempdir, 'warehouse')
+ cls.catalog = CatalogFactory.create({'warehouse': cls.warehouse})
+ cls.catalog.create_database('default', True)
+
+ pa_schema = pa.schema([
+ ('id', pa.int32()),
+ ('value', pa.string()),
+ ])
+ schema = Schema.from_pyarrow_schema(pa_schema, options={
+ 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true',
+ 'manifest.merge-min-count': '1',
+ })
+ cls.catalog.create_table('default.test_row_range_perf', schema, False)
+ cls.table = cls.catalog.get_table('default.test_row_range_perf')
+ cls.pa_schema = pa_schema
+
+ write_builder = cls.table.new_batch_write_builder()
+ for i in range(cls.NUM_COMMITS):
+ tw = write_builder.new_write()
+ tc = write_builder.new_commit()
+ start_id = i * cls.ROWS_PER_COMMIT
+ data = pa.Table.from_pydict({
+ 'id': list(range(start_id, start_id + cls.ROWS_PER_COMMIT)),
+ 'value': [f'v{j}' for j in range(cls.ROWS_PER_COMMIT)],
+ }, schema=pa_schema)
+ tw.write_arrow(data)
+ cmts = tw.prepare_commit()
+ for msg in cmts:
+ for nf in msg.new_files:
+ nf.first_row_id = start_id
+ tc.commit(cmts)
+ tw.close()
+ tc.close()
+
+ @classmethod
+ def tearDownClass(cls):
+ shutil.rmtree(cls.tempdir, ignore_errors=True)
+
+ def test_scan_constructs_all_entries_without_early_row_range_filter(self):
+ """With manifest.merge-min-count=1, all entries are in one manifest.
+ Querying with _ROW_ID BETWEEN 5 AND 14 should return 2 files, but
+ the current code constructs DataFileMeta for ALL 20 entries because
+ the row-range filter runs after full object construction."""
+ construction_count = [0]
+ original_init = DataFileMeta.__init__
+
+ def counting_init(self_meta, *args, **kwargs):
+ construction_count[0] += 1
+ original_init(self_meta, *args, **kwargs)
+
+ predicate = Predicate(method='between', index=None,
+ field='_ROW_ID', literals=[5, 14])
+
+ with patch.object(DataFileMeta, '__init__', counting_init):
+ rb = self.table.new_read_builder().with_filter(predicate)
+ splits = rb.new_scan().plan().splits()
+
+ total_files = sum(len(s.files) for s in splits)
+ self.assertEqual(total_files, 2)
+
+ actual = self.table.new_read_builder().with_filter(predicate) \
+ .new_read().to_arrow(splits)
+ self.assertEqual(sorted(actual.column('id').to_pylist()),
+ list(range(5, 15)))
+
+ # 2 matching files × 2 (ADD + DELETE from manifest merge) = 4
+ self.assertLessEqual(
+ construction_count[0], 2 * total_files,
+ f"Expected at most {2 * total_files} DataFileMeta constructions, "
+ f"got {construction_count[0]}")
+
+ def test_add_delete_pair_not_broken_by_early_filter(self):
+ """Guard ADD/DELETE merge semantics: when a file is overwritten,
+ its ADD and DELETE entries share the same _FIRST_ROW_ID. The early
+ filter must either keep both or drop both — never keep ADD but
+ drop DELETE (which would resurrect a deleted file)."""
+ pa_schema = self.pa_schema
+ self.catalog.create_table('default.test_add_delete_pair',
+ Schema.from_pyarrow_schema(pa_schema,
options={
+ 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true',
+ 'manifest.merge-min-count': '1',
+ }), False)
+ table = self.catalog.get_table('default.test_add_delete_pair')
+ wb = table.new_batch_write_builder()
+
+ # Commit 1: write file at row_id 0-9
+ tw, tc = wb.new_write(), wb.new_commit()
+ tw.write_arrow(pa.Table.from_pydict(
+ {'id': list(range(10)), 'value': ['old'] * 10}, schema=pa_schema))
+ cmts = tw.prepare_commit()
+ for m in cmts:
+ for nf in m.new_files:
+ nf.first_row_id = 0
+ tc.commit(cmts)
+ tw.close()
+ tc.close()
+
+ # Commit 2: overwrite same row_id 0-9 with new data
+ # This creates DELETE(old_file, row_id=0) + ADD(new_file, row_id=0)
+ tw = wb.new_write().with_write_type(['id', 'value'])
+ tc = wb.new_commit()
+ tw.write_arrow(pa.Table.from_pydict(
+ {'id': list(range(10)), 'value': ['new'] * 10}, schema=pa_schema))
+ cmts = tw.prepare_commit()
+ for m in cmts:
+ for nf in m.new_files:
+ nf.first_row_id = 0
+ tc.commit(cmts)
+ tw.close()
+ tc.close()
+
+ # Commit 3: write file at row_id 100-109 (unrelated, outside query
range)
+ tw, tc = wb.new_write(), wb.new_commit()
+ tw.write_arrow(pa.Table.from_pydict(
+ {'id': list(range(100, 110)), 'value': ['other'] * 10},
schema=pa_schema))
+ cmts = tw.prepare_commit()
+ for m in cmts:
+ for nf in m.new_files:
+ nf.first_row_id = 100
+ tc.commit(cmts)
+ tw.close()
+ tc.close()
+
+ # Query row_id [0, 9]: should return the NEW data, not the old
+ predicate = Predicate(method='between', index=None,
+ field='_ROW_ID', literals=[0, 9])
+ rb = table.new_read_builder().with_filter(predicate)
+ splits = rb.new_scan().plan().splits()
+ actual = rb.new_read().to_arrow(splits)
+
+ values = actual.column('value').to_pylist()
+ self.assertTrue(all(v == 'new' for v in values),
+ f"Expected all 'new' but got {values}. "
+ "Early filter may have dropped DELETE without its
ADD.")
+
+ # Also verify row_id [100, 109] is NOT in the result
+ ids = actual.column('id').to_pylist()
+ self.assertTrue(all(i < 100 for i in ids),
+ f"row_id [100,109] should be filtered out but got
ids={ids}")
+
+
+if __name__ == '__main__':
+ unittest.main()
