This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 1cca8fa1ed4913fc5f730a2b96701d9b00eaa77b Author: Jingsong Lee <[email protected]> AuthorDate: Mon Sep 14 15:31:04 2026 +0800 [python] Align manifest metadata and prune manifest files by bucket (#9807) --- .../pypaimon/manifest/manifest_file_manager.py | 12 ++++ .../pypaimon/manifest/manifest_list_manager.py | 10 ++++ .../pypaimon/manifest/schema/manifest_file_meta.py | 12 ++++ .../read/scanner/bucket_select_converter.py | 18 +++++- .../pypaimon/read/scanner/file_scanner.py | 12 ++++ .../tests/manifest/manifest_manager_test.py | 61 +++++++++++++++++++++ .../tests/manifest/manifest_schema_test.py | 10 +++- .../pypaimon/tests/pushdown_bucket_test.py | 64 ++++++++++++++++++++++ 8 files changed, 196 insertions(+), 3 deletions(-) diff --git a/paimon-python/pypaimon/manifest/manifest_file_manager.py b/paimon-python/pypaimon/manifest/manifest_file_manager.py index a3ac5cfe74..71a5a149e6 100644 --- a/paimon-python/pypaimon/manifest/manifest_file_manager.py +++ b/paimon-python/pypaimon/manifest/manifest_file_manager.py @@ -358,6 +358,13 @@ class ManifestFileManager: if schema_id is None: schema_id = self.table.table_schema.id + # Rescaling can mix bucket counts in one manifest. Only a common, + # positive count is safe for predicate-driven manifest pruning. + total_buckets = entries[0].total_buckets if entries else None + if total_buckets is not None and ( + total_buckets <= 0 or any(e.total_buckets != total_buckets for e in entries)): + total_buckets = None + partition_columns = list(zip(*(entry.partition.values for entry in entries))) if entries else [] partition_null_counts = [sum(1 for value in col if value is None) for col in partition_columns] partition_min_stats = [ @@ -400,6 +407,11 @@ class ManifestFileManager: null_counts=partition_null_counts, ), schema_id=schema_id, + min_bucket=min((e.bucket for e in entries), default=None), + max_bucket=max((e.bucket for e in entries), default=None), + min_level=min((e.file.level for e in entries), default=None), + max_level=max((e.file.level for e in entries), default=None), min_row_id=min_row_id, max_row_id=max_row_id, + total_buckets=total_buckets, ) diff --git a/paimon-python/pypaimon/manifest/manifest_list_manager.py b/paimon-python/pypaimon/manifest/manifest_list_manager.py index 5b867e1ae2..70ec8f2ae6 100644 --- a/paimon-python/pypaimon/manifest/manifest_list_manager.py +++ b/paimon-python/pypaimon/manifest/manifest_list_manager.py @@ -96,8 +96,13 @@ class ManifestListManager: num_deleted_files=record['_NUM_DELETED_FILES'], partition_stats=partition_stats, schema_id=record['_SCHEMA_ID'], + min_bucket=record.get('_MIN_BUCKET'), + max_bucket=record.get('_MAX_BUCKET'), + min_level=record.get('_MIN_LEVEL'), + max_level=record.get('_MAX_LEVEL'), min_row_id=record.get('_MIN_ROW_ID'), max_row_id=record.get('_MAX_ROW_ID'), + total_buckets=record.get('_TOTAL_BUCKETS'), extra_files=record.get('_EXTRA_FILES'), ) manifest_files.append(manifest_file_meta) @@ -119,8 +124,13 @@ class ManifestListManager: "_NULL_COUNTS": meta.partition_stats.null_counts, }, "_SCHEMA_ID": meta.schema_id, + "_MIN_BUCKET": meta.min_bucket, + "_MAX_BUCKET": meta.max_bucket, + "_MIN_LEVEL": meta.min_level, + "_MAX_LEVEL": meta.max_level, "_MIN_ROW_ID": meta.min_row_id, "_MAX_ROW_ID": meta.max_row_id, + "_TOTAL_BUCKETS": meta.total_buckets, "_EXTRA_FILES": meta.extra_files, } avro_records.append(avro_record) diff --git a/paimon-python/pypaimon/manifest/schema/manifest_file_meta.py b/paimon-python/pypaimon/manifest/schema/manifest_file_meta.py index 2681adacdd..388259b7c4 100644 --- a/paimon-python/pypaimon/manifest/schema/manifest_file_meta.py +++ b/paimon-python/pypaimon/manifest/schema/manifest_file_meta.py @@ -35,6 +35,13 @@ class ManifestFileMeta: max_row_id: Optional[int] = None extra_files: Optional[List[str]] = None + # Append new fields to preserve existing positional constructor arguments. + min_bucket: Optional[int] = None + max_bucket: Optional[int] = None + min_level: Optional[int] = None + max_level: Optional[int] = None + total_buckets: Optional[int] = None + MANIFEST_FILE_META_SCHEMA = { "type": "record", "name": "ManifestFileMeta", @@ -46,8 +53,13 @@ MANIFEST_FILE_META_SCHEMA = { {"name": "_NUM_DELETED_FILES", "type": "long"}, {"name": "_PARTITION_STATS", "type": PARTITION_STATS_SCHEMA}, {"name": "_SCHEMA_ID", "type": "long"}, + {"name": "_MIN_BUCKET", "type": ["null", "int"], "default": None}, + {"name": "_MAX_BUCKET", "type": ["null", "int"], "default": None}, + {"name": "_MIN_LEVEL", "type": ["null", "int"], "default": None}, + {"name": "_MAX_LEVEL", "type": ["null", "int"], "default": None}, {"name": "_MIN_ROW_ID", "type": ["null", "long"], "default": None}, {"name": "_MAX_ROW_ID", "type": ["null", "long"], "default": None}, + {"name": "_TOTAL_BUCKETS", "type": ["null", "int"], "default": None}, {"name": "_EXTRA_FILES", "type": ["null", {"type": "array", "items": "string"}], "default": None}, ] } diff --git a/paimon-python/pypaimon/read/scanner/bucket_select_converter.py b/paimon-python/pypaimon/read/scanner/bucket_select_converter.py index 67a9346157..c952ff886a 100644 --- a/paimon-python/pypaimon/read/scanner/bucket_select_converter.py +++ b/paimon-python/pypaimon/read/scanner/bucket_select_converter.py @@ -64,7 +64,7 @@ result — sound but possibly wider than the per-partition tight set. """ from itertools import product -from typing import Any, Callable, Dict, FrozenSet, List, Optional, Set, Tuple, Union +from typing import Any, Dict, FrozenSet, List, Optional, Set, Tuple, Union from pypaimon.common.predicate import Predicate from pypaimon.schema.data_types import DataField @@ -404,6 +404,17 @@ class _Selector: # forbids false-negatives. return True + def may_contain(self, min_bucket: int, max_bucket: int, total_buckets: int) -> bool: + """Conservatively test an inclusive manifest range without a known partition.""" + if min_bucket < 0 or max_bucket < min_bucket or total_buckets <= 0: + return True + try: + return any(min_bucket <= bucket <= max_bucket + for bucket in self._compute(None, total_buckets)) + except Exception: + # Use the same fail-open behavior as entry-level bucket selection. + return True + def _compute(self, partition, total_buckets: int) -> FrozenSet[int]: cache_key = (_partition_to_cache_key(partition, self._partition_fields), total_buckets) @@ -450,7 +461,7 @@ def create_bucket_selector( predicate: Optional[Predicate], bucket_key_fields: List[DataField], partition_fields: Optional[List[DataField]] = None, -) -> Optional[Callable[[Any, int, int], bool]]: +) -> Optional[_Selector]: """Try to derive a bucket selector from ``predicate`` constrained to ``bucket_key_fields``. @@ -466,6 +477,9 @@ def create_bucket_selector( Returns None when the predicate carries no usable bucket-key constraint at all (caller must NOT prune by bucket). + + ``may_contain(min_bucket, max_bucket, total_buckets)`` tests a + manifest's inclusive bucket range without partition specialisation. """ if predicate is None or not bucket_key_fields: return None diff --git a/paimon-python/pypaimon/read/scanner/file_scanner.py b/paimon-python/pypaimon/read/scanner/file_scanner.py index c114fd656e..c63d3cf9a4 100755 --- a/paimon-python/pypaimon/read/scanner/file_scanner.py +++ b/paimon-python/pypaimon/read/scanner/file_scanner.py @@ -744,6 +744,18 @@ class FileScanner: return not _get_all_fields(self.predicate).issubset(partition_keys) def _filter_manifest_file(self, file: ManifestFileMeta) -> bool: + # explain() counts bucket rejections at entry level, as with the early + # bucket filter. Keep reading those entries when collecting scan stats. + if self.scan_stats is None and file.min_bucket is not None and file.max_bucket is not None: + if self.only_read_real_buckets and file.max_bucket < 0: + return False + if (self._bucket_selector is not None + and file.min_bucket >= 0 + and file.total_buckets is not None + and file.total_buckets > 0 + and not self._bucket_selector.may_contain( + file.min_bucket, file.max_bucket, file.total_buckets)): + return False if not self.partition_key_predicate: return True return self.partition_key_predicate.test_by_simple_stats( diff --git a/paimon-python/pypaimon/tests/manifest/manifest_manager_test.py b/paimon-python/pypaimon/tests/manifest/manifest_manager_test.py index e556c55cf8..65be8ebdc5 100644 --- a/paimon-python/pypaimon/tests/manifest/manifest_manager_test.py +++ b/paimon-python/pypaimon/tests/manifest/manifest_manager_test.py @@ -22,6 +22,7 @@ import sys import tempfile import threading import unittest +from dataclasses import replace from io import BytesIO import fastavro @@ -268,6 +269,36 @@ class ManifestFileManagerTest(_ManifestManagerSetup): ) return entry + def test_manifest_bucket_and_level_stats(self): + manager = self._make_manager() + entries = [self._create_manifest_entry('a', bucket=2), + self._create_manifest_entry('b', bucket=6)] + entries[0].file.level = 3 + entries[1].file.level = 1 + entries[1].kind = 1 + for totals, expected in [([8, 8], 8), ([8, 16], None), + ([0, 8], None), ([8, -1], None)]: + with self.subTest(totals=totals): + for entry, total in zip(entries, totals): + entry.total_buckets = total + metas = manager.rolling_write(entries, 1024 * 1024, 'bucket-stats') + self.assertEqual(len(metas), 1) + meta = metas[0] + self.assertEqual((meta.min_bucket, meta.max_bucket), (2, 6)) + self.assertEqual((meta.min_level, meta.max_level), (1, 3)) + self.assertEqual((meta.num_added_files, meta.num_deleted_files), (1, 1)) + self.assertEqual(meta.total_buckets, expected) + + def test_rolling_manifest_bucket_stats_are_per_file(self): + manager = self._make_manager() + entries = [self._create_manifest_entry(str(i), bucket=i) for i in range(4)] + for entry in entries: + entry.total_buckets = 8 + metas = manager.rolling_write(entries, 1, 'rolling-bucket-stats') + self.assertEqual(len(metas), len(entries)) + self.assertEqual([(m.min_bucket, m.max_bucket, m.total_buckets) for m in metas], + [(i, i, 8) for i in range(4)]) + def test_filter_applied_after_read(self): manager = self._make_manager() @@ -467,6 +498,36 @@ class ManifestListManagerTest(_ManifestManagerSetup): ) manager.write(name, [meta]) + def test_bucket_and_level_stats_round_trip(self): + manager = self._make_manager() + legacy_meta = ManifestFileMeta( + 'legacy', 1024, 1, 0, SimpleStats.empty_stats(), 0, + 10, 109, ['extra']) + meta = replace(legacy_meta, file_name='new', min_bucket=0, max_bucket=7, + min_level=0, max_level=3, total_buckets=8) + manager.write('stats-list', [legacy_meta, meta]) + actual = manager.read('stats-list') + fields = ['min_bucket', 'max_bucket', 'min_level', 'max_level', + 'min_row_id', 'max_row_id', 'total_buckets', 'extra_files'] + for expected, restored in zip([legacy_meta, meta], actual): + self.assertEqual([getattr(restored, f) for f in fields], + [getattr(expected, f) for f in fields]) + + with manager.file_io.new_input_stream(f'{manager.manifest_path}/stats-list') as stream: + data = stream.read() + reader = fastavro.reader(BytesIO(data)) + self.assertEqual([f['name'] for f in reader.writer_schema['fields']][7:], + ['_MIN_BUCKET', '_MAX_BUCKET', '_MIN_LEVEL', '_MAX_LEVEL', + '_MIN_ROW_ID', '_MAX_ROW_ID', '_TOTAL_BUCKETS', '_EXTRA_FILES']) + self.assertEqual([r['_VERSION'] for r in reader], [2, 2]) + legacy_schema = reader.writer_schema + legacy_schema['fields'] = [f for f in legacy_schema['fields'] + if f['name'] not in {'_MIN_BUCKET', '_MAX_BUCKET', + '_MIN_LEVEL', '_MAX_LEVEL', '_TOTAL_BUCKETS'}] + records = list(fastavro.reader(BytesIO(data), reader_schema=legacy_schema)) + self.assertEqual([r['_MIN_ROW_ID'] for r in records], [10, 10]) + self.assertEqual([r['_EXTRA_FILES'] for r in records], [['extra'], ['extra']]) + def test_extra_files_round_trip(self): manager = self._make_manager() expected_extra_files = [None, [], ["extra-1", "extra-2"]] diff --git a/paimon-python/pypaimon/tests/manifest/manifest_schema_test.py b/paimon-python/pypaimon/tests/manifest/manifest_schema_test.py index 1e7bfa3670..368852d8f4 100644 --- a/paimon-python/pypaimon/tests/manifest/manifest_schema_test.py +++ b/paimon-python/pypaimon/tests/manifest/manifest_schema_test.py @@ -130,7 +130,8 @@ class ManifestSchemaTest(unittest.TestCase): expected_fields = [ "_VERSION", "_FILE_NAME", "_FILE_SIZE", "_NUM_ADDED_FILES", "_NUM_DELETED_FILES", "_PARTITION_STATS", "_SCHEMA_ID", - "_MIN_ROW_ID", "_MAX_ROW_ID", "_EXTRA_FILES", + "_MIN_BUCKET", "_MAX_BUCKET", "_MIN_LEVEL", "_MAX_LEVEL", + "_MIN_ROW_ID", "_MAX_ROW_ID", "_TOTAL_BUCKETS", "_EXTRA_FILES", ] for field_name in expected_fields: @@ -144,6 +145,9 @@ class ManifestSchemaTest(unittest.TestCase): self.assertEqual(field_map["_NUM_DELETED_FILES"]["type"], "long") self.assertEqual(field_map["_PARTITION_STATS"]["type"], PARTITION_STATS_SCHEMA) self.assertEqual(field_map["_SCHEMA_ID"]["type"], "long") + for name in ["_MIN_BUCKET", "_MAX_BUCKET", "_MIN_LEVEL", "_MAX_LEVEL", "_TOTAL_BUCKETS"]: + self.assertEqual(field_map[name]["type"], ["null", "int"]) + self.assertIsNone(field_map[name]["default"]) self.assertEqual(field_map["_MIN_ROW_ID"]["type"], ["null", "long"]) self.assertEqual(field_map["_MAX_ROW_ID"]["type"], ["null", "long"]) self.assertEqual(field_map["_EXTRA_FILES"]["type"], @@ -236,7 +240,11 @@ class ManifestSchemaTest(unittest.TestCase): self.assertIsNone(meta.min_row_id) self.assertIsNone(meta.max_row_id) self.assertIsNone(meta.extra_files) + for field in ['min_bucket', 'max_bucket', 'min_level', 'max_level', 'total_buckets']: + self.assertIsNone(getattr(meta, field)) buffer.seek(0) resolved_record = next(fastavro.reader(buffer, reader_schema=MANIFEST_FILE_META_SCHEMA)) self.assertIsNone(resolved_record["_EXTRA_FILES"]) + for field in ['_MIN_BUCKET', '_MAX_BUCKET', '_MIN_LEVEL', '_MAX_LEVEL', '_TOTAL_BUCKETS']: + self.assertIsNone(resolved_record[field]) diff --git a/paimon-python/pypaimon/tests/pushdown_bucket_test.py b/paimon-python/pypaimon/tests/pushdown_bucket_test.py index ca84da02a1..13292d6860 100644 --- a/paimon-python/pypaimon/tests/pushdown_bucket_test.py +++ b/paimon-python/pypaimon/tests/pushdown_bucket_test.py @@ -44,13 +44,18 @@ import random import shutil import tempfile import unittest +from dataclasses import replace from typing import Any, Dict, List +from unittest.mock import patch import pyarrow as pa import pytest from pypaimon import CatalogFactory, Schema from pypaimon.common.predicate_builder import PredicateBuilder +from pypaimon.manifest.manifest_list_manager import ManifestListManager +from pypaimon.manifest.schema.manifest_file_meta import ManifestFileMeta +from pypaimon.manifest.schema.simple_stats import SimpleStats from pypaimon.read.scanner.bucket_select_converter import ( MAX_VALUES, create_bucket_selector) from pypaimon.schema.data_types import AtomicType, DataField @@ -232,6 +237,25 @@ class BucketSelectConverterUnitTest(unittest.TestCase): create_bucket_selector(self.pb_id_val.equal('id', 1), [])) # -- Selector cache + rescale ------------------------------------- + def test_manifest_bucket_range_uses_inclusive_bounds_and_historical_bucket_count(self): + sel = create_bucket_selector( + self.pb_id_val.is_in('id', [1, 7]), [self.id_field]) + for total in [4, 8, 16, 8]: + expected = {_hash_bucket([v], [self.id_field], total) for v in [1, 7]} + for lower in range(total): + for upper in range(lower, total): + self.assertEqual(sel.may_contain(lower, upper, total), + any(lower <= bucket <= upper for bucket in expected)) + + def test_manifest_bucket_range_fails_open(self): + sel = create_bucket_selector(self.pb_id_val.equal('id', 7), [self.id_field]) + for lower, upper, total in [(-2, -1, 8), (-2, 3, 8), (3, 2, 8), + (0, 1, 0), (0, 1, -1)]: + self.assertTrue(sel.may_contain(lower, upper, total)) + invalid = create_bucket_selector( + self.pb_id_val.equal('id', 'invalid-bigint'), [self.id_field]) + self.assertTrue(invalid.may_contain(0, 0, 8)) + def test_selector_caches_per_total_buckets(self): """Selector must answer correctly when the same query applies to different ``total_buckets`` values (the rescale scenario).""" @@ -498,6 +522,7 @@ class PartitionAwareBucketSelectorUnitTest(unittest.TestCase): # 3-arg form with partition=None has the same semantics. for b in range(8): self.assertTrue(sel(None, b, 8)) + self.assertTrue(sel.may_contain(b, b, 8)) def test_selector_partition_not_matching_returns_empty_bucket_set(self): # ``part = 'a' AND id = 1`` on partition {part: 'c'} simplifies to @@ -734,6 +759,45 @@ class BucketPruningIntegrationTest(unittest.TestCase): "Equal on PK still narrows to the writer's bucket " "even when AND'd with a non-bucket-key predicate") + def test_manifest_bucket_pruning_skips_file_reads(self): + table = self._create_pk_table('manifest_bucket_pruning') + for i in range(self.NUM_BUCKETS): + self._write(table, [{'id': i, 'val': i * 11}]) + manifests = ManifestListManager(table).read_all(table.snapshot_manager().get_latest_snapshot()) + expected_buckets = self._expected_buckets(table, [0]) + expected_files = {m.file_name for m in manifests + if any(m.min_bucket <= b <= m.max_bucket for b in expected_buckets)} + self.assertTrue(expected_files) + self.assertLess(len(expected_files), len(manifests)) + + predicate = table.new_read_builder().new_predicate_builder().equal('id', 0) + with patch.object(table.file_io, 'new_input_stream', + wraps=table.file_io.new_input_stream) as read: + got, _ = self._read_with(table, predicate) + opened_manifests = {os.path.basename(str(call.args[0])) for call in read.call_args_list} + opened_manifests &= {m.file_name for m in manifests} + self.assertEqual(opened_manifests, expected_files) + self.assertEqual(got, [{'id': 0, 'val': 0}]) + + def test_manifest_bucket_pruning_keeps_unknown_metadata(self): + table = self._create_pk_table('manifest_bucket_fallback') + predicate = table.new_read_builder().new_predicate_builder().equal('id', 0) + scanner = table.new_read_builder().with_filter(predicate).new_scan().file_scanner + selected = next(iter(self._expected_buckets(table, [0]))) + other = (selected + 1) % self.NUM_BUCKETS + meta = ManifestFileMeta('manifest', 1, 1, 0, SimpleStats.empty_stats(), 0, + min_bucket=other, max_bucket=other, + total_buckets=self.NUM_BUCKETS) + self.assertFalse(scanner._filter_manifest_file(meta)) + for changes in [dict(min_bucket=None), dict(max_bucket=None), + dict(total_buckets=None), dict(total_buckets=0), dict(total_buckets=-1), + dict(min_bucket=-2), dict(min_bucket=selected, max_bucket=selected)]: + with self.subTest(changes=changes): + self.assertTrue(scanner._filter_manifest_file(replace(meta, **changes))) + scanner.only_read_real_buckets = True + self.assertFalse(scanner._filter_manifest_file(replace(meta, min_bucket=-2, max_bucket=-1))) + self.assertTrue(scanner._filter_manifest_file(replace(meta, min_bucket=-2, max_bucket=selected))) + def test_early_filter_skips_full_entry_decode_for_pruned_buckets(self): """Entries the bucket selector rejects must never reach ``GenericRowDeserializer.from_bytes`` for their partition / key
