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 878a4b01b0 [python] Early partition filter in manifest reading (#8429)
878a4b01b0 is described below
commit 878a4b01b030b3801af18d3a8c024bfd6a7d6d6c
Author: XiaoHongbo <[email protected]>
AuthorDate: Fri Jul 3 13:09:44 2026 +0800
[python] Early partition filter in manifest reading (#8429)
---
.../pypaimon/manifest/manifest_file_manager.py | 23 +++-
.../pypaimon/read/scanner/file_scanner.py | 8 +-
.../pypaimon/tests/reader_append_only_test.py | 145 +++++++++++++++++++++
3 files changed, 169 insertions(+), 7 deletions(-)
diff --git a/paimon-python/pypaimon/manifest/manifest_file_manager.py
b/paimon-python/pypaimon/manifest/manifest_file_manager.py
index 3e0ae9a866..02c9e7fa86 100644
--- a/paimon-python/pypaimon/manifest/manifest_file_manager.py
+++ b/paimon-python/pypaimon/manifest/manifest_file_manager.py
@@ -55,13 +55,15 @@ 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_record_filter: Optional[Callable[[dict],
bool]] = None
+ early_record_filter: Optional[Callable[[dict],
bool]] = None,
+ partition_filter=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_record_filter=early_record_filter)
+ early_record_filter=early_record_filter,
+ partition_filter=partition_filter)
def _entry_identifier(e: ManifestEntry) -> tuple:
return (
@@ -93,13 +95,17 @@ class ManifestFileManager:
def read(self, manifest_file_name: str, manifest_entry_filter=None,
drop_stats=True,
early_entry_filter: Optional[Callable[[int, int], bool]] = None,
- early_record_filter: Optional[Callable[[dict], bool]] = None
+ early_record_filter: Optional[Callable[[dict], bool]] = None,
+ partition_filter=None,
) -> List[ManifestEntry]:
"""
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.
+ partition_filter: optional Predicate tested against the entry partition
+ (deserialized before the _FILE block); non-matching entries skip
full
+ _FILE deserialization.
"""
manifest_file_path = f"{self.manifest_path}/{manifest_file_name}"
@@ -121,6 +127,12 @@ class ManifestFileManager:
continue
if early_record_filter is not None and not
early_record_filter(record):
continue
+ partition = None
+ if partition_filter is not None:
+ partition = GenericRowDeserializer.from_bytes(
+ record['_PARTITION'], self.partition_keys_fields)
+ if not partition_filter.test(partition):
+ continue
file_dict = dict(record['_FILE'])
key_dict = dict(file_dict['_KEY_STATS'])
key_stats = SimpleStats(
@@ -179,9 +191,12 @@ class ManifestFileManager:
first_row_id=file_dict['_FIRST_ROW_ID'] if '_FIRST_ROW_ID' in
file_dict else None,
write_cols=file_dict['_WRITE_COLS'] if '_WRITE_COLS' in
file_dict else None,
)
+ if partition is None:
+ partition = GenericRowDeserializer.from_bytes(
+ record['_PARTITION'], self.partition_keys_fields)
entry = ManifestEntry(
kind=record['_KIND'],
-
partition=GenericRowDeserializer.from_bytes(record['_PARTITION'],
self.partition_keys_fields),
+ partition=partition,
bucket=record['_BUCKET'],
total_buckets=record['_TOTAL_BUCKETS'],
file=file_meta
diff --git a/paimon-python/pypaimon/read/scanner/file_scanner.py
b/paimon-python/pypaimon/read/scanner/file_scanner.py
index 46eb340b43..57fe7e27b2 100755
--- a/paimon-python/pypaimon/read/scanner/file_scanner.py
+++ b/paimon-python/pypaimon/read/scanner/file_scanner.py
@@ -449,17 +449,19 @@ 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.
+ # Disable both early filters in explain mode (scan_stats) so 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)
+ partition_filter = None if self.scan_stats is not None \
+ else self.partition_key_predicate
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,
+ partition_filter=partition_filter,
)
def _build_early_bucket_filter(self):
diff --git a/paimon-python/pypaimon/tests/reader_append_only_test.py
b/paimon-python/pypaimon/tests/reader_append_only_test.py
index 7e0c5a6748..67bbfbedbd 100644
--- a/paimon-python/pypaimon/tests/reader_append_only_test.py
+++ b/paimon-python/pypaimon/tests/reader_append_only_test.py
@@ -1010,6 +1010,151 @@ class AoReaderTest(unittest.TestCase):
print(f"✓ Iteration {test_iteration + 1}/{iter_num} completed
successfully")
+ def test_is_in_with_partitions(self):
+ from pypaimon.manifest.manifest_file_manager import ManifestFileManager
+ from io import BytesIO
+ import fastavro
+
+ num_partitions = 5000
+ in_size = 2000
+
+ schema = Schema.from_pyarrow_schema(
+ pa.schema([('pk', pa.int32()), ('val', pa.string()),
+ ('pt', pa.string())]),
+ partition_keys=['pt'])
+ self.catalog.create_table(
+ 'default.test_is_in_partition_bench', schema, False)
+ table = self.catalog.get_table(
+ 'default.test_is_in_partition_bench')
+
+ wb = table.new_batch_write_builder()
+ w = wb.new_write()
+ bench_schema = pa.schema(
+ [('pk', pa.int32()), ('val', pa.string()),
+ ('pt', pa.string())])
+ for i in range(num_partitions):
+ w.write_arrow(pa.Table.from_pydict(
+ {'pk': [i], 'val': [f'v_{i}'], 'pt': [f'pt_{i}']},
+ schema=bench_schema))
+ wb.new_commit().commit(w.prepare_commit())
+ w.close()
+
+ builder = table.new_read_builder().new_predicate_builder()
+ in_values = [f'pt_{i}' for i in range(in_size)]
+ pred = builder.is_in('pt', in_values)
+
+ splits = table.new_read_builder().with_filter(
+ pred).new_scan().plan().splits()
+ self.assertEqual(len(splits), in_size)
+
+ from pypaimon.manifest.schema.data_file_meta import DataFileMeta
+
+ entry_counts = {'avro_total': 0, 'constructed': 0}
+ original_read = ManifestFileManager.read
+ original_dfm_init = DataFileMeta.__init__
+
+ def counting_read(self_mgr, manifest_file_name,
+ manifest_entry_filter=None,
+ drop_stats=True, early_entry_filter=None,
+ early_record_filter=None, partition_filter=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}"
+ with self_mgr.file_io.new_input_stream(path) as s:
+ for _ in fastavro.reader(BytesIO(s.read())):
+ entry_counts['avro_total'] += 1
+ return original_read(
+ self_mgr, manifest_file_name,
+ manifest_entry_filter, drop_stats,
+ early_entry_filter, early_record_filter, partition_filter)
+
+ def counting_dfm_init(self_dfm, *args, **kwargs):
+ entry_counts['constructed'] += 1
+ original_dfm_init(self_dfm, *args, **kwargs)
+
+ ManifestFileManager.read = counting_read
+ DataFileMeta.__init__ = counting_dfm_init
+ try:
+ table.new_read_builder().with_filter(
+ pred).new_scan().plan()
+ finally:
+ ManifestFileManager.read = original_read
+ DataFileMeta.__init__ = original_dfm_init
+
+ # all entries scanned, but only matching ones build a DataFileMeta
+ # (_FILE + drop_stats copy = 2 each); non-matching skipped
pre-construct.
+ self.assertEqual(entry_counts['avro_total'], num_partitions)
+ self.assertEqual(entry_counts['constructed'], in_size * 2)
+
+ def test_partition_filter_test_handles_isnull(self):
+ # The early filter keeps/drops an entry via Predicate.test on its exact
+ # partition value: isNull keeps null, drops non-null; equal is the
reverse.
+ from pypaimon.table.row.generic_row import GenericRow
+ from pypaimon.table.row.row_kind import RowKind
+
+ schema = Schema.from_pyarrow_schema(
+ pa.schema([('pk', pa.int32()), ('pt', pa.string())]),
+ partition_keys=['pt'])
+ self.catalog.create_table('default.test_isnull_pt_pred', schema, False)
+ table = self.catalog.get_table('default.test_isnull_pt_pred')
+ fields = table.table_schema.fields # predicate index is table-schema
space
+ row_null = GenericRow([1, None], fields, RowKind.INSERT)
+ row_x = GenericRow([2, 'x'], fields, RowKind.INSERT)
+
+ pb = table.new_read_builder().new_predicate_builder()
+ self.assertTrue(pb.is_null('pt').test(row_null))
+ self.assertFalse(pb.is_null('pt').test(row_x))
+ self.assertFalse(pb.equal('pt', 'x').test(row_null))
+ self.assertTrue(pb.equal('pt', 'x').test(row_x))
+
+ def test_early_partition_filter_isnull_scan(self):
+ # Black-box: drive the early manifest partition filter through an
actual
+ # scan (not just Predicate.test), over a null + non-null partition mix.
+ # isNull must keep only the null partition, and the non-matching
entries
+ # must be skipped before their _FILE block is deserialized.
+ from pypaimon.manifest.schema.data_file_meta import DataFileMeta
+
+ schema = Schema.from_pyarrow_schema(
+ pa.schema([('pk', pa.int32()), ('pt', pa.string())]),
+ partition_keys=['pt'])
+ self.catalog.create_table('default.early_isnull_pt', schema, False)
+ table = self.catalog.get_table('default.early_isnull_pt')
+ ws = pa.schema([('pk', pa.int32()), ('pt', pa.string())])
+ wb = table.new_batch_write_builder()
+ w = wb.new_write()
+ w.write_arrow(pa.Table.from_pydict({'pk': [0], 'pt': [None]},
schema=ws))
+ for i in range(1, 6):
+ w.write_arrow(pa.Table.from_pydict(
+ {'pk': [i], 'pt': [f'p{i}']}, schema=ws))
+ wb.new_commit().commit(w.prepare_commit())
+ w.close()
+
+ pb = table.new_read_builder().new_predicate_builder()
+
+ # isNull keeps only the null partition; count DataFileMeta
constructions to
+ # prove the 5 non-null entries were skipped by the early filter
pre-_FILE
+ # (each kept entry builds 2: _FILE + drop_stats copy).
+ constructed = {'n': 0}
+ orig_init = DataFileMeta.__init__
+
+ def counting_init(self_dfm, *args, **kwargs):
+ constructed['n'] += 1
+ orig_init(self_dfm, *args, **kwargs)
+
+ DataFileMeta.__init__ = counting_init
+ try:
+ null_splits = table.new_read_builder().with_filter(
+ pb.is_null('pt')).new_scan().plan().splits()
+ finally:
+ DataFileMeta.__init__ = orig_init
+ self.assertEqual([s.partition.values for s in null_splits], [[None]])
+ self.assertEqual(constructed['n'], 2)
+
+ # complement: equal keeps only the matching non-null partition
+ p3_splits = table.new_read_builder().with_filter(
+ pb.equal('pt', 'p3')).new_scan().plan().splits()
+ self.assertEqual([s.partition.values for s in p3_splits], [['p3']])
+
def _write_test_table(self, table):
write_builder = table.new_batch_write_builder()
