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 4eec1297a3 [python] Keep projected nested ROW predicates on native
reads (#10637)
4eec1297a3 is described below
commit 4eec1297a31f3baf968dcc3b7fa318f30d58d96c
Author: Jingsong Lee <[email protected]>
AuthorDate: Sun Oct 18 12:55:08 2026 +0800
[python] Keep projected nested ROW predicates on native reads (#10637)
---
paimon-python/pypaimon/common/predicate.py | 11 +-
paimon-python/pypaimon/read/native_plan.py | 7 +-
paimon-python/pypaimon/read/push_down_utils.py | 42 +++
paimon-python/pypaimon/read/read_builder.py | 19 +-
.../pypaimon/read/scanner/file_scanner.py | 37 +-
paimon-python/pypaimon/read/split_read.py | 8 +-
paimon-python/pypaimon/read/table_read.py | 72 +++-
paimon-python/pypaimon/tests/global_index_test.py | 2 +
.../pypaimon/tests/native_nested_predicate_test.py | 385 +++++++++++++++++++++
9 files changed, 554 insertions(+), 29 deletions(-)
diff --git a/paimon-python/pypaimon/common/predicate.py
b/paimon-python/pypaimon/common/predicate.py
index 8967e2cecb..bc7f564bfd 100644
--- a/paimon-python/pypaimon/common/predicate.py
+++ b/paimon-python/pypaimon/common/predicate.py
@@ -53,6 +53,8 @@ class Predicate:
index: Optional[int]
field: Optional[str]
literals: Optional[List[Any]] = None
+ # Physical source path for predicates built against a flat projection view.
+ field_path: Optional[List[str]] = None
testers: ClassVar[Dict[str, Any]] = {}
@@ -61,14 +63,16 @@ class Predicate:
method=self.method,
index=index,
field=self.field,
- literals=self.literals)
+ literals=self.literals,
+ field_path=self.field_path)
def new_literals(self, literals: List[Any]):
return Predicate(
method=self.method,
index=self.index,
field=self.field,
- literals=literals)
+ literals=literals,
+ field_path=self.field_path)
def test(self, record: InternalRow) -> bool:
if self.method == 'and':
@@ -88,6 +92,9 @@ class Predicate:
return all(p.test_by_simple_stats(stat, row_count) for p in
self.literals)
if self.method == 'or':
return any(p.test_by_simple_stats(stat, row_count) for p in
self.literals)
+ if self.field_path and len(self.field_path) > 1:
+ # Parent ROW stats do not describe a nested field transform.
+ return True
index = self.index
if index is None or index < 0:
diff --git a/paimon-python/pypaimon/read/native_plan.py
b/paimon-python/pypaimon/read/native_plan.py
index f968f85154..4c30c5e525 100644
--- a/paimon-python/pypaimon/read/native_plan.py
+++ b/paimon-python/pypaimon/read/native_plan.py
@@ -243,11 +243,16 @@ def _predicate_to_native(predicate: Predicate) -> dict:
'method': predicate.method,
'children': [_predicate_to_native(child) for child in children],
}
- return {
+ result = {
'method': predicate.method,
'field': predicate.field,
'literals': list(predicate.literals or []),
}
+ if predicate.field_path:
+ result['field'] = predicate.field_path[0]
+ if len(predicate.field_path) > 1:
+ result['nested_path'] = list(predicate.field_path[1:])
+ return result
def _resolved_rest_table_response(table):
diff --git a/paimon-python/pypaimon/read/push_down_utils.py
b/paimon-python/pypaimon/read/push_down_utils.py
index 1b096053e1..c9ae91d722 100644
--- a/paimon-python/pypaimon/read/push_down_utils.py
+++ b/paimon-python/pypaimon/read/push_down_utils.py
@@ -139,6 +139,48 @@ def predicate_field_names(predicate: Predicate) ->
Set[str]:
return involved_fields
+def predicate_for_physical_fields(predicate: Optional[Predicate], fields:
List[DataField]) -> Optional[Predicate]:
+ """Bind projected top-level leaves; omit nested leaves from file pruning.
+
+ Like Java's NestedFieldTransform, a nested ROW leaf cannot use the parent
+ column's stats, bucket key or scalar index. Its flat output name can also
+ coincide with an unrelated physical column. Keep exact filtering on the
+ original predicate after projection.
+ """
+ if predicate is None:
+ return None
+ if predicate.method in ('and', 'or'):
+ children = [predicate_for_physical_fields(child, fields) for child in
predicate.literals or []]
+ if predicate.method == 'or' and any(child is None for child in
children):
+ return None
+ kept = [child for child in children if child is not None]
+ combine = (PredicateBuilder.and_predicates if predicate.method == 'and'
+ else PredicateBuilder.or_predicates)
+ return combine(kept)
+ if not predicate.field_path:
+ return predicate
+ if len(predicate.field_path) != 1:
+ return None
+ name = predicate.field_path[0]
+ index = next((index for index, field in enumerate(fields) if field.name ==
name), None)
+ if index is None:
+ # Synthesized row-tracking columns are absent from table.fields but
+ # already carry the physical name used by the row-range selector.
+ if name == predicate.field:
+ return predicate
+ return None
+ return Predicate(predicate.method, index, name, predicate.literals,
predicate.field_path)
+
+
+def predicate_matches_physical_names(predicate: Optional[Predicate]) -> bool:
+ """Whether a predicate's output names denote the same physical columns."""
+ if predicate is None:
+ return True
+ if predicate.method in ('and', 'or'):
+ return all(predicate_matches_physical_names(child) for child in
predicate.literals or [])
+ return not predicate.field_path or predicate.field_path ==
[predicate.field]
+
+
def _get_all_fields(predicate: Predicate) -> Set[str]:
return predicate_field_names(predicate)
diff --git a/paimon-python/pypaimon/read/read_builder.py
b/paimon-python/pypaimon/read/read_builder.py
index cb5373cdc8..e4d3297d71 100644
--- a/paimon-python/pypaimon/read/read_builder.py
+++ b/paimon-python/pypaimon/read/read_builder.py
@@ -24,7 +24,7 @@ from pypaimon.common.predicate import Predicate
from pypaimon.common.predicate_builder import PredicateBuilder
from pypaimon.read.explain import ExplainResult, ExplainSplitInfo, PruningStat
from pypaimon.read.explain_render import render_predicate
-from pypaimon.read.push_down_utils import predicate_field_names
+from pypaimon.read.push_down_utils import predicate_field_names,
predicate_matches_physical_names
from pypaimon.read.query_auth_split import QueryAuthSplit
from pypaimon.read.scan_stats import ScanStats
from pypaimon.read.split import Split
@@ -86,9 +86,17 @@ def _normalize_sql_literals(expression: str) -> str:
class _ReadPredicateBuilder(PredicateBuilder):
- def __init__(self, fields, unsupported_fields):
+ def __init__(self, fields, unsupported_fields, paths=None):
super().__init__(fields)
self._unsupported_fields = unsupported_fields
+ self._paths = paths or {}
+
+ def _build_predicate(self, method, field, literals=None):
+ predicate = super()._build_predicate(method, field, literals)
+ path = self._paths.get(field)
+ if path:
+ predicate.field_path = list(path)
+ return predicate
def _get_field_index(self, field: str) -> int:
if field in self._unsupported_fields:
@@ -123,6 +131,7 @@ class ReadBuilder:
"""
if partition_filter is not None and (
not self.table.partition_keys
+ or not predicate_matches_physical_names(partition_filter)
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
@@ -208,10 +217,12 @@ class ReadBuilder:
# List projections expose flat leaf names to the Python predicate
# evaluator. Derive that view without changing the reader request.
fields = self.read_type()
+ paths = {}
if self._output_projection is not None and not
self._output_projection.named:
- fields = reader_adapter(fields, self._table_read_fields(),
self._output_projection)[0]
+ fields, name_paths = reader_adapter(fields,
self._table_read_fields(), self._output_projection)
+ paths = {field.name: path for field, path in zip(fields,
name_paths)}
return _ReadPredicateBuilder(
- fields, self._map_key_output_names())
+ fields, self._map_key_output_names(), paths)
def explain(self, verbose: bool = False) -> ExplainResult:
"""Produce a structured scan plan for this builder.
diff --git a/paimon-python/pypaimon/read/scanner/file_scanner.py
b/paimon-python/pypaimon/read/scanner/file_scanner.py
index bcbf73e033..b24624e69f 100755
--- a/paimon-python/pypaimon/read/scanner/file_scanner.py
+++ b/paimon-python/pypaimon/read/scanner/file_scanner.py
@@ -36,6 +36,8 @@ from pypaimon.read.plan import Plan
from pypaimon.read.scan_distribution import validate_shard, validate_slice
from pypaimon.read.push_down_utils import (_get_all_fields,
exclude_predicate_with_fields,
+ predicate_for_physical_fields,
+ predicate_matches_physical_names,
remove_row_id_filter,
rewrite_predicate_indices,
trim_and_transform_predicate,
@@ -275,16 +277,17 @@ class FileScanner:
def _set_predicate(self, predicate):
self.predicate = predicate
- row_ranges = _row_ranges_from_predicate(predicate) if predicate else
None
+ physical = predicate_for_physical_fields(predicate, self.table.fields)
+ row_ranges = _row_ranges_from_predicate(physical) if physical else None
self.predicate_for_stats = (
- remove_row_id_filter(predicate) if row_ranges is not None else
predicate)
+ remove_row_id_filter(physical) if row_ranges is not None else
physical)
self.predicate_for_stats = exclude_predicate_with_fields(
self.predicate_for_stats, SpecialFields.SYSTEM_FIELD_NAMES |
set(self.table.partition_keys))
self.primary_key_predicate = trim_and_transform_predicate(
- predicate, self.table.field_names, self.table.trimmed_primary_keys)
+ physical, self.table.field_names, self.table.trimmed_primary_keys)
if self._partition_predicate is None:
self.partition_key_predicate = trim_and_transform_predicate(
- predicate, self.table.field_names, self.table.partition_keys)
+ physical, self.table.field_names, self.table.partition_keys)
else:
self.partition_key_predicate = rewrite_predicate_indices(
trim_predicate_by_fields(self._partition_predicate,
self.table.partition_keys),
@@ -418,8 +421,9 @@ class FileScanner:
def _apply_primary_key_sorted_indexes(self, splits, snapshot=None):
snapshot = snapshot if snapshot is not None else self._scanned_snapshot
+ predicate = predicate_for_physical_fields(self.predicate,
self.table.fields)
if (not self.table.options.global_index_enabled()
- or self.predicate is None
+ or predicate is None
or snapshot is None
or not splits):
return splits
@@ -449,7 +453,7 @@ class FileScanner:
evaluated = primary_key_sorted_index_scan.evaluate(
index_plan,
self.table.fields,
- self.predicate,
+ predicate,
definitions,
primary_key_sorted_index_scan.reader_factory(self.table),
)
@@ -482,7 +486,8 @@ class FileScanner:
if isinstance(global_index_result, ScoredGlobalIndexResult):
score_getter = global_index_result.score_getter()
if row_ranges is None and self.predicate is not None:
- row_ranges = _row_ranges_from_predicate(self.predicate)
+ row_ranges = _row_ranges_from_predicate(
+ predicate_for_physical_fields(self.predicate,
self.table.fields))
if row_ranges is not None and not row_ranges:
return [], DataEvolutionSplitGenerator(
@@ -545,7 +550,8 @@ class FileScanner:
def _eval_global_index(self, snapshot=None):
# Snapshot indexes describe current state, not historical change
events.
- if self.is_streaming or self.predicate is None:
+ predicate = predicate_for_physical_fields(self.predicate,
self.table.fields)
+ if self.is_streaming or predicate is None:
return None
# Check if global index is enabled
@@ -558,20 +564,20 @@ class FileScanner:
scanner = DataEvolutionGlobalIndexScanner.create(
self.table,
partition_filter=self.partition_key_predicate,
- predicate=self.predicate,
+ predicate=predicate,
snapshot=snapshot,
)
if scanner is None:
return None
with scanner:
- evaluation = scanner.scan_with_coverage(self.predicate)
+ evaluation = scanner.scan_with_coverage(predicate)
if evaluation is None:
return None
scalar_mode = self.table.options.scalar_index_search_mode()
return _GlobalIndexPlanningResult(
evaluation.result,
scanner.unindexed_ranges(
- self.predicate,
+ predicate,
search_mode=scalar_mode,
contributing_field_ids=(
evaluation.contributing_field_ids),
@@ -729,7 +735,7 @@ class FileScanner:
if self._user_predicate is not None:
partition_keys = set(self.table.partition_keys or [])
non_partition_fields = _get_all_fields(self._user_predicate) -
partition_keys
- if non_partition_fields:
+ if non_partition_fields or not
predicate_matches_physical_names(self._user_predicate):
raise ValueError(
"chunk_shuffle predicate must reference only partition
keys; "
"got non-partition fields: "
@@ -768,6 +774,8 @@ class FileScanner:
"""Mirror Java ``SnapshotReaderImpl.hasNonPartitionFilter``."""
if self.predicate is None:
return False
+ if not predicate_matches_physical_names(self.predicate):
+ return True
partition_keys = set(self.table.partition_keys or [])
return not _get_all_fields(self.predicate).issubset(partition_keys)
@@ -806,6 +814,9 @@ class FileScanner:
"""
if self.predicate is None:
return None
+ predicate = predicate_for_physical_fields(self.predicate,
self.table.fields)
+ if predicate is None:
+ return None
# ``bucket_mode()`` returns HASH_FIXED only when ``options.bucket()
# > 0``; other modes (DYNAMIC / POSTPONE / UNAWARE / CROSS_PARTITION)
# have no fixed hash → bucket mapping at write time and must NOT
@@ -845,7 +856,7 @@ class FileScanner:
if name in self.table.field_dict
]
return create_bucket_selector(
- self.predicate, bucket_key_fields,
+ predicate, bucket_key_fields,
partition_fields=partition_fields,
)
diff --git a/paimon-python/pypaimon/read/split_read.py
b/paimon-python/pypaimon/read/split_read.py
index 9563abf54e..a50deffb09 100644
--- a/paimon-python/pypaimon/read/split_read.py
+++ b/paimon-python/pypaimon/read/split_read.py
@@ -40,6 +40,8 @@ from pypaimon.read.interval_partition import
IntervalPartition, SortedRun
from pypaimon.read.partition_info import PartitionInfo
from pypaimon.read.push_down_utils import (
predicate_field_names,
+ predicate_for_physical_fields,
+ predicate_matches_physical_names,
predicate_supports_arrow_filter,
rewrite_predicate_indices,
trim_predicate_by_fields,
@@ -195,6 +197,7 @@ class SplitRead(ABC):
if (
self.predicate is not None
and
predicate_field_names(self.predicate).issubset(read_type_names)
+ and predicate_matches_physical_names(self.predicate)
):
self.predicate_for_reader = rewrite_predicate_indices(
self.predicate, read_type
@@ -260,12 +263,13 @@ class SplitRead(ABC):
# see missing/NULL versions and renumber positional row IDs.
return None
elif self.table.is_primary_key_table:
- pk_predicate = trim_predicate_by_fields(self.predicate,
self.table.primary_keys)
+ physical = predicate_for_physical_fields(self.predicate,
self.table.fields)
+ pk_predicate = trim_predicate_by_fields(physical,
self.table.primary_keys)
if not pk_predicate:
return None
return pk_predicate
else:
- return self.predicate
+ return predicate_for_physical_fields(self.predicate,
self.table.fields)
def _has_row_tracking_predicate(self) -> bool:
return (self.table.options.row_tracking_enabled()
diff --git a/paimon-python/pypaimon/read/table_read.py
b/paimon-python/pypaimon/read/table_read.py
index c041b49be4..05a77f3615 100644
--- a/paimon-python/pypaimon/read/table_read.py
+++ b/paimon-python/pypaimon/read/table_read.py
@@ -177,8 +177,11 @@ class TableRead:
self.predicate = predicate
self.read_type = read_type
self.output_projection = output_projection
- self._adapter_read_type = (self._reader_adapter()[0]
- if output_projection is not None else
read_type)
+ adapter = (self._reader_adapter() if output_projection is not None
+ else (read_type, [[field.name] for field in read_type]))
+ self._adapter_read_type = adapter[0]
+ physical_names = {field.name for field in self._table_read_fields()}
+ self.predicate = self._bind_predicate_to_adapter(self.predicate,
*adapter, physical_names)
# Split readers may need predicate-only columns that are absent from
# the requested output. Read the widened schema internally, then use
# ``_output_column_names`` to project batches back to ``read_type``.
@@ -460,13 +463,59 @@ class TableRead:
if (self.predicate is None or self.output_projection is None
or self.output_projection.named):
return adapter
- # Keep complete source columns for authorization and result extraction.
- # Only a predicate on a flat list-projection leaf needs an additional
- # local leaf view, even when its parent ROW is requested in full.
- projected = reader_adapter(self.read_type, self._table_read_fields(),
self.output_projection)
+ # Derive the flat working view for exact Python filtering, retaining
+ # requested whole columns and the additional top-level filter inputs.
+ columns = list(self.output_projection.columns)
+ paths = [path for _, path in columns]
+ physical_names = {field.name for field in self._table_read_fields()}
+ has_physical_path = False
+ # Predicates carrying physical paths were built against a flat view.
+ # Keep that layout stable when a projection change rebinds their
+ # aliases, including names which collide with top-level columns.
+ pending = [self.predicate]
+ while pending:
+ predicate = pending.pop()
+ if predicate.method in ('and', 'or'):
+ pending.extend(predicate.literals or [])
+ else:
+ path = predicate.field_path or ([predicate.field] if
predicate.field in physical_names else None)
+ if path:
+ has_physical_path = True
+ if len(path) == 1 and path[0] in physical_names and path
not in paths:
+ # Keep existing top-level filter-only columns in the
+ # flat working view. Projection assigns their aliases
+ # together with the user outputs, preventing
collisions.
+ columns.append((path[0], path))
+ paths.append(path)
+ projected = reader_adapter(self.read_type, self._table_read_fields(),
OutputProjection(columns))
extra_names = {field.name for field in projected[0]} - {field.name for
field in adapter[0]}
+ if has_physical_path:
+ return projected
return projected if predicate_field_names(self.predicate) &
extra_names else adapter
+ @staticmethod
+ def _bind_predicate_to_adapter(predicate, fields, paths, physical_names):
+ if predicate is None:
+ return None
+ if predicate.method in ('and', 'or'):
+ return predicate.new_literals([
+ TableRead._bind_predicate_to_adapter(child, fields, paths,
physical_names)
+ for child in predicate.literals or []])
+ path = predicate.field_path or ([predicate.field] if predicate.field
in physical_names else None)
+ if not path:
+ return predicate
+ matches = [(index, field) for index, (field, candidate_path) in
enumerate(zip(fields, paths))
+ if candidate_path == path]
+ if matches:
+ index, field = matches[0]
+ bound_path = predicate.field_path or (path if field.name !=
predicate.field else None)
+ return Predicate(predicate.method, index, field.name,
predicate.literals, bound_path)
+ if len(path) == 1 and path[0] not in {field.name for field in fields}:
+ # Existing top-level filter-only columns are appended separately.
+ return Predicate(predicate.method, predicate.index, path[0],
predicate.literals, predicate.field_path)
+ raise ValueError("Predicate field %r with path %r is not in the reader
projection"
+ % (predicate.field, path))
+
def _output_arrow_schema(self, splits=None):
schema = read_arrow_schema(self.read_type)
masks = set()
@@ -854,8 +903,17 @@ class TableRead:
def _flatten_native_nested_batch(self, batch, schema):
_, paths = self._reader_adapter()
+ # Rust already evaluated the complete predicate and projected away
+ # filter-only fields. Flatten just the user output dependencies; the
+ # Python split reader retains the additional adapter inputs internally.
+ outputs = adapter_output_paths(self.output_projection,
self._adapter_read_type, paths)
+ needed = {path[0] for _, path in outputs}
+ fields_and_paths = [(field, path) for field, path in
zip(self._adapter_read_type, paths)
+ if field.name in needed]
+ adapter_fields = [field for field, _ in fields_and_paths]
+ paths = [path for _, path in fields_and_paths]
arrays = [extract_array(batch, path) for path in paths]
- fields = list(read_arrow_schema(self._adapter_read_type))
+ fields = list(read_arrow_schema(adapter_fields))
# Authorization can mask a NOT NULL field. Keep the captured source
# nullability when flattening, including whole-parent/MAP projections.
projected = output_schema(batch.schema, OutputProjection([
diff --git a/paimon-python/pypaimon/tests/global_index_test.py
b/paimon-python/pypaimon/tests/global_index_test.py
index c07833a70f..434b0b6411 100644
--- a/paimon-python/pypaimon/tests/global_index_test.py
+++ b/paimon-python/pypaimon/tests/global_index_test.py
@@ -316,6 +316,7 @@ class GlobalIndexScalarFallbackTest(unittest.TestCase):
class _Table:
options = _Options()
+ fields = [DataField(0, 'id', AtomicType('INT'))]
predicate = Predicate(method="equal", index=0, field="id",
literals=[1])
scanner = FileScanner.__new__(FileScanner)
@@ -405,6 +406,7 @@ class GlobalIndexScalarFallbackTest(unittest.TestCase):
class _Table:
options = _Options()
+ fields = [DataField(0, 'id', AtomicType('INT'))]
scanner = FileScanner.__new__(FileScanner)
scanner.is_streaming = False
diff --git a/paimon-python/pypaimon/tests/native_nested_predicate_test.py
b/paimon-python/pypaimon/tests/native_nested_predicate_test.py
new file mode 100644
index 0000000000..983a338790
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_nested_predicate_test.py
@@ -0,0 +1,385 @@
+# 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.
+
+"""Existing projected ROW predicates must use native scans and reads."""
+
+from contextlib import ExitStack
+from datetime import datetime
+from decimal import Decimal
+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.read.native_plan import _predicate_to_native
+
+
+pytestmark = pytest.mark.native_plan
+
+
+def _table(path, mode):
+ catalog = CatalogFactory.create({'warehouse': str(path)})
+ catalog.create_database('db', True)
+ detail = pa.struct([('score', pa.int32()), ('label', pa.string()),
+ ('amount', pa.decimal128(12, 2)), ('instant',
pa.timestamp('ms'))])
+ schema = pa.schema([('id', pa.int32()), ('profile', pa.struct([('detail',
detail), ('unused', pa.string())])),
+ ('keep', pa.string())])
+ options = {'bucket': '1' if mode == 'pk' else '-1',
'write.native.enabled': 'false',
+ 'scan.native-plan.enabled': 'false', 'read.native.enabled':
'false', 'parquet.block.size': '1'}
+ catalog.create_table('db.t', Schema.from_pyarrow_schema(
+ schema, primary_keys=['id'] if mode == 'pk' else None,
options=options), False)
+ table = catalog.get_table('db.t')
+
+ def profile(score, label):
+ return {'detail': {'score': score, 'label': label, 'amount':
Decimal('1.25'),
+ 'instant': datetime(2024, 2, 29, 12)}, 'unused':
'unselected'}
+ initial = [{'id': 1, 'profile': profile(7, 'old'), 'keep': 'a'},
+ {'id': 2, 'profile': profile(1, 'first'), 'keep': 'b'},
+ {'id': 3, 'profile': None, 'keep': 'c'},
+ {'id': 4, 'profile': {'detail': None, 'unused': 'x'}, 'keep':
'd'},
+ {'id': 5, 'profile': profile(None, None), 'keep': 'e'},
+ {'id': 6, 'profile': profile(9, '中文'), 'keep': 'f'}]
+ changed = [{'id': 1, 'profile': profile(30, 'new'), 'keep': 'g'},
+ {'id': 2, 'profile': profile(7, 'winner'), 'keep': 'h'}]
+ for rows in [initial, changed]:
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(pa.Table.from_pylist(rows, schema=schema))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+ return table
+
+
+_CASES = [
+ ('score', 'equal', [7]), ('score', 'not_equal', [7]),
+ ('score', 'less_than', [7]), ('score', 'greater_or_equal', [7]),
+ ('score', 'is_null', []), ('score', 'is_not_null', []),
+ ('score', 'is_in', [[None, 7, 9]]), ('score', 'is_not_in', [[7]]),
+ ('score', 'between', [1, 9]), ('score', 'not_between', [1, 9]),
+ ('score', 'is_in', [[]]), ('score', 'is_not_in', [[]]),
+ ('score', 'is_in', [[None]]), ('score', 'is_not_in', [[None, 7]]),
+ ('score', 'between', [9, 1]), ('score', 'not_between', [9, 1]),
+ ('label', 'equal', ['中文']), ('label', 'startswith', ['w']),
+ ('label', 'contains', ['中']), ('label', 'endswith', ['文']),
+ ('label', 'like', ['_文']), ('label', 'like', ['w%']),
+ ('amount', 'equal', [Decimal('1.25')]),
+ ('instant', 'greater_or_equal', [datetime(2024, 1, 1)]),
+]
+
+
+def _read(table, native, stream, case, projection, limit, compound):
+ view = table.copy({'read.native.enabled': str(native).lower(),
+ 'scan.native-plan.enabled': str(native).lower()})
+ builder = view.new_stream_read_builder() if stream else
view.new_read_builder()
+ column, method, literals = case
+ leaf = 'profile.detail.' + column
+ builder.with_projection([leaf, 'id'])
+ predicates = builder.new_predicate_builder()
+ predicate = getattr(predicates, method)('profile_detail_' + column,
*literals)
+ wire = _predicate_to_native(predicate)
+ assert wire['field'] == 'profile'
+ assert wire['nested_path'] == ['detail', column]
+ assert _predicate_to_native(predicate.new_index(predicate.index)) == wire
+ assert _predicate_to_native(predicate.new_literals(predicate.literals)) ==
wire
+ if compound:
+ top = predicates.equal('id', 6)
+ combine = PredicateBuilder.and_predicates if compound == 'and' else
PredicateBuilder.or_predicates
+ predicate = combine([predicate, top])
+ builder.with_filter(predicate).with_projection(projection or [leaf, 'id'])
+ if limit is not None:
+ builder.with_limit(limit)
+ if stream:
+ # A snapshot plan isolates read semantics from the streaming poll loop.
+ splits = view.new_read_builder().new_scan().plan().splits()
+ else:
+ scan = builder.new_scan()
+ with ExitStack() as stack:
+ if native:
+ stack.enter_context(patch.object(scan.file_scanner, 'scan',
+
side_effect=AssertionError('Python scan fallback')))
+ splits = scan.plan().splits()
+ read = builder.new_read()
+ with ExitStack() as stack:
+ if native:
+ stack.enter_context(patch.object(read, '_create_split_read',
+
side_effect=AssertionError('Python read fallback')))
+ result = read.to_arrow_batch_reader(splits).read_all()
+ return result.sort_by('id').to_pylist()
+
+
[email protected]('mode', ['append', 'pk'])
[email protected]('stream', [False, True], ids=['batch', 'stream'])
[email protected]('case', _CASES, ids=[c[0] + '-' + c[1] for c in
_CASES])
+def test_nested_projection_filters_stay_native(tmp_path, mode, stream, case):
+ table = _table(tmp_path, mode)
+ expected = _read(table, False, stream, case, None, None, None)
+ actual = _read(table, True, stream, case, None, None, None)
+ assert actual == expected
+ if case == ('score', 'equal', [7]):
+ assert [row['id'] for row in actual] == ([2] if mode == 'pk' else [1,
2])
+ elif case == ('score', 'is_null', []):
+ assert [row['id'] for row in actual] == [3, 4, 5]
+
+
[email protected]('mode', ['append', 'pk'])
[email protected]('projection', [
+ None, ['profile', 'profile.detail.score', 'id'],
+ ['profile.detail.score', 'profile.detail.score', 'id'],
+])
[email protected]('compound', [None, 'and', 'or'])
[email protected]('limit', [None, 1])
+def test_nested_filter_composition_and_projection_changes(tmp_path, mode,
projection, compound, limit):
+ table = _table(tmp_path, mode)
+ case = ('score', 'equal', [7])
+ expected = _read(table, False, False, case, projection, limit, compound)
+ actual = _read(table, True, False, case, projection, limit, compound)
+ # LIMIT has no row-order contract. Here all matching IDs are equal or the
+ # OR includes id6; validate limited rows against the complete result.
+ if limit:
+ allowed = _read(table, False, False, case, projection, None, compound)
+ assert len(actual) == min(len(allowed), limit)
+ assert all(row in allowed for row in actual)
+ else:
+ assert actual == expected
+
+
[email protected]('mode', ['append', 'pk'])
[email protected]('native', [False, True], ids=['python', 'native'])
[email protected]('nested_first', [False, True])
[email protected]('nested_filter', [False, True])
[email protected]('partitioned', [False, True])
[email protected]('compound', [None, 'and', 'or'])
[email protected]('reorder_projection', [False, True])
+def test_nested_predicate_alias_collisions(
+ tmp_path, mode, native, nested_first, nested_filter, partitioned,
compound, reorder_projection):
+ import pickle
+
+ schema = pa.schema([('id', pa.int32()), ('a', pa.struct([('b',
pa.int32())])), ('a_b', pa.int32())])
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('db', True)
+ catalog.create_table('db.t', Schema.from_pyarrow_schema(
+ schema, primary_keys=(['a_b', 'id'] if partitioned else ['id']) if
mode == 'pk' else None,
+ partition_keys=['a_b'] if partitioned else None,
+ options={'bucket': '8' if mode == 'pk' else '-1',
'write.native.enabled': 'false',
+ 'metadata.stats-mode': 'full'}), False)
+ table = catalog.get_table('db.t')
+ for row in [
+ {'id': 11, 'a': {'b': 7}, 'a_b': 99},
+ {'id': 12, 'a': {'b': 99}, 'a_b': 7},
+ {'id': 13, 'a': None, 'a_b': 7},
+ ]:
+ writer_builder = table.new_batch_write_builder()
+ writer, commit = writer_builder.new_write(),
writer_builder.new_commit()
+ try:
+ # Separate files expose false stats pruning when the unrelated
+ # real a_b column excludes the nested literal 7.
+ writer.write_arrow(pa.Table.from_pylist([row], schema=schema))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+
+ table = table.copy({'read.native.enabled': str(native).lower(),
+ 'scan.native-plan.enabled': str(native).lower()})
+ projection = ['a.b', 'a_b', 'id'] if nested_first else ['a_b', 'a.b', 'id']
+ builder = table.new_read_builder().with_projection(projection)
+ # The second occurrence of a_b receives a suffix, whether its source is a
+ # nested leaf or a real top-level column. Predicates retain that identity.
+ alias = 'a_b' if nested_first == nested_filter else 'a_b__0'
+ predicate = builder.new_predicate_builder().equal(alias, 7)
+ predicate =
pickle.loads(pickle.dumps(predicate.new_index(predicate.index)))
+ wire = _predicate_to_native(predicate)
+ assert wire['field'] == ('a' if nested_filter else 'a_b')
+ assert wire.get('nested_path') == (['b'] if nested_filter else None)
+ if partitioned and nested_first and nested_filter:
+ with pytest.raises(ValueError, match='Partition filter must reference
only partition keys'):
+ builder.with_partition_filter(predicate)
+ if compound == 'and':
+ predicate = PredicateBuilder.and_predicates([
+ predicate, builder.new_predicate_builder().less_or_equal('id',
12)])
+ elif compound == 'or':
+ predicate = PredicateBuilder.or_predicates([
+ predicate, builder.new_predicate_builder().equal('id', 13)])
+ builder.with_filter(predicate)
+ if reorder_projection:
+ builder.with_projection([projection[1], projection[0], 'id'])
+ if mode == 'append' and partitioned and nested_filter:
+ # A nested alias may equal a partition name. Filtering after chunk
+ # generation would invalidate its declared merged row count.
+ with pytest.raises(ValueError, match='chunk_shuffle predicate must
reference only partition keys'):
+ builder.new_scan().with_chunk_shuffle(1, 2).plan()
+ scan, read = builder.new_scan(), builder.new_read()
+ with ExitStack() as stack:
+ if native:
+ stack.enter_context(patch.object(scan.file_scanner, 'scan',
+
side_effect=AssertionError('Python scan fallback')))
+ stack.enter_context(patch.object(read, '_create_split_read',
+
side_effect=AssertionError('Python read fallback')))
+ rows =
read.to_arrow_batch_reader(scan.plan().splits()).read_all().to_pylist()
+ expected = {11} if nested_filter else {12, 13}
+ if compound == 'and':
+ expected &= {11, 12}
+ elif compound == 'or':
+ expected |= {13}
+ assert sorted(row['id'] for row in rows) == sorted(expected)
+
+
[email protected]('native', [False, True], ids=['python', 'native'])
+def
test_removed_nested_filter_path_does_not_bind_to_colliding_column(tmp_path,
native):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('db', True)
+ schema = pa.schema([('a', pa.struct([('b', pa.int32())])), ('a_b',
pa.int32())])
+ catalog.create_table('db.t', Schema.from_pyarrow_schema(schema), False)
+ table = catalog.get_table('db.t').copy({'read.native.enabled':
str(native).lower()})
+ builder = table.new_read_builder().with_projection(['a.b', 'a_b'])
+ predicate = builder.new_predicate_builder().equal('a_b', 7)
+ builder.with_filter(predicate).with_projection(['a_b'])
+ with pytest.raises(ValueError, match='Predicate field .* is not in the
reader projection'):
+ builder.new_read()
+
+
[email protected]('mode', ['append', 'pk'])
[email protected]('native', [False, True], ids=['python', 'native'])
[email protected]('projected_predicate', [False, True],
ids=['physical', 'projected'])
[email protected]('compound', [None, 'and', 'or'])
[email protected]('projection', [
+ ['a.b', 'id'], ['a.b', 'a.b', 'id'], ['a', 'a.b', 'id'],
+])
+def test_nested_projection_keeps_physical_filter_only_inputs(
+ tmp_path, mode, native, projected_predicate, compound, projection):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('db', True)
+ schema = pa.schema([('id', pa.int32()), ('a', pa.struct([('b',
pa.int32())])), ('a_b', pa.int32())])
+ catalog.create_table('db.t', Schema.from_pyarrow_schema(
+ schema, primary_keys=['id'] if mode == 'pk' else None,
+ options={'bucket': '1' if mode == 'pk' else '-1',
'write.native.enabled': 'false'}), False)
+ table = catalog.get_table('db.t')
+ for rows in [
+ [{'id': 1, 'a': {'b': 30}, 'a_b': 99}],
+ [{'id': 1, 'a': {'b': 7}, 'a_b': 99}, {'id': 2, 'a': {'b': 1}, 'a_b':
7},
+ {'id': 3, 'a': {'b': 9}, 'a_b': 99}, {'id': 4, 'a': None, 'a_b': 7}],
+ ]:
+ write_builder = table.new_batch_write_builder()
+ writer, commit = write_builder.new_write(), write_builder.new_commit()
+ try:
+ writer.write_arrow(pa.Table.from_pylist(rows, schema=schema))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+ table = table.copy({'read.native.enabled': str(native).lower(),
+ 'scan.native-plan.enabled': str(native).lower()})
+ builder = table.new_read_builder().with_projection(['a_b', 'id'])
+ predicates = builder.new_predicate_builder() if projected_predicate else
PredicateBuilder(table.fields)
+ physical = predicates.equal('a_b', 7)
+ assert physical.field_path == (['a_b'] if projected_predicate else None)
+ builder.with_projection(projection)
+ if compound:
+ nested_builder = builder.new_predicate_builder()
+ nested = (nested_builder.less_than('a_b', 10) if compound == 'and'
+ else nested_builder.equal('a_b', 7))
+ combine = PredicateBuilder.and_predicates if compound == 'and' else
PredicateBuilder.or_predicates
+ predicate = combine([nested, physical])
+ else:
+ predicate = physical
+ builder.with_filter(predicate)
+ scan, read = builder.new_scan(), builder.new_read()
+ schema_names = [field.name for field in builder._output_fields()]
+ with ExitStack() as stack:
+ if native:
+ stack.enter_context(patch.object(scan.file_scanner, 'scan',
+
side_effect=AssertionError('Python scan fallback')))
+ stack.enter_context(patch.object(read, '_create_split_read',
+
side_effect=AssertionError('Python read fallback')))
+ splits = scan.plan().splits()
+ for result in (read.to_arrow(splits),
read.to_arrow_batch_reader(splits).read_all()):
+ assert result.schema.names == schema_names
+ assert sorted(result.column('id').to_pylist()) == (
+ [2] if compound == 'and' else [1, 2, 4] if compound == 'or'
else [2, 4])
+ # Iterator reads exercise the same exact filter on Python record readers,
+ # including merged PK records and predicate-only alias collisions.
+ ids = [row.get_field(len(projection) - 1) for row in
read.to_iterator(splits)]
+ assert sorted(ids) == (
+ [2] if compound == 'and' else [1, 2, 4] if compound == 'or' else [2,
4])
+
+
[email protected]('native', [False, True], ids=['python', 'native'])
+def test_nested_projection_keeps_synthesized_row_id_filter_input(tmp_path,
native):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('db', True)
+ schema = pa.schema([('id', pa.int32()), ('_ROW', pa.struct([('ID',
pa.int64())]))])
+ catalog.create_table('db.t', Schema.from_pyarrow_schema(
+ schema, options={'bucket': '-1', 'row-tracking.enabled': 'true',
'write.native.enabled': 'false'}), False)
+ table = catalog.get_table('db.t')
+ write_builder = table.new_batch_write_builder()
+ writer, commit = write_builder.new_write(), write_builder.new_commit()
+ try:
+ writer.write_arrow(pa.Table.from_pylist([
+ {'id': 1, '_ROW': {'ID': 100}}, {'id': 2, '_ROW': {'ID': 0}}],
schema=schema))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+ table = table.copy({'read.native.enabled': str(native).lower(),
+ 'scan.native-plan.enabled': str(native).lower()})
+ builder = table.new_read_builder().with_projection(['_ROW_ID', 'id'])
+ predicate = builder.new_predicate_builder().equal('_ROW_ID', 0)
+ builder.with_filter(predicate).with_projection(['_ROW.ID', 'id'])
+ read, scan = builder.new_read(), builder.new_scan()
+ with ExitStack() as stack:
+ if native:
+ stack.enter_context(patch.object(scan.file_scanner, 'scan',
+
side_effect=AssertionError('Python scan fallback')))
+ stack.enter_context(patch.object(read, '_create_split_read',
+
side_effect=AssertionError('Python read fallback')))
+ splits = scan.plan().splits()
+ for result in (read.to_arrow(splits),
read.to_arrow_batch_reader(splits).read_all()):
+ assert result.to_pylist() == [{'_ROW_ID': 100, 'id': 1}]
+ assert [row.get_field(1) for row in read.to_iterator(splits)] == [1]
+
+
[email protected]('mode', ['append', 'pk'])
[email protected]('evolution', ['promote', 'rename', 'replace'])
[email protected]('projection', [None, 'full'])
+def test_nested_predicate_schema_evolution(tmp_path, mode, evolution,
projection):
+ from pypaimon.schema.data_types import AtomicType
+ from pypaimon.schema.schema_change import SchemaChange
+
+ table = _table(tmp_path, mode)
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ path = ['profile', 'detail', 'score']
+ if evolution == 'promote':
+ changes = [SchemaChange.update_column_type(path, AtomicType('BIGINT'))]
+ case = ('score', 'equal', [7])
+ elif evolution == 'rename':
+ changes = [SchemaChange.rename_column(path, 'renamed')]
+ case = ('renamed', 'equal', [7])
+ else:
+ changes = [SchemaChange.drop_column(path),
SchemaChange.add_column(path, AtomicType('INT'))]
+ case = ('score', 'is_null', [])
+ catalog.alter_table('db.t', changes, False)
+ table = catalog.get_table('db.t')
+ if projection == 'full':
+ projection = ['profile', 'profile.detail.' + case[0], 'id']
+ expected = _read(table, False, False, case, projection, None, None)
+ actual = _read(table, True, False, case, projection, None, None)
+ assert actual == expected
