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 790ec13c7e [python] Align manifest metadata and prune manifest files
by bucket (#9807)
790ec13c7e is described below
commit 790ec13c7e1b771ec23c163dab1fd58eec3de549
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
