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 1ea6e749ea [python] Fix external partition predicate index mismatch in
partition pruning (#8507)
1ea6e749ea is described below
commit 1ea6e749eaecca2b7e587d729cde7f97a82f7f59
Author: XiaoHongbo <[email protected]>
AuthorDate: Thu Jul 9 12:17:07 2026 +0800
[python] Fix external partition predicate index mismatch in partition
pruning (#8507)
---
.../pypaimon/read/scanner/file_scanner.py | 11 +++-
.../pypaimon/tests/partition_predicate_test.py | 73 +++++++++++++++++++++-
2 files changed, 80 insertions(+), 4 deletions(-)
diff --git a/paimon-python/pypaimon/read/scanner/file_scanner.py
b/paimon-python/pypaimon/read/scanner/file_scanner.py
index 153b424b13..cce1d3dd50 100755
--- a/paimon-python/pypaimon/read/scanner/file_scanner.py
+++ b/paimon-python/pypaimon/read/scanner/file_scanner.py
@@ -35,7 +35,9 @@ from pypaimon.read.plan import Plan
from pypaimon.read.push_down_utils import (_get_all_fields,
exclude_predicate_with_fields,
remove_row_id_filter,
- trim_and_transform_predicate)
+ rewrite_predicate_indices,
+ trim_and_transform_predicate,
+ trim_predicate_by_fields)
from pypaimon.read.scan_stats import ScanStats
from pypaimon.read.scanner.append_table_split_generator import \
AppendTableSplitGenerator
@@ -247,7 +249,12 @@ class FileScanner:
self.partition_key_predicate = trim_and_transform_predicate(
self.predicate, self.table.field_names,
self.table.partition_keys)
else:
- self.partition_key_predicate = partition_predicate
+ # External predicate may carry full-schema indices and
non-partition
+ # leaves; drop non-partition leaves and rebind the rest by name to
the
+ # partition-row layout (matches derived path / Java), else
IndexError.
+ self.partition_key_predicate = rewrite_predicate_indices(
+ trim_predicate_by_fields(partition_predicate,
self.table.partition_keys),
+ self.table.partition_keys_fields)
options = self.table.options
# Get split target size and open file cost from table options
self.target_split_size = options.source_split_target_size()
diff --git a/paimon-python/pypaimon/tests/partition_predicate_test.py
b/paimon-python/pypaimon/tests/partition_predicate_test.py
index 2be76739e4..c362def4aa 100644
--- a/paimon-python/pypaimon/tests/partition_predicate_test.py
+++ b/paimon-python/pypaimon/tests/partition_predicate_test.py
@@ -118,15 +118,84 @@ class
TestFileScannerPartitionPredicate(unittest.TestCase):
predicate=predicate, partition_predicate=partition_predicate,
)
- def test_partition_predicate_used_directly(self, *_):
+ def test_partition_predicate_already_partition_layout_is_idempotent(self,
*_):
+ # Already partition-layout: rebind is a no-op on the index.
pred = _partition_builder.equal('dt', '2024-01-15')
scanner = self._scanner(partition_predicate=pred)
- self.assertIs(scanner.partition_key_predicate, pred)
+ self.assertEqual(scanner.partition_key_predicate.field, 'dt')
+ self.assertEqual(scanner.partition_key_predicate.index, 0)
self.assertIsNone(scanner.predicate)
self.assertIsNone(scanner.predicate_for_stats)
self.assertIsNone(scanner.primary_key_predicate)
+ def
test_full_schema_partition_predicate_rewritten_to_partition_index(self, *_):
+ # Full-schema 'region' index is 3; partition row [dt, region] needs 1.
+ full_pred = PredicateBuilder(TABLE_FIELDS).equal('region', 'us-east-1')
+ self.assertEqual(full_pred.index, 3)
+ scanner = self._scanner(partition_predicate=full_pred)
+ self.assertEqual(scanner.partition_key_predicate.index, 1)
+
+ self.assertTrue(scanner._filter_manifest_entry(
+ _manifest_entry(['2024-01-15', 'us-east-1'])))
+ self.assertFalse(scanner._filter_manifest_entry(
+ _manifest_entry(['2024-01-15', 'us-west-2'])))
+ self.assertTrue(scanner._filter_manifest_file(
+ _manifest_file_meta(['2024-01-15', 'us-east-1'], ['2024-01-15',
'us-east-1'])))
+ self.assertFalse(scanner._filter_manifest_file(
+ _manifest_file_meta(['2024-01-15', 'us-west-2'], ['2024-01-15',
'us-west-2'])))
+
+ def test_partition_predicate_drops_non_partition_leaves(self, *_):
+ # Mixed filter: keep+rebind the partition leaf ('region'), drop the
+ # non-partition leaf ('name') instead of raising.
+ builder = PredicateBuilder(TABLE_FIELDS)
+ mixed = builder.and_predicates([
+ builder.equal('region', 'us-east-1'),
+ builder.equal('name', 'foo'),
+ ])
+ scanner = self._scanner(partition_predicate=mixed)
+ self.assertEqual(scanner.partition_key_predicate.field, 'region')
+ self.assertEqual(scanner.partition_key_predicate.index, 1)
+ self.assertTrue(scanner._filter_manifest_entry(
+ _manifest_entry(['2024-01-15', 'us-east-1'])))
+ self.assertFalse(scanner._filter_manifest_entry(
+ _manifest_entry(['2024-01-15', 'us-west-2'])))
+
+ def test_reordered_partition_keys_keep_matching_partition(self, *_):
+ # Partition keys reordered vs schema: 'region' is schema field 0 but
+ # partition-row field 1. Full-schema index 0 would silently test 'dt'
+ # (in-range wrong field) and DROP matching rows -> data loss. Rebind
fixes it.
+ table_fields = [
+ DataField(0, 'region', AtomicType('STRING')),
+ DataField(1, 'dt', AtomicType('STRING')),
+ DataField(2, 'id', AtomicType('INT')),
+ ]
+ partition_fields = [
+ DataField(0, 'dt', AtomicType('STRING')),
+ DataField(1, 'region', AtomicType('STRING')),
+ ]
+ table = _mock_scanner_table()
+ table.field_names = ['region', 'dt', 'id']
+ table.fields = table_fields
+ table.partition_keys = ['dt', 'region']
+ table.partition_keys_fields = partition_fields
+ table.table_schema = Mock(id=0, fields=table_fields)
+
+ full_pred = PredicateBuilder(table_fields).equal('region', 'us-east-1')
+ self.assertEqual(full_pred.index, 0)
+ scanner = FileScanner(table, lambda: ([], None),
+ predicate=None, partition_predicate=full_pred)
+ self.assertEqual(scanner.partition_key_predicate.index, 1)
+
+ def entry(vals):
+ return ManifestEntry(kind=0, partition=GenericRow(vals,
partition_fields),
+ bucket=0, total_buckets=1, file=Mock())
+
+ # region matches -> MUST keep (guards against the silent drop)
+ self.assertTrue(scanner._filter_manifest_entry(entry(['2024-01-15',
'us-east-1'])))
+ # region differs (but dt equals the literal) -> MUST drop
+ self.assertFalse(scanner._filter_manifest_entry(entry(['us-east-1',
'us-west-2'])))
+
def test_no_partition_predicate_derives_from_predicate(self, *_):
full_pred = PredicateBuilder(TABLE_FIELDS).equal('dt', '2024-01-15')
scanner = self._scanner(predicate=full_pred)
