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 f8cbb66c45 [python] Use native partition selectors and preserve row
filters (#10576)
f8cbb66c45 is described below
commit f8cbb66c45c97ac056171798088b3e1a04d5937f
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Oct 15 13:49:41 2026 +0800
[python] Use native partition selectors and preserve row filters (#10576)
---
.github/workflows/ci-python.yml | 2 +
paimon-python/pypaimon/read/native_plan.py | 5 +-
paimon-python/pypaimon/read/read_builder.py | 11 +-
.../pypaimon/read/scanner/file_scanner.py | 6 +-
paimon-python/pypaimon/read/split_read.py | 3 +
paimon-python/pypaimon/read/table_scan.py | 22 +--
.../pypaimon/tests/native_partition_filter_test.py | 197 +++++++++++++++++++++
.../pypaimon/tests/partition_predicate_test.py | 22 +++
.../pypaimon/tests/reader_split_generator_test.py | 17 +-
9 files changed, 268 insertions(+), 17 deletions(-)
diff --git a/.github/workflows/ci-python.yml b/.github/workflows/ci-python.yml
index 797bed8bf6..c34c60d745 100644
--- a/.github/workflows/ci-python.yml
+++ b/.github/workflows/ci-python.yml
@@ -220,6 +220,8 @@ jobs:
normalize_array_for_write,
read_arrow_schema,
)
+ from pypaimon_rust.datafusion import ReadBuilder
+ assert callable(getattr(ReadBuilder, 'with_partition_filter', None))
PY
- name: Run Python tests with native planning, reading, writing,
updating and committing
diff --git a/paimon-python/pypaimon/read/native_plan.py
b/paimon-python/pypaimon/read/native_plan.py
index a024b44c41..315096fe35 100644
--- a/paimon-python/pypaimon/read/native_plan.py
+++ b/paimon-python/pypaimon/read/native_plan.py
@@ -451,7 +451,8 @@ def native_plan(
row_position_slice: Optional[Tuple[int, int]] = None,
row_position_shard: Optional[Tuple[int, int]] = None,
chunk_shuffle: Optional[Tuple[int, int]] = None,
- shard: Optional[Tuple[int, int]] = None) -> Plan:
+ shard: Optional[Tuple[int, int]] = None,
+ partition_predicate: Optional[Predicate] = None) -> Plan:
"""Plan with pypaimon_rust, preserving snapshot metadata.
Native conversion or planning failures are handled by TableScan, which
@@ -464,6 +465,8 @@ def native_plan(
"scan.native-plan.enabled needs pypaimon-rust>=0.3.0 (split
planning API)")
builder = _configure_native_read_builder(
_native_read_builder(table), predicate, limit, projection)
+ if partition_predicate is not None:
+ builder =
builder.with_partition_filter(_predicate_to_native(partition_predicate))
if row_ranges is not None:
builder = builder.with_row_ranges(row_ranges)
if incremental_range is None:
diff --git a/paimon-python/pypaimon/read/read_builder.py
b/paimon-python/pypaimon/read/read_builder.py
index 6bf52e773c..f990d7ded6 100644
--- a/paimon-python/pypaimon/read/read_builder.py
+++ b/paimon-python/pypaimon/read/read_builder.py
@@ -115,7 +115,16 @@ class ReadBuilder:
self._predicate = predicate
return self
- def with_partition_filter(self, partition_filter: Predicate) ->
'ReadBuilder':
+ def with_partition_filter(self, partition_filter: Optional[Predicate]) ->
'ReadBuilder':
+ """Replace scan-only partition selection; None restores derived
pruning.
+
+ As with Java's PartitionPredicate, only partition fields are allowed.
+ Ordinary row predicates remain independent and are checked by the
reader.
+ """
+ if partition_filter is not None and (
+ not self.table.partition_keys
+ or not
predicate_field_names(partition_filter).issubset(self.table.partition_keys)):
+ raise ValueError('Partition filter must reference only partition
keys of a partitioned table')
self._partition_filter = partition_filter
return self
diff --git a/paimon-python/pypaimon/read/scanner/file_scanner.py
b/paimon-python/pypaimon/read/scanner/file_scanner.py
index 7a60208a41..b938a7f7d2 100755
--- a/paimon-python/pypaimon/read/scanner/file_scanner.py
+++ b/paimon-python/pypaimon/read/scanner/file_scanner.py
@@ -231,6 +231,7 @@ class FileScanner:
self.table: FileStoreTable = table
self.manifest_scanner = manifest_scanner
self.predicate = predicate
+ self._explicit_partition_filter = partition_predicate is not None
row_ranges = (
_row_ranges_from_predicate(predicate) if predicate else None
)
@@ -749,7 +750,10 @@ class FileScanner:
return splits
if self.data_evolution and self.deletion_vectors_enabled:
return splits
- if self._has_non_partition_filter() or
self.auth_has_non_partition_filter:
+ if (self._has_non_partition_filter() or
self.auth_has_non_partition_filter
+ # An explicit selector replaces ordinary partition pruning.
+ # Count rows only after the reader has checked the full filter.
+ or (self._explicit_partition_filter and self.predicate is not
None)):
return splits
scanned_row_count = 0
diff --git a/paimon-python/pypaimon/read/split_read.py
b/paimon-python/pypaimon/read/split_read.py
index e6faeba3f1..e23db8092d 100644
--- a/paimon-python/pypaimon/read/split_read.py
+++ b/paimon-python/pypaimon/read/split_read.py
@@ -1092,6 +1092,9 @@ class RawFileSplitRead(SplitRead):
if (self.predicate_for_reader
and (self.table.is_primary_key_table
or not self._arrow_filter_pushdown_enabled
+ # Partition values are injected after the physical read;
+ # their predicates cannot have been evaluated by Parquet.
+ or
predicate_field_names(self.predicate).intersection(self.table.partition_keys)
or self.deletion_file_readers
or self._physical_row_ranges
or any(file.schema_id != self.table.table_schema.id
diff --git a/paimon-python/pypaimon/read/table_scan.py
b/paimon-python/pypaimon/read/table_scan.py
index 771b287295..40f0383d30 100755
--- a/paimon-python/pypaimon/read/table_scan.py
+++ b/paimon-python/pypaimon/read/table_scan.py
@@ -142,12 +142,11 @@ class TableScan:
database_name = self.table.identifier.get_database_name()
if not database_name or database_name == UNKNOWN_DATABASE:
return False
- if self.table.options.query_auth_enabled:
- # The binding currently carries explicit partition selectors as
- # ordinary filters. Keep them on Python, where masked scan-only
- # selectors can be rejected as Java requires.
- if self.partition_predicate is not None:
+ if self.partition_predicate is not None:
+ from pypaimon.read.native_plan import native_method_available
+ if not native_method_available('ReadBuilder',
'with_partition_filter'):
return False
+ if self.table.options.query_auth_enabled:
# The filesystem path has no catalog-issued authorization session.
loader = getattr(getattr(self.table, 'catalog_environment', None),
'catalog_loader', None)
from pypaimon.read.native_plan import _catalog_metastore
@@ -219,20 +218,17 @@ class TableScan:
extra_options['row_position_slice'] = (
fs.start_pos_of_this_subtask,
fs.end_pos_of_this_subtask)
row_ranges = self._native_row_ranges()
- native_predicate = self.predicate
if self.partition_predicate is not None:
- native_predicate = PredicateBuilder.and_predicates([
- predicate for predicate in (
- native_predicate,
- self.file_scanner.partition_key_predicate,
- ) if predicate is not None
- ])
+ # Explicit partition selectors are scan-only. Keep the ordinary
+ # row predicate separate so core can apply Java's override and
+ # masked-partition validation semantics.
+ extra_options['partition_predicate'] =
fs.partition_key_predicate
# Append positions still precede LIMIT. Rust owns PK shards and
# source-backed sorted-index refinement before applying LIMIT.
defer_limit = has_distribution and not
self.table.is_primary_key_table
plan = native_plan(
self.table,
- predicate=native_predicate,
+ predicate=self.predicate,
limit=None if defer_limit else self.limit,
projection=(
[field.name for field in self._read_type]
diff --git a/paimon-python/pypaimon/tests/native_partition_filter_test.py
b/paimon-python/pypaimon/tests/native_partition_filter_test.py
new file mode 100644
index 0000000000..ad6ac844a8
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_partition_filter_test.py
@@ -0,0 +1,197 @@
+# 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.
+
+"""Explicit partition selectors through REST Python and Native scan/read
paths."""
+
+import json
+from contextlib import ExitStack
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import Schema
+from pypaimon.api.api_response import AuthTableQueryResponse
+from pypaimon.common.predicate_builder import PredicateBuilder
+from pypaimon.read.native_plan import _native_table, native_plan,
native_runtime_available
+from pypaimon.schema.schema_change import SchemaChange
+from pypaimon.tests import native_plan_rest_test
+
+rest_catalog = native_plan_rest_test.rest_catalog
+
+pytestmark = [pytest.mark.native_plan, pytest.mark.skipif(
+ not native_runtime_available(), reason='Rust planner required')]
+
+
[email protected](params=[(mode, auth) for mode in ('append', 'pk', 'de') for
auth in (False, True)],
+ ids=lambda case: '%s-%s' % (case[0], 'auth' if case[1] else
'plain'))
+def source(rest_catalog, request):
+ catalog, server = rest_catalog
+ mode, auth = request.param
+ schema = pa.schema([('id', pa.int32()), ('p', pa.string()),
+ ('value', pa.int32()), ('region', pa.string())])
+ options = {'read.batch-size': '1'}
+ if mode == 'pk':
+ options['bucket'] = '1'
+ elif mode == 'de':
+ options.update({'data-evolution.enabled': 'true',
'row-tracking.enabled': 'true'})
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ schema, partition_keys=['region', 'p'],
+ primary_keys=['id', 'p', 'region'] if mode == 'pk' else [],
options=options), False)
+ table = catalog.get_table('default.t')
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(pa.Table.from_pylist([
+ {'id': 1, 'p': 'a', 'value': 10, 'region': 'east'},
+ {'id': 2, 'p': 'b', 'value': 20, 'region': 'east'},
+ {'id': 3, 'p': 'a', 'value': 30, 'region': 'west'},
+ {'id': 4, 'p': 'b', 'value': 40, 'region': 'west'},
+ ], schema=schema))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+ if auth:
+ catalog.alter_table(table.identifier,
[SchemaChange.set_option('query-auth.enabled', 'true')])
+ table = catalog.get_table('default.t')
+ return table, server, auth
+
+
+def _read(table, native, selector, predicate=None, limit=None,
selector_first=False):
+ table = table.copy({'scan.native-plan.enabled': str(native).lower(),
+ 'read.native.enabled': str(native).lower()})
+ builder = table.new_read_builder().with_projection(['id'])
+ if selector_first:
+ builder.with_partition_filter(selector)
+ if predicate is not None:
+ builder.with_filter(predicate)
+ if not selector_first:
+ builder.with_partition_filter(selector)
+ if limit is not None:
+ builder.with_limit(limit)
+ scan = builder.new_scan()
+ with ExitStack() as stack:
+ if native:
+ stack.enter_context(patch.object(scan.file_scanner, 'scan',
+
side_effect=AssertionError('Native plan fell back')))
+
stack.enter_context(patch('pypaimon.read.table_read.TableRead._create_split_read',
+ side_effect=AssertionError('Native read
fell back')))
+ plan = scan.plan()
+ rows = builder.new_read().to_arrow(plan.splits()).to_pylist()
+ if native:
+ assert all(getattr(split, '_native_split', None) is not None for split
in plan.splits())
+ return plan, sorted(row['id'] for row in rows)
+
+
[email protected]('selector_first', [False, True])
+def
test_partition_selector_keeps_the_ordinary_predicate_on_hidden_columns(source,
selector_first):
+ table, _, _ = source
+ predicates = PredicateBuilder(table.fields)
+ for native in (False, True):
+ plan, result = _read(table, native, predicates.equal('region', 'east'),
+ predicates.equal('p', 'a'),
selector_first=selector_first)
+ assert plan.splits()
+ assert result == [1]
+
+
[email protected]('limit', [None, 0, 1])
+def test_conflicting_selectors_never_return_rows(source, limit):
+ table, _, _ = source
+ predicates = PredicateBuilder(table.fields)
+ for native in (False, True):
+ _, result = _read(table, native, predicates.equal('region', 'east'),
+ predicates.equal('region', 'west'), limit=limit)
+ assert result == []
+
+
+def test_limit_counts_only_rows_matching_the_ordinary_predicate(source):
+ table, _, _ = source
+ predicates = PredicateBuilder(table.fields)
+ selector = predicates.is_in('region', ['east', 'west'])
+ for native in (False, True):
+ _, result = _read(table, native, selector, predicates.equal('region',
'west'), limit=1)
+ assert len(result) == 1 and result[0] in (3, 4)
+
+
+def test_replacing_and_clearing_partition_selector(source):
+ table, _, _ = source
+ predicates = PredicateBuilder(table.fields)
+ for native in (False, True):
+ builder = table.copy({'scan.native-plan.enabled': str(native).lower(),
+ 'read.native.enabled':
str(native).lower()}).new_read_builder()
+ builder.with_projection(['id']).with_filter(predicates.equal('p', 'a'))
+ builder.with_partition_filter(predicates.equal('region', 'east'))
+ builder.with_partition_filter(predicates.equal('region', 'west'))
+ plan = builder.new_scan().plan()
+ assert builder.new_read().to_arrow(plan.splits()).to_pylist() ==
[{'id': 3}]
+ builder.with_partition_filter(None)
+ plan = builder.new_scan().plan()
+ assert sorted(row['id'] for row in
builder.new_read().to_arrow(plan.splits()).to_pylist()) == [1, 3]
+
+
+def test_unmasked_partition_selector_with_a_masked_row_predicate(source):
+ table, server, auth = source
+ if not auth:
+ pytest.skip('requires query authorization')
+ server.table_query_auth_store[table.identifier.get_full_name()] =
AuthTableQueryResponse(
+ column_masking={'p': json.dumps({'name': 'UPPER', 'inputs': [{'index':
1, 'name': 'p', 'type': 'STRING'}]})})
+ predicates = PredicateBuilder(table.fields)
+ for native in (False, True):
+ _, result = _read(table, native, predicates.equal('region', 'east'),
+ predicates.equal('p', 'A'), limit=1)
+ assert result == [1]
+ # Masking p must not forbid a selector that references only region.
+ _, result = _read(table, native, predicates.equal('region', 'west'))
+ assert result == [3, 4]
+
+
+def test_native_plan_rejects_masked_selector_before_any_data_is_read(source):
+ table, server, auth = source
+ if not auth:
+ pytest.skip('requires query authorization')
+ server.table_query_auth_store[table.identifier.get_full_name()] =
AuthTableQueryResponse(
+ column_masking={'p': json.dumps({'name': 'CONCAT', 'inputs':
['masked']})})
+ for limit in (0, 1):
+ with pytest.raises(NotImplementedError, match="partition key 'p' is
masked"):
+ native_plan(table, projection=['id'], limit=limit,
+
partition_predicate=PredicateBuilder(table.fields).equal('p', 'missing'))
+
+
+def test_native_authorization_includes_scan_only_partition_operands(source):
+ table, server, auth = source
+ if not auth:
+ pytest.skip('requires query authorization')
+ selections = []
+
+ def authorize(select):
+ selections.append(select)
+ return AuthTableQueryResponse()
+
+ server.table_query_auth_store[table.identifier.get_full_name()] = authorize
+ _read(table, True, PredicateBuilder(table.fields).equal('region', 'east'))
+ assert selections and all(set(selection) == {'id', 'region'} for selection
in selections)
+
+
+def
test_native_reader_rechecks_partition_filter_on_independently_planned_splits(source):
+ table, _, _ = source
+ native = _native_table(table)
+ plan = native.new_read_builder().new_scan().plan()
+ builder = native.new_read_builder().with_projection(['id']).with_filter(
+ {'method': 'equal', 'field': 'region', 'literals': ['west']})
+ rows = [row for batch in builder.new_read().read_arrow(plan.splits()) for
row in batch.to_pylist()]
+ assert sorted(row['id'] for row in rows) == [3, 4]
diff --git a/paimon-python/pypaimon/tests/partition_predicate_test.py
b/paimon-python/pypaimon/tests/partition_predicate_test.py
index c362def4aa..f1d0e96fb2 100644
--- a/paimon-python/pypaimon/tests/partition_predicate_test.py
+++ b/paimon-python/pypaimon/tests/partition_predicate_test.py
@@ -474,3 +474,25 @@ class
TestCommitScannerPartitionPredicate(unittest.TestCase):
self.assertEqual(len(result), 1)
self.assertEqual(tuple(result[0].partition.values), ('p1', 'us'))
+
+
+class ReadBuilderPartitionFilterTest(unittest.TestCase):
+ def
test_public_selector_rejects_non_partition_fields_without_changing_state(self):
+ from types import SimpleNamespace
+ from pypaimon.read.read_builder import ReadBuilder
+
+ builder = ReadBuilder(SimpleNamespace(partition_keys=['dt', 'region']))
+ predicates = PredicateBuilder(TABLE_FIELDS)
+ selector = predicates.equal('region', 'us-east-1')
+ builder.with_partition_filter(selector)
+ for invalid in [predicates.equal('name', 'x'),
+ predicates.and_predicates([selector,
predicates.equal('name', 'x')]),
+ predicates.or_predicates([selector,
predicates.equal('name', 'x')])]:
+ with self.assertRaisesRegex(ValueError, 'only partition keys'):
+ builder.with_partition_filter(invalid)
+ self.assertIs(builder._partition_filter, selector)
+ self.assertIs(builder.with_partition_filter(None), builder)
+ self.assertIsNone(builder._partition_filter)
+ unpartitioned = ReadBuilder(SimpleNamespace(partition_keys=[]))
+ with self.assertRaisesRegex(ValueError, 'partitioned table'):
+ unpartitioned.with_partition_filter(selector)
diff --git a/paimon-python/pypaimon/tests/reader_split_generator_test.py
b/paimon-python/pypaimon/tests/reader_split_generator_test.py
index 6f8d1b1962..c2610112b5 100644
--- a/paimon-python/pypaimon/tests/reader_split_generator_test.py
+++ b/paimon-python/pypaimon/tests/reader_split_generator_test.py
@@ -338,7 +338,9 @@ class ApplyPushDownLimitUnitTest(unittest.TestCase):
limit,
has_non_partition_filter=False,
data_evolution=False,
- deletion_vectors_enabled=False):
+ deletion_vectors_enabled=False,
+ explicit_partition_filter=False,
+ has_row_filter=False):
from pypaimon.read.scanner.file_scanner import FileScanner
class _FakeScanner:
@@ -349,6 +351,8 @@ class ApplyPushDownLimitUnitTest(unittest.TestCase):
scanner.data_evolution = data_evolution
scanner.deletion_vectors_enabled = deletion_vectors_enabled
scanner.auth_has_non_partition_filter = False
+ scanner._explicit_partition_filter = explicit_partition_filter
+ scanner.predicate = object() if has_row_filter else None
scanner._has_non_partition_filter = lambda: has_non_partition_filter
return FileScanner._apply_push_down_limit(scanner, splits)
@@ -407,6 +411,17 @@ class ApplyPushDownLimitUnitTest(unittest.TestCase):
[s_raw, s_raw, s_raw], limit=5, has_non_partition_filter=True)
self.assertEqual(len(result), 3)
+ def
test_explicit_partition_selector_defers_limit_until_ordinary_filter(self):
+ first = self._split(raw_convertible=True, row_count=10,
merged_row_count=10)
+ later = self._split(raw_convertible=True, row_count=10,
merged_row_count=10)
+ splits = [first, later]
+ self.assertEqual(self._apply(splits, limit=1,
explicit_partition_filter=True,
+ has_row_filter=True), splits)
+ # Without a residual row filter, the explicit selection has already
+ # constrained every split, so merged-row-count LIMIT remains safe.
+ self.assertEqual(self._apply(splits, limit=1,
explicit_partition_filter=True), [first])
+ self.assertEqual(self._apply(splits, limit=1, has_row_filter=True),
[first])
+
def test_data_evolution_deletion_vectors_disable_limit_pushdown(self):
s_raw = self._split(raw_convertible=True, row_count=10,
merged_row_count=10)
result = self._apply(
