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 8033b47633 [python] Fix filters on non-projected columns (#8509)
8033b47633 is described below
commit 8033b476330cf04a951be30d34b602c3760e0c99
Author: Dapeng Sun(孙大鹏) <[email protected]>
AuthorDate: Thu Jul 9 12:04:41 2026 +0800
[python] Fix filters on non-projected columns (#8509)
---
paimon-python/pypaimon/read/push_down_utils.py | 34 ++++-
.../read/reader/blob_descriptor_convert_reader.py | 3 +-
.../pypaimon/read/reader/data_file_batch_reader.py | 4 +-
.../pypaimon/read/reader/field_indices.py | 48 +++++++
.../read/reader/filter_record_batch_reader.py | 34 ++++-
.../read/reader/iface/record_batch_reader.py | 23 +++-
.../pypaimon/read/reader/limited_record_reader.py | 4 +-
.../read/reader/nested_leaf_batch_reader.py | 4 +
.../read/reader/outer_projection_record_reader.py | 19 +--
.../read/reader/row_range_filter_record_reader.py | 3 +-
.../pypaimon/read/reader/shard_batch_reader.py | 1 +
paimon-python/pypaimon/read/split_read.py | 120 +++++++++--------
paimon-python/pypaimon/read/table_read.py | 75 ++++++++++-
.../pypaimon/tests/data_evolution_test.py | 9 +-
.../tests/projection_predicate_index_test.py | 145 +++++++++++++++++----
.../pypaimon/tests/reader_parallel_test.py | 15 +++
16 files changed, 413 insertions(+), 128 deletions(-)
diff --git a/paimon-python/pypaimon/read/push_down_utils.py
b/paimon-python/pypaimon/read/push_down_utils.py
index 9c12872d9f..ebf4da3be5 100644
--- a/paimon-python/pypaimon/read/push_down_utils.py
+++ b/paimon-python/pypaimon/read/push_down_utils.py
@@ -21,6 +21,13 @@ from pypaimon.common.predicate import Predicate
from pypaimon.common.predicate_builder import PredicateBuilder
from pypaimon.schema.data_types import DataField
+_UNSAFE_ARROW_FILTER_METHODS = frozenset([
+ 'startsWith',
+ 'endsWith',
+ 'contains',
+ 'like',
+])
+
def extract_partition_spec_from_predicate(
predicate: Predicate, partition_keys: List[str]
@@ -117,16 +124,39 @@ def _change_index(input_predicate: Predicate, mapping:
Dict[int, int]):
return input_predicate.new_index(mapping[input_predicate.index])
-def _get_all_fields(predicate: Predicate) -> Set[str]:
+def predicate_field_names(predicate: Predicate) -> Set[str]:
+ """Return all column names referenced by predicate leaves."""
if predicate.field is not None:
return {predicate.field}
involved_fields = set()
if predicate.literals:
for sub_predicate in predicate.literals:
- involved_fields.update(_get_all_fields(sub_predicate))
+ involved_fields.update(predicate_field_names(sub_predicate))
return involved_fields
+def _get_all_fields(predicate: Predicate) -> Set[str]:
+ return predicate_field_names(predicate)
+
+
+def predicate_supports_arrow_filter(predicate: Optional[Predicate]) -> bool:
+ """Whether ``predicate.to_arrow()`` is safe for batch filtering.
+
+ PyArrow 6 accepts dataset expressions for comparisons, null checks, and
+ isin, but string match compute functions do not accept dataset expressions.
+ Predicate.to_arrow() currently falls back to a truthy expression or None
for
+ those methods, which is safe for file pruning but not for final row
filters.
+ """
+ if predicate is None:
+ return True
+ if predicate.method == 'and' or predicate.method == 'or':
+ return all(
+ predicate_supports_arrow_filter(p)
+ for p in (predicate.literals or [])
+ )
+ return predicate.method not in _UNSAFE_ARROW_FILTER_METHODS
+
+
def remove_row_id_filter(predicate: Predicate) -> Optional[Predicate]:
from pypaimon.table.special_fields import SpecialFields
diff --git
a/paimon-python/pypaimon/read/reader/blob_descriptor_convert_reader.py
b/paimon-python/pypaimon/read/reader/blob_descriptor_convert_reader.py
index f6c5f73cfe..48cee27988 100644
--- a/paimon-python/pypaimon/read/reader/blob_descriptor_convert_reader.py
+++ b/paimon-python/pypaimon/read/reader/blob_descriptor_convert_reader.py
@@ -59,8 +59,7 @@ class BlobInlineConvertReader(RecordBatchReader):
self._table = table
self._prescan_reader_factory = prescan_reader_factory
self._blob_parallelism = blob_parallelism
- self.file_io = inner.file_io
- self.blob_field_indices = inner.blob_field_indices
+ self._adopt_metadata(inner)
# Preserve original BlobViewStruct bytes when resolve disabled: skip
both
# view resolution (Stage 1) and descriptor-to-data resolution (Stage
2).
resolve_enabled = CoreOptions.blob_view_resolve_enabled(
diff --git a/paimon-python/pypaimon/read/reader/data_file_batch_reader.py
b/paimon-python/pypaimon/read/reader/data_file_batch_reader.py
index 0183396fe7..1de9f7a405 100644
--- a/paimon-python/pypaimon/read/reader/data_file_batch_reader.py
+++ b/paimon-python/pypaimon/read/reader/data_file_batch_reader.py
@@ -344,7 +344,9 @@ class DataFileBatchReader(RecordBatchReader):
# Handle _ROW_ID field
if SpecialFields.ROW_ID.name in self.system_fields.keys():
idx = self.system_fields[SpecialFields.ROW_ID.name]
- if self.row_id_offsets is not None:
+ if self.first_row_id is None:
+ arrays[idx] = pa.nulls(record_batch.num_rows, type=pa.int64())
+ elif self.row_id_offsets is not None:
end = self._row_id_cursor + record_batch.num_rows
row_ids = [self.first_row_id + o for o in
self.row_id_offsets[self._row_id_cursor:end]]
arrays[idx] = pa.array(row_ids, type=pa.int64())
diff --git a/paimon-python/pypaimon/read/reader/field_indices.py
b/paimon-python/pypaimon/read/reader/field_indices.py
new file mode 100644
index 0000000000..02060a2f51
--- /dev/null
+++ b/paimon-python/pypaimon/read/reader/field_indices.py
@@ -0,0 +1,48 @@
+# 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.
+
+"""Helpers for carrying typed field metadata through projection wrappers."""
+
+from typing import Any, Iterable, List, Optional, Sequence, Set
+
+from pypaimon.schema.data_types import DataField, VectorType
+
+
+def blob_field_indices(fields: List[DataField]) -> Set[int]:
+ return {
+ i for i, f in enumerate(fields)
+ if hasattr(f.type, 'type') and f.type.type == 'BLOB'
+ }
+
+
+def vector_field_indices(fields: List[DataField]) -> Set[int]:
+ return {i for i, f in enumerate(fields) if isinstance(f.type, VectorType)}
+
+
+def project_top_level_field_indices(
+ source_indices: Optional[Iterable[int]],
+ path_specs: Sequence[Any],
+) -> Optional[Set[int]]:
+ """Map top-level field indices through path specs with
top_idx/sub_names."""
+ if source_indices is None:
+ return None
+ source = set(source_indices)
+ return {
+ proj_pos
+ for proj_pos, spec in enumerate(path_specs)
+ if not spec.sub_names and spec.top_idx in source
+ }
diff --git a/paimon-python/pypaimon/read/reader/filter_record_batch_reader.py
b/paimon-python/pypaimon/read/reader/filter_record_batch_reader.py
index a6039c46a8..fdebbc99cb 100644
--- a/paimon-python/pypaimon/read/reader/filter_record_batch_reader.py
+++ b/paimon-python/pypaimon/read/reader/filter_record_batch_reader.py
@@ -20,9 +20,14 @@ from typing import List, Optional
import pyarrow as pa
import pyarrow.dataset as ds
+import polars
from pypaimon.common.predicate import Predicate
-from pypaimon.read.reader.iface.record_batch_reader import RecordBatchReader
+from pypaimon.read.push_down_utils import predicate_supports_arrow_filter
+from pypaimon.read.reader.iface.record_batch_reader import (
+ InternalRowWrapperIterator,
+ RecordBatchReader,
+)
from pypaimon.schema.data_types import DataField
logger = logging.getLogger(__name__)
@@ -46,8 +51,8 @@ class FilterRecordBatchReader(RecordBatchReader):
self.predicate = predicate
self.field_names = field_names
self.schema_fields = schema_fields
- self.file_io = reader.file_io
- self.blob_field_indices = reader.blob_field_indices
+ self._use_arrow_filter = predicate_supports_arrow_filter(predicate)
+ self._adopt_metadata(reader)
def read_arrow_batch(self) -> Optional[pa.RecordBatch]:
while True:
@@ -62,7 +67,11 @@ class FilterRecordBatchReader(RecordBatchReader):
continue
def _filter_batch(self, batch: pa.RecordBatch) -> Optional[pa.RecordBatch]:
+ if not self._use_arrow_filter:
+ return self._filter_batch_by_row(batch)
expr = self.predicate.to_arrow()
+ if expr is None:
+ return self._filter_batch_by_row(batch)
result = ds.InMemoryDataset(pa.Table.from_batches([batch])).scanner(
filter=expr
).to_table()
@@ -81,6 +90,25 @@ class FilterRecordBatchReader(RecordBatchReader):
schema=result.schema,
)
+ def _filter_batch_by_row(self, batch: pa.RecordBatch) ->
Optional[pa.RecordBatch]:
+ df = polars.from_arrow(pa.Table.from_batches([batch]))
+ iterator = InternalRowWrapperIterator(
+ self._iter_df_rows(df),
+ df.width,
+ self.file_io,
+ self.blob_field_indices,
+ self.vector_field_indices,
+ )
+ selected = []
+ pos = 0
+ for row in iter(iterator.next, None):
+ if self.predicate.test(row):
+ selected.append(pos)
+ pos += 1
+ if not selected:
+ return None
+ return batch.take(pa.array(selected, type=pa.int64()))
+
def return_batch_pos(self) -> int:
pos = getattr(self.reader, 'return_batch_pos', lambda: 0)()
return pos if pos is not None else 0
diff --git a/paimon-python/pypaimon/read/reader/iface/record_batch_reader.py
b/paimon-python/pypaimon/read/reader/iface/record_batch_reader.py
index 6cd45bf63d..7888f2fab5 100644
--- a/paimon-python/pypaimon/read/reader/iface/record_batch_reader.py
+++ b/paimon-python/pypaimon/read/reader/iface/record_batch_reader.py
@@ -19,7 +19,7 @@ from abc import abstractmethod
from typing import Iterator, Optional, TypeVar
import polars
-from pyarrow import RecordBatch
+from pyarrow import RecordBatch, Table
from pypaimon.read.reader.iface.record_iterator import RecordIterator
from pypaimon.read.reader.iface.record_reader import RecordReader
@@ -38,6 +38,11 @@ class RecordBatchReader(RecordReader):
blob_field_indices = None
vector_field_indices = None
+ def _adopt_metadata(self, reader: "RecordBatchReader") -> None:
+ self.file_io = reader.file_io
+ self.blob_field_indices = reader.blob_field_indices
+ self.vector_field_indices = reader.vector_field_indices
+
@abstractmethod
def read_arrow_batch(self) -> Optional[RecordBatch]:
"""
@@ -53,22 +58,31 @@ class RecordBatchReader(RecordReader):
arrow_batch = self.read_arrow_batch()
if arrow_batch is None:
return None
- return polars.from_arrow(arrow_batch)
+ # Older Polars versions accept Arrow Table but not RecordBatch.
+ return polars.from_arrow(Table.from_batches([arrow_batch]))
def tuple_iterator(self) -> Optional[Iterator[tuple]]:
df = self.read_next_df()
if df is None:
return None
- return df.iter_rows()
+ return self._iter_df_rows(df)
def read_batch(self) -> Optional[RecordIterator[InternalRow]]:
df = self.read_next_df()
if df is None:
return None
return InternalRowWrapperIterator(
- df.iter_rows(), df.width, self.file_io,
+ self._iter_df_rows(df), df.width, self.file_io,
self.blob_field_indices, self.vector_field_indices)
+ @staticmethod
+ def _iter_df_rows(df) -> Iterator[tuple]:
+ iter_rows = getattr(df, "iter_rows", None)
+ if iter_rows is not None:
+ return iter_rows()
+ # Older Polars exposes eager row iteration as rows().
+ return iter(df.rows())
+
class InternalRowWrapperIterator(RecordIterator[InternalRow]):
def __init__(self, iterator: Iterator[tuple], width: int,
@@ -94,6 +108,7 @@ class RowPositionReader(RecordBatchReader):
self._data_reader = data_reader
self._row_iterator = RowPositionRecordIterator()
self.batch_pos = 0
+ self._adopt_metadata(data_reader)
def read_arrow_batch(self) -> Optional[RecordBatch]:
batch = self._data_reader.read_arrow_batch()
diff --git a/paimon-python/pypaimon/read/reader/limited_record_reader.py
b/paimon-python/pypaimon/read/reader/limited_record_reader.py
index f78221d3f8..a4eab01986 100644
--- a/paimon-python/pypaimon/read/reader/limited_record_reader.py
+++ b/paimon-python/pypaimon/read/reader/limited_record_reader.py
@@ -88,9 +88,7 @@ class LimitedRecordBatchReader(RecordBatchReader):
self._inner = inner
self._limit = limit
self.count = 0
- self.file_io = inner.file_io
- self.blob_field_indices = inner.blob_field_indices
- self.vector_field_indices = inner.vector_field_indices
+ self._adopt_metadata(inner)
def read_arrow_batch(self) -> Optional[RecordBatch]:
if self.count >= self._limit:
diff --git a/paimon-python/pypaimon/read/reader/nested_leaf_batch_reader.py
b/paimon-python/pypaimon/read/reader/nested_leaf_batch_reader.py
index 8f042a9229..cd849454a7 100644
--- a/paimon-python/pypaimon/read/reader/nested_leaf_batch_reader.py
+++ b/paimon-python/pypaimon/read/reader/nested_leaf_batch_reader.py
@@ -21,6 +21,7 @@ import pyarrow as pa
import pyarrow.compute as pc
from pyarrow import RecordBatch
+from pypaimon.read.reader.field_indices import blob_field_indices,
vector_field_indices
from pypaimon.read.reader.iface.record_batch_reader import RecordBatchReader
from pypaimon.schema.data_types import DataField, PyarrowFieldParser
@@ -44,6 +45,9 @@ class NestedLeafBatchReader(RecordBatchReader):
self._inner = inner
self._paths = name_paths
self._schema = PyarrowFieldParser.from_paimon_schema(output_fields)
+ self.file_io = inner.file_io
+ self.blob_field_indices = blob_field_indices(output_fields)
+ self.vector_field_indices = vector_field_indices(output_fields)
def read_arrow_batch(self) -> Optional[RecordBatch]:
batch = self._inner.read_arrow_batch()
diff --git
a/paimon-python/pypaimon/read/reader/outer_projection_record_reader.py
b/paimon-python/pypaimon/read/reader/outer_projection_record_reader.py
index 124a957183..e8bb475097 100644
--- a/paimon-python/pypaimon/read/reader/outer_projection_record_reader.py
+++ b/paimon-python/pypaimon/read/reader/outer_projection_record_reader.py
@@ -27,6 +27,7 @@ wrapper extracts the user-visible flat columns afterwards.
from typing import Any, List, Optional
+from pypaimon.read.reader.field_indices import project_top_level_field_indices
from pypaimon.read.reader.iface.record_iterator import RecordIterator
from pypaimon.read.reader.iface.record_reader import RecordReader
from pypaimon.table.row.internal_row import InternalRow
@@ -62,20 +63,10 @@ class
OuterProjectionRecordReader(RecordReader[InternalRow]):
self._inner = inner
self._flat_arity = len(name_paths)
self._file_io = file_io
- self._blob_field_indices = None
- if blob_field_indices is not None:
- self._blob_field_indices = {
- proj_pos
- for proj_pos, spec in enumerate(self._specs)
- if not spec.sub_names and spec.top_idx in blob_field_indices
- }
- self._vector_field_indices = None
- if vector_field_indices is not None:
- self._vector_field_indices = {
- proj_pos
- for proj_pos, spec in enumerate(self._specs)
- if not spec.sub_names and spec.top_idx in vector_field_indices
- }
+ self._blob_field_indices = project_top_level_field_indices(
+ blob_field_indices, self._specs)
+ self._vector_field_indices = project_top_level_field_indices(
+ vector_field_indices, self._specs)
def read_batch(self) -> Optional[RecordIterator[InternalRow]]:
inner_batch = self._inner.read_batch()
diff --git
a/paimon-python/pypaimon/read/reader/row_range_filter_record_reader.py
b/paimon-python/pypaimon/read/reader/row_range_filter_record_reader.py
index 5f97fb4ad5..d25434f396 100644
--- a/paimon-python/pypaimon/read/reader/row_range_filter_record_reader.py
+++ b/paimon-python/pypaimon/read/reader/row_range_filter_record_reader.py
@@ -32,8 +32,7 @@ class RowIdFilterRecordBatchReader(RecordBatchReader):
self.reader = reader
self.current_row_id = first_row_id
self.row_id_ranges = row_id_ranges
- self.file_io = reader.file_io
- self.blob_field_indices = reader.blob_field_indices
+ self._adopt_metadata(reader)
def read_arrow_batch(self) -> Optional[RecordBatch]:
while True:
diff --git a/paimon-python/pypaimon/read/reader/shard_batch_reader.py
b/paimon-python/pypaimon/read/reader/shard_batch_reader.py
index 3d5cf77f68..aa00814b82 100644
--- a/paimon-python/pypaimon/read/reader/shard_batch_reader.py
+++ b/paimon-python/pypaimon/read/reader/shard_batch_reader.py
@@ -31,6 +31,7 @@ class ShardBatchReader(RecordBatchReader):
self.start_pos = start_pos
self.end_pos = end_pos
self.current_pos = 0
+ self._adopt_metadata(reader)
def read_arrow_batch(self) -> Optional[RecordBatch]:
# Check if reader is FormatBlobReader (blob type)
diff --git a/paimon-python/pypaimon/read/split_read.py
b/paimon-python/pypaimon/read/split_read.py
index 9167b7824a..d230975d5c 100644
--- a/paimon-python/pypaimon/read/split_read.py
+++ b/paimon-python/pypaimon/read/split_read.py
@@ -32,7 +32,11 @@ from pypaimon.globalindex import Range
from pypaimon.manifest.schema.data_file_meta import DataFileMeta
from pypaimon.read.interval_partition import IntervalPartition, SortedRun
from pypaimon.read.partition_info import PartitionInfo
-from pypaimon.read.push_down_utils import rewrite_predicate_indices,
trim_predicate_by_fields
+from pypaimon.read.push_down_utils import (
+ predicate_field_names,
+ rewrite_predicate_indices,
+ trim_predicate_by_fields,
+)
from pypaimon.read.reader.concat_batch_reader import (
BlobFallbackBatchReader, ConcatBatchReader,
MergeAllBatchReader, DataEvolutionMergeReader)
@@ -42,6 +46,8 @@ from pypaimon.read.reader.data_file_batch_reader import
DataFileBatchReader
from pypaimon.read.reader.drop_delete_reader import DropDeleteRecordReader
from pypaimon.read.reader.empty_record_reader import EmptyFileRecordReader
from pypaimon.read.reader.field_bunch import BlobBunch, DataBunch, FieldBunch,
VectorBunch
+from pypaimon.read.reader.field_indices import (
+ blob_field_indices, vector_field_indices)
from pypaimon.read.reader.filter_record_reader import FilterRecordReader
from pypaimon.read.reader.format_avro_reader import FormatAvroReader
from pypaimon.read.reader.blob_descriptor_convert_reader import
BlobInlineConvertReader
@@ -65,7 +71,6 @@ from pypaimon.read.reader.aggregation_merge_function import (
AggregateMergeFunction, build_field_aggregators)
from pypaimon.read.reader.sort_merge_reader import (SortMergeReaderWithMinHeap,
builtin_seq_comparator)
-from pypaimon.read.push_down_utils import _get_all_fields
from pypaimon.read.split import Split
from pypaimon.read.sliced_split import SlicedSplit
from pypaimon.schema.data_types import DataField, PyarrowFieldParser
@@ -81,16 +86,6 @@ ROW_SIDECAR_FORMAT = CoreOptions.FILE_FORMAT_ROW
_COMPRESS_EXTENSIONS = frozenset(['gz', 'bz2', 'deflate', 'snappy', 'lz4',
'zst'])
-def _blob_field_indices(fields: List[DataField]) -> set:
- return {i for i, f in enumerate(fields)
- if hasattr(f.type, 'type') and f.type.type == 'BLOB'}
-
-
-def _vector_field_indices(fields: List[DataField]) -> set:
- from pypaimon.schema.data_types import VectorType
- return {i for i, f in enumerate(fields) if isinstance(f.type, VectorType)}
-
-
def format_identifier(file_name):
idx = file_name.rfind('.')
assert idx != -1, "%s is not a legal file name." % file_name
@@ -151,7 +146,7 @@ class SplitRead(ABC):
read_type_names = {f.name for f in read_type}
if (
self.predicate is not None
- and _get_all_fields(self.predicate).issubset(read_type_names)
+ and
predicate_field_names(self.predicate).issubset(read_type_names)
):
self.predicate_for_reader = rewrite_predicate_indices(
self.predicate, read_type
@@ -309,7 +304,9 @@ class SplitRead(ABC):
raise NotImplementedError(
"Nested-field projection is not supported on Vortex files")
ordered_read_fields = [name_to_field[n] for n in read_file_fields
if n in name_to_field]
- predicate_fields = _get_all_fields(self.push_down_predicate) if
self.push_down_predicate else set()
+ predicate_fields = (
+ predicate_field_names(self.push_down_predicate)
+ if self.push_down_predicate else set())
format_reader = FormatVortexReader(self.table.file_io, file_path,
ordered_read_fields,
read_arrow_predicate,
batch_size=batch_size,
row_indices=row_indices,
@@ -787,50 +784,35 @@ class RawFileSplitRead(SplitRead):
concat_reader = ConcatBatchReader(
data_readers, file_io=self.table.file_io,
- blob_field_indices=_blob_field_indices(self.read_fields),
- vector_field_indices=_vector_field_indices(self.read_fields))
- # if the table is appendonly table, we don't need extra filter, all
predicates has pushed down
+ blob_field_indices=blob_field_indices(self.read_fields),
+ vector_field_indices=vector_field_indices(self.read_fields))
+ reader = concat_reader
if self.table.is_primary_key_table and self.predicate_for_reader:
- reader = FilterRecordReader(concat_reader,
self.predicate_for_reader)
- if self.outer_extract_name_paths:
- # Row-level extraction: the filter evaluates rows in the
- # widened top-level coordinate space, so extract after it.
- from pypaimon.read.reader.outer_projection_record_reader
import \
- OuterProjectionRecordReader
- reader = OuterProjectionRecordReader(
- reader, [f.name for f in self.read_fields],
- self.outer_extract_name_paths,
- file_io=self.table.file_io,
- blob_field_indices=_blob_field_indices(self.read_fields),
-
vector_field_indices=_vector_field_indices(self.read_fields))
- if self.limit is not None:
- reader = LimitedRecordReader(reader, self.limit)
- else:
- reader = concat_reader
- if self.outer_extract_name_paths:
- from pypaimon.read.reader.nested_leaf_batch_reader import \
- NestedLeafBatchReader
- reader = NestedLeafBatchReader(
- reader, self.outer_extract_name_paths,
- self.outer_flat_read_type)
- # A predicate on a projected nested leaf cannot be pushed down:
- # its leaf path is absent from the widened top-level read
- # fields, so SplitRead.__init__ dropped it
(predicate_for_reader
- # is None). Without re-applying it the filter is silently lost
- # and every row is returned. Re-evaluate it on the extracted
- # flat batches, whose column names match the predicate fields;
- # trim to the projected columns so a filter on a non-projected
- # column keeps the existing "dropped" semantics rather than
- # referencing a missing column.
- if self.predicate is not None and self.predicate_for_reader is
None:
- flat_names = [f.name for f in self.outer_flat_read_type]
- trimmed = trim_predicate_by_fields(self.predicate,
flat_names)
- if trimmed is not None:
- from pypaimon.read.reader.filter_record_batch_reader \
- import FilterRecordBatchReader
- reader = FilterRecordBatchReader(reader, trimmed)
- if self.limit is not None:
- reader = LimitedRecordBatchReader(reader, self.limit)
+ reader = FilterRecordBatchReader(
+ reader,
+ self.predicate_for_reader,
+ field_names=[f.name for f in self.read_fields],
+ schema_fields=self.read_fields,
+ )
+ if self.outer_extract_name_paths:
+ from pypaimon.read.reader.nested_leaf_batch_reader import \
+ NestedLeafBatchReader
+ reader = NestedLeafBatchReader(
+ reader, self.outer_extract_name_paths,
+ self.outer_flat_read_type)
+ # A predicate on a projected nested leaf cannot be pushed down:
+ # its leaf path is absent from the widened top-level read fields,
+ # so SplitRead.__init__ dropped it (predicate_for_reader is None).
+ # Without re-applying it the filter is silently lost and every row
+ # is returned. Re-evaluate it on the extracted flat batches, whose
+ # column names match the predicate fields.
+ if self.predicate is not None and self.predicate_for_reader is
None:
+ flat_names = [f.name for f in self.outer_flat_read_type]
+ trimmed = trim_predicate_by_fields(self.predicate, flat_names)
+ if trimmed is not None:
+ reader = FilterRecordBatchReader(reader, trimmed)
+ if self.limit is not None:
+ reader = LimitedRecordBatchReader(reader, self.limit)
return reader
def _all_data_fields_from(self, fields):
@@ -960,8 +942,8 @@ class MergeFileSplitRead(SplitRead):
reader, [f.name for f in inner_value_fields],
self.outer_extract_name_paths,
file_io=self.table.file_io,
- blob_field_indices=_blob_field_indices(inner_value_fields),
- vector_field_indices=_vector_field_indices(inner_value_fields))
+ blob_field_indices=blob_field_indices(inner_value_fields),
+ vector_field_indices=vector_field_indices(inner_value_fields))
# A predicate on a projected nested leaf is not pushed down (its
leaf
# path is absent from the widened-to-full-ROW read fields, so it
was
# dropped in __init__). Without re-applying it after extraction the
@@ -995,7 +977,9 @@ class DataEvolutionSplitRead(SplitRead):
split: Split,
row_tracking_enabled: bool,
nested_name_paths: Optional[List[List[str]]] = None,
- limit: Optional[int] = None):
+ limit: Optional[int] = None,
+ outer_extract_name_paths: Optional[List[List[str]]] = None,
+ outer_flat_read_type: Optional[List[DataField]] = None):
self.row_ranges = None
actual_split = split
if isinstance(split, IndexedSplit):
@@ -1006,6 +990,8 @@ class DataEvolutionSplitRead(SplitRead):
nested_name_paths=nested_name_paths,
limit=limit,
)
+ self.outer_extract_name_paths = outer_extract_name_paths
+ self.outer_flat_read_type = outer_flat_read_type
def _push_down_predicate(self) -> Optional[Predicate]:
# Data evolution: files may have different schemas, so we don't push
predicate
@@ -1051,8 +1037,8 @@ class DataEvolutionSplitRead(SplitRead):
merge_reader = ConcatBatchReader(
suppliers, file_io=self.table.file_io,
- blob_field_indices=_blob_field_indices(self.read_fields),
- vector_field_indices=_vector_field_indices(self.read_fields))
+ blob_field_indices=blob_field_indices(self.read_fields),
+ vector_field_indices=vector_field_indices(self.read_fields))
if self.predicate_for_reader is not None:
reader = FilterRecordBatchReader(
merge_reader,
@@ -1063,6 +1049,16 @@ class DataEvolutionSplitRead(SplitRead):
else:
reader = merge_reader
+ if self.outer_extract_name_paths:
+ if self.outer_flat_read_type is None:
+ raise ValueError(
+ "outer_flat_read_type is required when
outer_extract_name_paths "
+ "is set")
+ from pypaimon.read.reader.nested_leaf_batch_reader import \
+ NestedLeafBatchReader
+ reader = NestedLeafBatchReader(
+ reader, self.outer_extract_name_paths,
self.outer_flat_read_type)
+
if self.limit is not None:
reader = LimitedRecordBatchReader(reader, self.limit)
diff --git a/paimon-python/pypaimon/read/table_read.py
b/paimon-python/pypaimon/read/table_read.py
index b19fff109d..7c4a98f3d1 100644
--- a/paimon-python/pypaimon/read/table_read.py
+++ b/paimon-python/pypaimon/read/table_read.py
@@ -23,6 +23,7 @@ import pandas
import pyarrow
from pypaimon.common.predicate import Predicate
+from pypaimon.read.push_down_utils import predicate_field_names
from pypaimon.read.reader.iface.record_batch_reader import RecordBatchReader
from pypaimon.read.split import Split
from pypaimon.read.split_read import (DataEvolutionSplitRead,
@@ -99,6 +100,12 @@ class TableRead:
self.table: FileStoreTable = table
self.predicate = predicate
self.read_type = read_type
+ # 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``.
+ self._predicate_extra_fields =
self._predicate_fields_outside_read_type()
+ self._scan_read_type = self.read_type + self._predicate_extra_fields
+ self._output_column_names = [f.name for f in self.read_type]
self.include_row_kind = include_row_kind
self.nested_name_paths = nested_name_paths
self.limit = limit
@@ -222,6 +229,7 @@ class TableRead:
for batch in iter(reader.read_arrow_batch, None):
if remaining is not None and batch.num_rows >
remaining:
batch = batch.slice(0, remaining)
+ batch = self._project_batch_to_output(batch)
if self.include_row_kind:
batch = self._add_row_kind_column_to_batch(batch,
"+I")
yield batch
@@ -388,6 +396,7 @@ class TableRead:
break
if allowed < batch.num_rows:
batch = batch.slice(0, allowed)
+ batch = self._project_batch_to_output(batch)
if self.include_row_kind:
batch = self._add_row_kind_column_to_batch(batch, "+I")
out.append(batch)
@@ -627,13 +636,14 @@ class TableRead:
def _build_split_read(self, split: Split) -> SplitRead:
if self.table.is_primary_key_table and not split.raw_convertible:
- inner_read_type = self.read_type
+ inner_read_type = self._scan_read_type
outer_extract_name_paths: Optional[List[List[str]]] = None
if self.nested_name_paths and any(
len(p) > 1 for p in self.nested_name_paths):
# Inner: full ROW for the merge function. Outer: extract
# the requested sub-paths back to the user's flat schema.
- inner_read_type = self._widen_to_top_level_for_merge()
+ inner_read_type = self._with_predicate_extra_fields(
+ self._widen_to_top_level_for_merge())
outer_extract_name_paths = self.nested_name_paths
# When the user's projection drops a ``sequence.field``, the merge
@@ -661,6 +671,10 @@ class TableRead:
# user's requested (flat) columns in order.
outer_extract_name_paths = [
[f.name] for f in self.read_type]
+ if outer_extract_name_paths is None and
self._needs_output_projection():
+ # Split readers own the output projection for iterator reads;
+ # the TableRead Arrow projection below is a final guard.
+ outer_extract_name_paths = self._output_extract_name_paths()
return MergeFileSplitRead(
table=self.table,
predicate=self.predicate,
@@ -678,17 +692,24 @@ class TableRead:
raise NotImplementedError(
"Nested-field projection on data-evolution tables is "
"not yet supported")
+ outer_extract_name_paths = None
+ if self._needs_output_projection():
+ # Keep iterator output narrow inside DataEvolutionSplitRead.
+ outer_extract_name_paths = self._output_extract_name_paths()
return DataEvolutionSplitRead(
table=self.table,
predicate=self.predicate,
- read_type=self.read_type,
+ read_type=self._scan_read_type,
split=split,
row_tracking_enabled=True,
nested_name_paths=self.nested_name_paths,
+ outer_extract_name_paths=outer_extract_name_paths,
+ outer_flat_read_type=(
+ self.read_type if outer_extract_name_paths else None),
limit=self.limit,
)
else:
- inner_read_type = self.read_type
+ inner_read_type = self._scan_read_type
outer_extract_name_paths: Optional[List[List[str]]] = None
if self.nested_name_paths and any(
len(p) > 1 for p in self.nested_name_paths):
@@ -697,8 +718,13 @@ class TableRead:
# only valid against the latest schema, not each file's own
# names/types), then extract the requested sub-paths back to
# the user's flat schema.
- inner_read_type = self._widen_to_top_level_for_merge()
+ inner_read_type = self._with_predicate_extra_fields(
+ self._widen_to_top_level_for_merge())
outer_extract_name_paths = self.nested_name_paths
+ if outer_extract_name_paths is None and
self._needs_output_projection():
+ # Split readers own the output projection for iterator reads;
+ # the TableRead Arrow projection below is a final guard.
+ outer_extract_name_paths = self._output_extract_name_paths()
return RawFileSplitRead(
table=self.table,
predicate=self.predicate,
@@ -711,6 +737,45 @@ class TableRead:
limit=self.limit,
)
+ def _project_batch_to_output(self, batch: pyarrow.RecordBatch) ->
pyarrow.RecordBatch:
+ if not self._needs_output_projection():
+ return batch
+ if batch.schema.names == self._output_column_names:
+ return batch
+ name_to_pos = {name: i for i, name in enumerate(batch.schema.names)}
+ arrays = [batch.column(name_to_pos[name]) for name in
self._output_column_names]
+ fields = [batch.schema.field(name_to_pos[name]) for name in
self._output_column_names]
+ return pyarrow.RecordBatch.from_arrays(
+ arrays, schema=pyarrow.schema(fields))
+
+ def _needs_output_projection(self) -> bool:
+ return bool(self._predicate_extra_fields)
+
+ def _output_extract_name_paths(self) -> List[List[str]]:
+ return [[f.name] for f in self.read_type]
+
+ def _with_predicate_extra_fields(self, fields: List[DataField]) ->
List[DataField]:
+ names = {f.name for f in fields}
+ extras = [f for f in self._predicate_extra_fields if f.name not in
names]
+ return fields + extras
+
+ def _predicate_fields_outside_read_type(self) -> List[DataField]:
+ if self.predicate is None:
+ return []
+ read_names = {f.name for f in self.read_type}
+ predicate_fields = predicate_field_names(self.predicate)
+ missing = predicate_fields - read_names
+ if not missing:
+ return []
+ return [f for f in self._table_read_fields() if f.name in missing]
+
+ def _table_read_fields(self) -> List[DataField]:
+ from pypaimon.table.special_fields import SpecialFields
+ fields = self.table.fields
+ if self.table.options.row_tracking_enabled():
+ fields = SpecialFields.row_type_with_row_tracking(fields)
+ return fields
+
def _widen_to_top_level_for_merge(self) -> List[DataField]:
"""Unique top-level fields from ``self.nested_name_paths``, in path
order."""
table_fields_by_name = {f.name: f for f in self.table.fields}
diff --git a/paimon-python/pypaimon/tests/data_evolution_test.py
b/paimon-python/pypaimon/tests/data_evolution_test.py
index b0090f1a56..ffa10c018e 100644
--- a/paimon-python/pypaimon/tests/data_evolution_test.py
+++ b/paimon-python/pypaimon/tests/data_evolution_test.py
@@ -796,9 +796,10 @@ class DataEvolutionTest(unittest.TestCase):
)
self.assertEqual(
len(result_non_projected),
- 3,
- "Filter c > 150 with projection [id]: c not in read_type so filter
is dropped, all 3 rows returned.",
+ 1,
+ "Filter c > 150 with projection [id] should still evaluate c
internally.",
)
+ self.assertEqual(result_non_projected["id"].tolist(), [3])
self.assertEqual(
list(result_non_projected.columns),
["id"],
@@ -810,13 +811,13 @@ class DataEvolutionTest(unittest.TestCase):
try:
rows_from_iterator = list(table_read.to_iterator(splits))
except ValueError as e:
- if "Expected Arrow table or array" in str(e):
+ if "Expected Arrow" in str(e) and "RecordBatch" in str(e):
self.skipTest(
"RecordBatchReader path uses
polars.from_arrow(RecordBatch) which fails; "
"skip to_iterator projection assertion on this path"
)
raise
- self.assertEqual(len(rows_from_iterator), 3, "to_iterator should
return same row count as to_pandas")
+ self.assertEqual(len(rows_from_iterator), 1, "to_iterator should
return same row count as to_pandas")
for row in rows_from_iterator:
self.assertIsInstance(row, OffsetRow)
self.assertEqual(
diff --git a/paimon-python/pypaimon/tests/projection_predicate_index_test.py
b/paimon-python/pypaimon/tests/projection_predicate_index_test.py
index 493efb21a0..2953456223 100644
--- a/paimon-python/pypaimon/tests/projection_predicate_index_test.py
+++ b/paimon-python/pypaimon/tests/projection_predicate_index_test.py
@@ -91,11 +91,38 @@ class ProjectionPredicateIndexTest(unittest.TestCase):
w.close()
return self.catalog.get_table(full)
+ def _populate_data_evolution(self, table_name: str):
+ schema = Schema.from_pyarrow_schema(
+ self.pa_schema,
+ options={
+ 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true',
+ },
+ )
+ full = 'default.{}'.format(table_name)
+ self.catalog.create_table(full, schema, False)
+ table = self.catalog.get_table(full)
+ wb = table.new_batch_write_builder()
+ w = wb.new_write()
+ w.write_arrow(self.data)
+ wb.new_commit().commit(w.prepare_commit())
+ w.close()
+ return self.catalog.get_table(full)
+
def _read(self, read_builder):
scan = read_builder.new_scan()
read = read_builder.new_read()
return read.to_arrow(scan.plan().splits())
+ @staticmethod
+ def _rows_by_id(table):
+ data = table.to_pydict()
+ rows = [
+ {name: values[i] for name, values in data.items()}
+ for i in range(table.num_rows)
+ ]
+ return sorted(rows, key=lambda row: row['id'])
+
def test_pk_filter_on_non_pk_with_projection_keeping_filter_column(self):
"""The OffsetRow handed to FilterRecordReader uses read_type indices.
Before the fix this used to raise IndexError because the predicate
@@ -108,12 +135,12 @@ class ProjectionPredicateIndexTest(unittest.TestCase):
# Projection narrows read_type from [id, name, value] to [id, value]
rb = table.new_read_builder().with_projection(['id',
'value']).with_filter(pred)
- actual = self._read(rb).sort_by('id')
+ actual = self._read(rb)
+ rows = self._rows_by_id(actual)
self.assertEqual(actual.num_rows, 1)
self.assertEqual(actual.column_names, ['id', 'value'])
- self.assertEqual(actual.column('id').to_pylist(), [3])
- self.assertEqual(actual.column('value').to_pylist(), [30])
+ self.assertEqual(rows, [{'id': 3, 'value': 30}])
def test_pk_filter_on_non_pk_with_projection_reordering_columns(self):
"""Reordering the projection (value before id) changes the column
@@ -125,12 +152,15 @@ class ProjectionPredicateIndexTest(unittest.TestCase):
pred = pb.greater_than('value', 15)
rb = table.new_read_builder().with_projection(['value',
'id']).with_filter(pred)
- actual = self._read(rb).sort_by('id')
+ actual = self._read(rb)
+ rows = self._rows_by_id(actual)
self.assertEqual(actual.num_rows, 2)
self.assertEqual(actual.column_names, ['value', 'id'])
- self.assertEqual(actual.column('id').to_pylist(), [2, 3])
- self.assertEqual(actual.column('value').to_pylist(), [20, 30])
+ self.assertEqual(rows, [
+ {'value': 20, 'id': 2},
+ {'value': 30, 'id': 3},
+ ])
def test_pk_filter_no_projection_still_works(self):
"""Sanity: with no projection, behaviour must match the pre-fix
path."""
@@ -140,47 +170,77 @@ class ProjectionPredicateIndexTest(unittest.TestCase):
pred = pb.equal('value', 20)
rb = table.new_read_builder().with_filter(pred)
- actual = self._read(rb).sort_by('id')
+ actual = self._read(rb)
+ rows = self._rows_by_id(actual)
self.assertEqual(actual.num_rows, 1)
- self.assertEqual(actual.column('name').to_pylist(), ['b'])
- self.assertEqual(actual.column('value').to_pylist(), [20])
-
- def test_pk_filter_column_outside_projection_drops_filter(self):
- """When the filter column is *not* projected we cannot evaluate the
- predicate at row level — the contract is to fall through (no
- FilterRecordReader). The reader still produces the projected columns,
- and the count must not be smaller than the unfiltered projection.
- """
+ self.assertEqual(rows, [{'id': 2, 'name': 'b', 'value': 20}])
+
+ def test_pk_filter_column_outside_projection_still_filters(self):
+ """The filter column does not have to be visible in the final
output."""
table = self._populate_pk('test_pk_filter_outside_proj')
pb = table.new_read_builder().new_predicate_builder()
pred = pb.equal('value', 30)
- # value is NOT in the projection — predicate cannot apply at row level
rb = table.new_read_builder().with_projection(['id']).with_filter(pred)
- actual = self._read(rb).sort_by('id')
+ actual = self._read(rb)
+
+ self.assertEqual(actual.num_rows, 1)
+ self.assertEqual(actual.column_names, ['id'])
+ self.assertEqual(sorted(actual.column('id').to_pylist()), [3])
- # All three rows are still returned because the filter is dropped,
- # not misapplied with a stale index.
- self.assertEqual(actual.num_rows, 3)
+ def test_pk_string_filter_outside_projection_still_filters(self):
+ table = self._populate_pk('test_pk_string_filter_outside_proj')
+
+ pb = table.new_read_builder().new_predicate_builder()
+ pred = pb.startswith('name', 'a')
+
+ rb = table.new_read_builder().with_projection(['id']).with_filter(pred)
+ actual = self._read(rb)
+
+ self.assertEqual(actual.num_rows, 1)
self.assertEqual(actual.column_names, ['id'])
- self.assertEqual(actual.column('id').to_pylist(), [1, 2, 3])
+ self.assertEqual(actual.column('id').to_pylist(), [1])
def test_append_only_filter_with_projection_unchanged(self):
- """Append-only path uses arrow file-level pushdown by field name, not
- by index. Guard against any regression introduced by the fix.
- """
table = self._populate_append('test_append_proj_filter')
pb = table.new_read_builder().new_predicate_builder()
pred = pb.equal('value', 30)
rb = table.new_read_builder().with_projection(['id',
'value']).with_filter(pred)
- actual = self._read(rb).sort_by('id')
+ actual = self._read(rb)
+ rows = self._rows_by_id(actual)
self.assertEqual(actual.num_rows, 1)
self.assertEqual(actual.column_names, ['id', 'value'])
+ self.assertEqual(rows, [{'id': 3, 'value': 30}])
+
+ def test_append_only_filter_column_outside_projection_still_filters(self):
+ table = self._populate_append('test_append_filter_outside_proj')
+
+ pb = table.new_read_builder().new_predicate_builder()
+ pred = pb.equal('value', 30)
+
+ rb = table.new_read_builder().with_projection(['id']).with_filter(pred)
+ actual = self._read(rb)
+
+ self.assertEqual(actual.num_rows, 1)
+ self.assertEqual(actual.column_names, ['id'])
+ self.assertEqual(actual.column('id').to_pylist(), [3])
+
+ def
test_data_evolution_filter_column_outside_projection_still_filters(self):
+ table = self._populate_data_evolution('test_de_filter_outside_proj')
+
+ pb = table.new_read_builder().new_predicate_builder()
+ pred = pb.equal('value', 30)
+
+ rb = table.new_read_builder().with_projection(['id']).with_filter(pred)
+ actual = self._read(rb)
+
+ self.assertEqual(actual.num_rows, 1)
+ self.assertEqual(actual.column_names, ['id'])
self.assertEqual(actual.column('id').to_pylist(), [3])
@@ -262,6 +322,39 @@ class RewritePredicateIndicesUnitTest(unittest.TestCase):
self.assertIsNone(rewrite_predicate_indices(None, []))
+ def test_arrow_filter_support_marks_only_unsafe_string_predicates(self):
+ from pypaimon.read.push_down_utils import
predicate_supports_arrow_filter
+
+ pb = self._build_predicate()
+ safe = pb.greater_than('c', 1)
+ unsafe = pb.startswith('b', 'a')
+ mixed = pb.and_predicates([safe, unsafe])
+
+ self.assertTrue(predicate_supports_arrow_filter(safe))
+ self.assertFalse(predicate_supports_arrow_filter(unsafe))
+ self.assertFalse(predicate_supports_arrow_filter(mixed))
+
+ def test_missing_first_row_id_materializes_null_row_ids(self):
+ from pypaimon.read.reader.data_file_batch_reader import
DataFileBatchReader
+ from pypaimon.table.special_fields import SpecialFields
+
+ reader = object.__new__(DataFileBatchReader)
+ reader.system_fields = {SpecialFields.ROW_ID.name: 0}
+ reader.first_row_id = None
+ reader.row_id_offsets = None
+ reader._row_id_cursor = 0
+
+ batch = pa.RecordBatch.from_arrays(
+ [pa.array([0, 0], type=pa.int64())],
+ schema=pa.schema([
+ pa.field(SpecialFields.ROW_ID.name, pa.int64(), nullable=False)
+ ]),
+ )
+
+ actual = reader._assign_row_tracking(batch)
+
+ self.assertEqual(actual.column(0).to_pylist(), [None, None])
+
def test_raises_when_leaf_field_missing(self):
from pypaimon.read.push_down_utils import rewrite_predicate_indices
from pypaimon.schema.data_types import AtomicType, DataField
diff --git a/paimon-python/pypaimon/tests/reader_parallel_test.py
b/paimon-python/pypaimon/tests/reader_parallel_test.py
index adad137341..9e9897fb50 100644
--- a/paimon-python/pypaimon/tests/reader_parallel_test.py
+++ b/paimon-python/pypaimon/tests/reader_parallel_test.py
@@ -400,6 +400,21 @@ class ParallelReaderPrimaryKeyTest(unittest.TestCase):
df_p = parallel.to_pandas().sort_values(['dt',
'user_id']).reset_index(drop=True)
self.assertTrue(df_s.equals(df_p))
+ def test_parallel_row_kind_survives_output_projection(self):
+ rb_full = self.table.new_read_builder()
+ predicate = rb_full.new_predicate_builder().equal('behavior',
'v2-updated')
+ rb = self.table.new_read_builder().with_projection(
+ ['user_id', 'dt']).with_filter(predicate)
+ splits = rb.new_scan().plan().splits()
+ read = rb.new_read()
+ read.include_row_kind = True
+
+ result = read.to_arrow(splits, parallelism=4)
+
+ self.assertEqual(result.schema.names, ['_row_kind', 'user_id', 'dt'])
+ self.assertEqual(result.num_rows, 5)
+ self.assertEqual(set(result.column('_row_kind').to_pylist()), {'+I'})
+
if __name__ == '__main__':
unittest.main()
