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 d164d49556 [python] Defer BLOB payload reads until after filtering
(#8808)
d164d49556 is described below
commit d164d49556dc566293e1dcb311e71b62590ee2a8
Author: XiaoHongbo <[email protected]>
AuthorDate: Fri Aug 7 13:23:19 2026 +0800
[python] Defer BLOB payload reads until after filtering (#8808)
---
docs/docs/pypaimon/blob.md | 6 +
.../pypaimon/read/reader/auth_masking_reader.py | 9 +-
.../read/reader/deferred_blob_resolve_reader.py | 78 ++++
paimon-python/pypaimon/read/split_read.py | 92 +++-
paimon-python/pypaimon/read/table_read.py | 147 +++++-
.../pypaimon/tests/auth_masking_reader_test.py | 12 +
.../pypaimon/tests/deferred_blob_resolve_test.py | 493 +++++++++++++++++++++
7 files changed, 799 insertions(+), 38 deletions(-)
diff --git a/docs/docs/pypaimon/blob.md b/docs/docs/pypaimon/blob.md
index ddb9c40be4..dddc32bf2f 100644
--- a/docs/docs/pypaimon/blob.md
+++ b/docs/docs/pypaimon/blob.md
@@ -136,6 +136,12 @@ Without `blob-as-descriptor=true`, blob values are
materialized before
`row.get_blob(...)` returns; `new_input_stream()` then reads from
in-memory bytes, not from storage.
+For data-evolution reads, PyPaimon applies user filters, row-level
authorization
+filters, and limits before materializing projected scalar BLOB payloads. A user
+or authorization filter that references a BLOB value keeps that field eager.
+Column masking is applied after payload materialization. ARRAY and MAP elements
+containing BLOB values are not deferred.
+
## Lower-level: `Blob.from_bytes`
When you already have raw or descriptor bytes (for example from a custom
diff --git a/paimon-python/pypaimon/read/reader/auth_masking_reader.py
b/paimon-python/pypaimon/read/reader/auth_masking_reader.py
index d76aa51e44..aeee5e13e9 100644
--- a/paimon-python/pypaimon/read/reader/auth_masking_reader.py
+++ b/paimon-python/pypaimon/read/reader/auth_masking_reader.py
@@ -140,8 +140,7 @@ class AuthFilterReader(RecordBatchReader):
def __init__(self, inner_reader: RecordBatchReader, filter_fn:
Callable[[pa.RecordBatch], pa.Array]):
self._inner = inner_reader
self._filter_fn = filter_fn
- self.blob_field_indices = inner_reader.blob_field_indices
- self.vector_field_indices = inner_reader.vector_field_indices
+ self._adopt_metadata(inner_reader)
def read_arrow_batch(self) -> Optional[pa.RecordBatch]:
batch = self._inner.read_arrow_batch()
@@ -160,8 +159,7 @@ class AuthMaskingReader(RecordBatchReader):
self._inner = inner_reader
self._masking_rules = masking_rules
self._read_fields = read_fields
- self.blob_field_indices = inner_reader.blob_field_indices
- self.vector_field_indices = inner_reader.vector_field_indices
+ self._adopt_metadata(inner_reader)
read_field_names = {f.name for f in read_fields}
parsed = {}
for col, tj in masking_rules.items():
@@ -219,8 +217,7 @@ class ColumnProjectReader(RecordBatchReader):
def __init__(self, inner_reader: RecordBatchReader, columns: List[str]):
self._inner = inner_reader
self._columns = columns
- self.blob_field_indices = inner_reader.blob_field_indices
- self.vector_field_indices = inner_reader.vector_field_indices
+ self._adopt_metadata(inner_reader)
def read_arrow_batch(self) -> Optional[pa.RecordBatch]:
batch = self._inner.read_arrow_batch()
diff --git a/paimon-python/pypaimon/read/reader/deferred_blob_resolve_reader.py
b/paimon-python/pypaimon/read/reader/deferred_blob_resolve_reader.py
new file mode 100644
index 0000000000..9056f00dcb
--- /dev/null
+++ b/paimon-python/pypaimon/read/reader/deferred_blob_resolve_reader.py
@@ -0,0 +1,78 @@
+# 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.
+
+from typing import List, Optional
+
+import pyarrow as pa
+from pyarrow import RecordBatch
+
+from pypaimon.read.reader.iface.record_batch_reader import RecordBatchReader
+from pypaimon.table.row.blob import Blob
+
+
+class DeferredBlobResolveReader(RecordBatchReader):
+ """Materialize projected BLOB payloads after row filtering.
+
+ This must remain the outermost BLOB materialization layer because adopted
+ metadata still identifies the materialized columns as logical BLOB fields.
+ """
+
+ def __init__(self, inner: RecordBatchReader, file_io,
+ blob_field_names: List[str], blob_parallelism: int = 1):
+ self._inner = inner
+ self._file_io = file_io
+ self._blob_field_names = blob_field_names
+ self._blob_parallelism = max(1, blob_parallelism)
+ self._adopt_metadata(inner)
+
+ def read_arrow_batch(self) -> Optional[RecordBatch]:
+ batch = self._inner.read_arrow_batch()
+ if batch is None:
+ return None
+
+ columns = list(batch.columns)
+ fields = list(batch.schema)
+ changed = False
+ for field_name in self._blob_field_names:
+ column_index = batch.schema.get_field_index(field_name)
+ if column_index < 0:
+ continue
+ values = batch.column(column_index).to_pylist()
+ blobs = [Blob.from_bytes(value, self._file_io) for value in values]
+ if self._blob_parallelism > 1:
+ payloads = self._file_io.read_blobs_concurrent(
+ blobs, self._blob_parallelism)
+ else:
+ payloads = [blob.to_data() if blob else None for blob in blobs]
+ source_field = batch.schema.field(column_index)
+ columns[column_index] = pa.array(payloads, type=pa.large_binary())
+ fields[column_index] = pa.field(
+ field_name,
+ pa.large_binary(),
+ nullable=source_field.nullable,
+ metadata=source_field.metadata,
+ )
+ changed = True
+ if not changed:
+ return batch
+ return pa.RecordBatch.from_arrays(
+ columns,
+ schema=pa.schema(fields, metadata=batch.schema.metadata),
+ )
+
+ def close(self) -> None:
+ self._inner.close()
diff --git a/paimon-python/pypaimon/read/split_read.py
b/paimon-python/pypaimon/read/split_read.py
index d34c1800e0..2f6f09665f 100644
--- a/paimon-python/pypaimon/read/split_read.py
+++ b/paimon-python/pypaimon/read/split_read.py
@@ -42,7 +42,10 @@ from pypaimon.read.reader.concat_batch_reader import (
MergeAllBatchReader, DataEvolutionMergeReader)
from pypaimon.read.reader.concat_record_reader import ConcatRecordReader
+from pypaimon.read.reader.auth_masking_reader import AuthFilterReader
from pypaimon.read.reader.data_file_batch_reader import DataFileBatchReader
+from pypaimon.read.reader.deferred_blob_resolve_reader import \
+ DeferredBlobResolveReader
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
@@ -81,6 +84,32 @@ from pypaimon.utils.data_evolution_utils import
retrieve_anchor_file
KEY_PREFIX = "_KEY_"
KEY_FIELD_ID_START = 1000000
NULL_FIELD_INDEX = -1
+
+
+def deferred_blob_field_names(table, read_fields: List[DataField],
+ predicate: Optional[Predicate],
+ limit: Optional[int],
+ has_post_filter: bool = False) -> set:
+ # An auth filter also selects rows; defer past it too, like a
predicate/limit.
+ if ((predicate is None and limit is None and not has_post_filter)
+ or CoreOptions.blob_as_descriptor(table.options)):
+ return set()
+
+ inline_fields = (
+ CoreOptions.blob_descriptor_fields(table.options)
+ | CoreOptions.blob_view_fields(table.options)
+ )
+ predicate_fields = (
+ predicate_field_names(predicate) if predicate is not None else set()
+ )
+ return {
+ read_fields[index].name
+ for index in blob_field_indices(read_fields)
+ if read_fields[index].name not in inline_fields
+ and read_fields[index].name not in predicate_fields
+ }
+
+
ROW_SIDECAR_FORMAT = CoreOptions.FILE_FORMAT_ROW
_COMPRESS_EXTENSIONS = frozenset(['gz', 'bz2', 'deflate', 'snappy', 'lz4',
'zst'])
@@ -301,7 +330,7 @@ class SplitRead(ABC):
if has_nested:
raise NotImplementedError(
"Nested-field projection is not supported on BLOB files")
- blob_as_descriptor =
CoreOptions.blob_as_descriptor(self.table.options)
+ blob_as_descriptor =
self._read_blob_as_descriptor(read_file_fields)
blob_parallelism = self._blob_parallelism
format_reader = FormatBlobReader(self.table.file_io, file_path,
read_file_fields,
self.read_fields,
read_arrow_predicate, blob_as_descriptor,
@@ -446,6 +475,12 @@ class SplitRead(ABC):
return reader
+ def _read_blob_as_descriptor(self, field_names: List[str]) -> bool:
+ if CoreOptions.blob_as_descriptor(self.table.options):
+ return True
+ deferred_fields = getattr(self, '_deferred_blob_fields', set())
+ return any(field_name in deferred_fields for field_name in field_names)
+
@staticmethod
def _row_sidecar_file_name(file: DataFileMeta) -> Optional[str]:
row_files = [
@@ -1027,7 +1062,10 @@ class DataEvolutionSplitRead(SplitRead):
nested_name_paths: Optional[List[List[str]]] = None,
limit: Optional[int] = None,
outer_extract_name_paths: Optional[List[List[str]]] = None,
- outer_flat_read_type: Optional[List[DataField]] = None):
+ outer_flat_read_type: Optional[List[DataField]] = None,
+ post_merge_filter=None,
+ eager_blob_fields=None,
+ post_filter_after_inline=False):
self.row_ranges = None
actual_split = split
if isinstance(split, IndexedSplit):
@@ -1040,6 +1078,11 @@ class DataEvolutionSplitRead(SplitRead):
)
self.outer_extract_name_paths = outer_extract_name_paths
self.outer_flat_read_type = outer_flat_read_type
+ self._post_merge_filter = post_merge_filter
+ # Apply the auth filter after inline BLOB resolution, so scalar BLOBs
still defer.
+ self._post_filter_after_inline = post_filter_after_inline
+ self._eager_blob_fields = set(eager_blob_fields or [])
+ self._deferred_blob_fields = self._deferred_blob_field_names()
def _push_down_predicate(self) -> Optional[Predicate]:
# Data evolution: files may have different schemas, so we don't push
predicate
@@ -1059,8 +1102,35 @@ class DataEvolutionSplitRead(SplitRead):
prescan_reader_factory=lambda names:
self._create_prescan_reader(names),
blob_parallelism=blob_parallelism)
+ if self._post_filter_after_inline:
+ if self._post_merge_filter is not None:
+ reader = AuthFilterReader(reader, self._post_merge_filter)
+ if self.limit is not None:
+ reader = LimitedRecordBatchReader(reader, self.limit)
+
+ if self._deferred_blob_fields:
+ blob_names = [
+ field.name for field in self.read_fields
+ if field.name in self._deferred_blob_fields
+ ]
+ reader = DeferredBlobResolveReader(
+ reader,
+ self.table.file_io,
+ blob_names,
+ blob_parallelism=self._blob_parallelism,
+ )
+
return reader
+ def _deferred_blob_field_names(self) -> set:
+ return deferred_blob_field_names(
+ self.table,
+ self.read_fields,
+ self.predicate_for_reader,
+ self.limit,
+ has_post_filter=self._post_merge_filter is not None,
+ ) - self._eager_blob_fields
+
def _create_raw_reader(self) -> RecordReader:
"""Core read logic: split_by_row_id -> suppliers -> ConcatBatchReader
-> filter."""
files = self.split.files
@@ -1097,6 +1167,9 @@ class DataEvolutionSplitRead(SplitRead):
else:
reader = merge_reader
+ if self._post_merge_filter is not None and not
self._post_filter_after_inline:
+ reader = AuthFilterReader(reader, self._post_merge_filter)
+
if self.outer_extract_name_paths:
if self.outer_flat_read_type is None:
raise ValueError(
@@ -1107,7 +1180,7 @@ class DataEvolutionSplitRead(SplitRead):
reader = NestedLeafBatchReader(
reader, self.outer_extract_name_paths,
self.outer_flat_read_type)
- if self.limit is not None:
+ if self.limit is not None and not self._post_filter_after_inline:
reader = LimitedRecordBatchReader(reader, self.limit)
return reader
@@ -1170,17 +1243,16 @@ class DataEvolutionSplitRead(SplitRead):
if not prescan_fields:
return EmptyRecordBatchReader()
- # When there's a normal field predicate, don't push down limit to
prescan reader
- # because the outer reader will apply predicate+limit filtering,
- # while prescan reader would only apply limit without normal field
predicate
- # TODO support limit+predicate push down
+ # Skip limit push-down when the outer reader also selects rows
(predicate or auth
+ # filter): prescan's first-N rows would differ from the outer set.
TODO: push down.
+ skip_limit = self.predicate is not None or self._post_merge_filter is
not None
prescan_read = DataEvolutionSplitRead(
table=self.table,
predicate=self.predicate,
read_type=prescan_fields,
split=self.split,
row_tracking_enabled=False,
- limit=None if self.predicate else self.limit,
+ limit=None if skip_limit else self.limit,
)
prescan_read.row_ranges = self.row_ranges
return prescan_read._create_raw_reader()
@@ -1329,7 +1401,7 @@ class DataEvolutionSplitRead(SplitRead):
[read_fields[0]]
).field(0).type,
self.row_ranges,
- CoreOptions.blob_as_descriptor(self.table.options),
+ self._read_blob_as_descriptor([read_fields[0].name]),
deletion_vector=deletion_vector,
batch_size=batch_size,
blob_parallelism=self._blob_parallelism,
@@ -1384,7 +1456,7 @@ class DataEvolutionSplitRead(SplitRead):
read_fields,
self.read_fields,
None,
- CoreOptions.blob_as_descriptor(self.table.options),
+ self._read_blob_as_descriptor(read_fields),
batch_size=self.table.options.read_batch_size(),
row_indices=row_indices,
blob_parallelism=blob_parallelism,
diff --git a/paimon-python/pypaimon/read/table_read.py
b/paimon-python/pypaimon/read/table_read.py
index 6e641b5096..c2ba44545a 100644
--- a/paimon-python/pypaimon/read/table_read.py
+++ b/paimon-python/pypaimon/read/table_read.py
@@ -24,16 +24,18 @@ import pandas
import pyarrow
from pypaimon.common.predicate import Predicate
+from pypaimon.common.predicate_json_parser import extract_referenced_fields
from pypaimon.read.push_down_utils import predicate_field_names
from pypaimon.read.query_auth_split import QueryAuthSplit
from pypaimon.read.reader.auth_masking_reader import (
AuthFilterReader, AuthMaskingReader, ColumnProjectReader,
RecordReaderToBatchAdapter, BatchToRecordReaderAdapter)
from pypaimon.read.reader.iface.record_batch_reader import RecordBatchReader
+from pypaimon.read.reader.limited_record_reader import LimitedRecordBatchReader
from pypaimon.read.split import Split
from pypaimon.read.split_read import (DataEvolutionSplitRead,
MergeFileSplitRead, RawFileSplitRead,
- SplitRead)
+ SplitRead, deferred_blob_field_names)
from pypaimon.schema.data_types import DataField, PyarrowFieldParser
from pypaimon.table.row.offset_row import OffsetRow
@@ -111,6 +113,15 @@ class TableRead:
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._deferred_blob_fields = (
+ deferred_blob_field_names(
+ self.table,
+ self._scan_read_type,
+ self.predicate,
+ limit,
+ )
+ if self.table.options.data_evolution_enabled() else set()
+ )
self.include_row_kind = include_row_kind
self.nested_name_paths = nested_name_paths
self.limit = limit
@@ -124,7 +135,9 @@ class TableRead:
for split in splits:
if limit is not None and count >= limit:
return
- reader = self.__create_reader_for_split(split)
+ remaining = None if limit is None else limit - count
+ reader = self.__create_reader_for_split(
+ split, limit=remaining)
try:
for batch in iter(reader.read_batch, None):
for row in iter(batch.next, None):
@@ -188,7 +201,9 @@ class TableRead:
order. Must be ``>= 1``. Note that with ``>= 2`` (or auto)
and a ``limit`` set, the returned rows are an arbitrary
subset of the requested size, since which splits fill the row
- quota first is non-deterministic.
+ quota first is non-deterministic. Data-evolution reads with
+ deferred BLOB resolution run serially when a limit may discard
+ rows, so payloads are not materialized from discarded splits.
blob_parallelism: number of threads for concurrent blob reads
within each batch. ``None`` or ``1`` (default) reads blobs
serially; ``>= 2`` uses a thread pool with ``pread`` for
@@ -230,7 +245,8 @@ class TableRead:
for split in splits:
if remaining is not None and remaining <= 0:
break
- reader = self.__create_reader_for_split(split, blob_parallelism)
+ reader = self.__create_reader_for_split(
+ split, blob_parallelism, limit=remaining)
try:
if isinstance(reader, RecordBatchReader):
for batch in iter(reader.read_arrow_batch, None):
@@ -331,7 +347,25 @@ class TableRead:
overhead, no behavior change). A single split is never
parallelized since there is nothing to fan out across.
"""
- return effective >= 2 and len(splits) >= 2
+ deferred_limit_may_prune = (
+ self.limit is not None
+ and self._deferred_blob_fields
+ and not self._limit_covers_all_splits(splits)
+ )
+ return (effective >= 2 and len(splits) >= 2
+ and not deferred_limit_may_prune)
+
+ def _limit_covers_all_splits(self, splits: List[Split]) -> bool:
+ """Return whether split metadata proves that LIMIT cannot drop rows."""
+ total_rows = 0
+ for split in splits:
+ merged_row_count = split.merged_row_count()
+ if merged_row_count is None:
+ return False
+ total_rows += merged_row_count
+ if total_rows > self.limit:
+ return False
+ return True
def _to_arrow_parallel(
self,
@@ -650,12 +684,33 @@ class TableRead:
dataset = TorchDataset(self, splits)
return dataset
- def _create_split_read(self, split: Split, blob_parallelism: int = 1,
read_type=None) -> SplitRead:
- sr = self._build_split_read(split, read_type)
+ def _create_split_read(self, split: Split, blob_parallelism: int = 1,
+ read_type=None, limit: Optional[int] = None,
+ push_down_limit: bool = True,
+ post_merge_filter=None,
+ eager_blob_fields=None,
+ post_filter_after_inline: bool = False) ->
SplitRead:
+ sr = self._build_split_read(
+ split,
+ read_type,
+ limit,
+ push_down_limit,
+ post_merge_filter,
+ eager_blob_fields,
+ post_filter_after_inline,
+ )
sr._blob_parallelism = blob_parallelism
return sr
- def _build_split_read(self, split: Split, read_type=None) -> SplitRead:
+ def _build_split_read(self, split: Split, read_type=None,
+ limit: Optional[int] = None,
+ push_down_limit: bool = True,
+ post_merge_filter=None,
+ eager_blob_fields=None,
+ post_filter_after_inline: bool = False) -> SplitRead:
+ effective_limit = (
+ self.limit if limit is None else limit
+ ) if push_down_limit else None
effective_read_type = read_type if read_type is not None else
self.read_type
scan_read_type = self._with_predicate_extra_fields(read_type) if
read_type is not None else self._scan_read_type
if self.table.is_primary_key_table and not split.raw_convertible:
@@ -705,7 +760,7 @@ class TableRead:
outer_extract_name_paths=outer_extract_name_paths,
outer_flat_read_type=(
effective_read_type if outer_extract_name_paths else None),
- limit=self.limit,
+ limit=effective_limit,
)
elif self.table.options.data_evolution_enabled():
if self.nested_name_paths and any(
@@ -726,7 +781,10 @@ class TableRead:
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,
+ limit=effective_limit,
+ post_merge_filter=post_merge_filter,
+ eager_blob_fields=eager_blob_fields,
+ post_filter_after_inline=post_filter_after_inline,
)
else:
inner_read_type = scan_read_type
@@ -752,7 +810,7 @@ class TableRead:
outer_extract_name_paths=outer_extract_name_paths,
outer_flat_read_type=(
effective_read_type if outer_extract_name_paths else None),
- limit=self.limit,
+ limit=effective_limit,
)
def _project_batch_to_output(self, batch: pyarrow.RecordBatch) ->
pyarrow.RecordBatch:
@@ -812,18 +870,27 @@ class TableRead:
widened.append(field)
return widened
- def __create_reader_for_split(self, split, blob_parallelism=1):
+ def __create_reader_for_split(self, split, blob_parallelism=1,
+ limit: Optional[int] = None):
auth_result = None
if isinstance(split, QueryAuthSplit):
auth_result = split.auth_result
split = split.split
if auth_result is not None:
- return self.__authed_reader(split, auth_result, blob_parallelism)
- else:
- return self._create_split_read(split,
blob_parallelism=blob_parallelism).create_reader()
-
- def __authed_reader(self, split, auth_result, blob_parallelism=1):
+ return self.__authed_reader(
+ split, auth_result, blob_parallelism, limit)
+ if limit is None:
+ return self._create_split_read(
+ split, blob_parallelism=blob_parallelism).create_reader()
+ return self._create_split_read(
+ split,
+ blob_parallelism=blob_parallelism,
+ limit=limit,
+ ).create_reader()
+
+ def __authed_reader(self, split, auth_result, blob_parallelism=1,
+ limit: Optional[int] = None):
table_fields = self.table.fields
read_fields = self.read_type
@@ -832,9 +899,34 @@ class TableRead:
if extra_fields:
effective_read_type = read_fields + extra_fields
- reader = self._create_split_read(
- split, blob_parallelism=blob_parallelism,
- read_type=effective_read_type).create_reader()
+ filter_fn = auth_result.extract_row_filter()
+ effective_limit = self.limit if limit is None else limit
+ auth_fields = (
+ self._auth_filter_field_names(auth_result, effective_read_type)
+ if filter_fn is not None else set()
+ )
+ inline_blob_fields = (
+ self.table.options.blob_descriptor_fields()
+ | self.table.options.blob_view_fields()
+ )
+ embed_filter = (
+ filter_fn is not None
+ and self.table.options.data_evolution_enabled()
+ )
+ # If the auth filter references an inline BLOB, run it after inline
resolution (in
+ # the split read) so it sees resolved payloads while scalar BLOBs
still defer.
+ post_filter_after_inline = embed_filter and bool(auth_fields &
inline_blob_fields)
+ split_read = self._create_split_read(
+ split,
+ blob_parallelism=blob_parallelism,
+ read_type=effective_read_type,
+ limit=limit,
+ push_down_limit=filter_fn is None or embed_filter,
+ post_merge_filter=filter_fn if embed_filter else None,
+ eager_blob_fields=auth_fields if embed_filter else None,
+ post_filter_after_inline=post_filter_after_inline,
+ )
+ reader = split_read.create_reader()
needs_convert_back = False
if not isinstance(reader, RecordBatchReader):
@@ -842,9 +934,10 @@ class TableRead:
reader = RecordReaderToBatchAdapter(reader, schema,
include_row_kind=self.include_row_kind)
needs_convert_back = True
- filter_fn = auth_result.extract_row_filter()
- if filter_fn:
+ if filter_fn and not embed_filter:
reader = AuthFilterReader(reader, filter_fn)
+ if effective_limit is not None:
+ reader = LimitedRecordBatchReader(reader, effective_limit)
if auth_result.column_masking:
reader = AuthMaskingReader(reader, auth_result.column_masking,
effective_read_type)
@@ -858,6 +951,16 @@ class TableRead:
return reader
+ @staticmethod
+ def _auth_filter_field_names(auth_result, read_fields) -> set:
+ filters = getattr(auth_result, "filter", None)
+ if not filters:
+ return {field.name for field in read_fields}
+ names = set()
+ for filter_json in filters:
+ names.update(extract_referenced_fields(filter_json))
+ return names
+
@staticmethod
def convert_rows_to_arrow_batch(row_tuples: List[tuple], schema:
pyarrow.Schema) -> pyarrow.RecordBatch:
columns_data = zip(*row_tuples)
diff --git a/paimon-python/pypaimon/tests/auth_masking_reader_test.py
b/paimon-python/pypaimon/tests/auth_masking_reader_test.py
index d9b40f9741..7745f94a5a 100644
--- a/paimon-python/pypaimon/tests/auth_masking_reader_test.py
+++ b/paimon-python/pypaimon/tests/auth_masking_reader_test.py
@@ -292,6 +292,18 @@ class TestMaskingFieldValidation(unittest.TestCase):
class TestAuthFilterReader(unittest.TestCase):
+ def test_adopts_reader_metadata(self):
+ inner = _FakeBatchReader([])
+ inner.file_io = object()
+ inner.blob_field_indices = frozenset([1])
+ inner.vector_field_indices = frozenset([2])
+
+ reader = AuthFilterReader(inner, lambda batch: None)
+
+ self.assertIs(reader.file_io, inner.file_io)
+ self.assertEqual(reader.blob_field_indices, inner.blob_field_indices)
+ self.assertEqual(reader.vector_field_indices,
inner.vector_field_indices)
+
def test_filters_rows(self):
import pyarrow.compute as pc
diff --git a/paimon-python/pypaimon/tests/deferred_blob_resolve_test.py
b/paimon-python/pypaimon/tests/deferred_blob_resolve_test.py
new file mode 100644
index 0000000000..2fbe0fb627
--- /dev/null
+++ b/paimon-python/pypaimon/tests/deferred_blob_resolve_test.py
@@ -0,0 +1,493 @@
+# 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.
+
+import json
+import os
+import shutil
+import tempfile
+import unittest
+from unittest.mock import patch
+
+import pyarrow as pa
+import pyarrow.compute as pc
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.catalog.table_query_auth import TableQueryAuthResult
+from pypaimon.read.query_auth_split import QueryAuthSplit
+from pypaimon.table.row.blob import BlobRef
+
+
+_ROW_COUNT = 10
+_TABLE_OPTIONS = {
+ "row-tracking.enabled": "true",
+ "data-evolution.enabled": "true",
+}
+
+
+class _BlobCountingFileIO:
+
+ def __init__(self, inner):
+ self._inner = inner
+ self.blobs_fetched = 0
+ self.concurrent_read_calls = 0
+
+ def read_blobs_concurrent(self, blobs, parallelism):
+ self.concurrent_read_calls += 1
+ self.blobs_fetched += sum(blob is not None for blob in blobs)
+ return self._inner.read_blobs_concurrent(blobs, parallelism)
+
+ def __getattr__(self, name):
+ return getattr(self._inner, name)
+
+
+class _RejectScoreOneAuthResult:
+ column_masking = None
+ filter = [json.dumps({
+ "kind": "LEAF",
+ "transform": {
+ "name": "FIELD_REF",
+ "fieldRef": {"index": 2, "name": "score", "type": "INT"},
+ },
+ "function": "NOT_EQUAL",
+ "literals": [1],
+ })]
+
+ @staticmethod
+ def get_extra_fields_for_filter(read_fields, table_fields):
+ return []
+
+ @staticmethod
+ def extract_row_filter():
+ return lambda batch: pc.not_equal(batch.column("score"), 1)
+
+
+class _PayloadAuthResult:
+ column_masking = None
+
+ def __init__(self, expected_payload):
+ self._expected_payload = expected_payload
+ self.filter = [json.dumps({
+ "kind": "LEAF",
+ "transform": {
+ "name": "FIELD_REF",
+ "fieldRef": {
+ "index": 1,
+ "name": "payload",
+ "type": "BYTES",
+ },
+ },
+ "function": "EQUAL",
+ "literals": [],
+ })]
+
+ @staticmethod
+ def get_extra_fields_for_filter(read_fields, table_fields):
+ return []
+
+ def extract_row_filter(self):
+ return lambda batch: pc.equal(
+ batch.column("payload"), self._expected_payload)
+
+
+class DeferredBlobResolveTest(unittest.TestCase):
+
+ @classmethod
+ def setUpClass(cls):
+ cls.tempdir = tempfile.mkdtemp()
+ original_to_data = BlobRef.to_data
+
+ def counted_to_data(blob):
+ file_io = getattr(blob._uri_reader, "_file_io", None)
+ if isinstance(file_io, _BlobCountingFileIO):
+ file_io.blobs_fetched += 1
+ return original_to_data(blob)
+
+ cls.blob_ref_to_data_patch = patch.object(
+ BlobRef, "to_data", counted_to_data)
+ cls.blob_ref_to_data_patch.start()
+ cls.catalog = CatalogFactory.create({
+ "warehouse": os.path.join(cls.tempdir, "warehouse")
+ })
+ cls.catalog.create_database("default", False)
+ cls.schema = pa.schema([
+ ("sample_id", pa.string()),
+ ("payload", pa.large_binary()),
+ ("score", pa.int32()),
+ ])
+
+ @classmethod
+ def tearDownClass(cls):
+ cls.blob_ref_to_data_patch.stop()
+ shutil.rmtree(cls.tempdir, ignore_errors=True)
+
+ def _create_table(self, name, extra_options=None, payloads=None,
+ partition_keys=None, sample_ids=None):
+ options = dict(_TABLE_OPTIONS)
+ options.update(extra_options or {})
+ identifier = "default.%s" % name
+ self.catalog.create_table(
+ identifier,
+ Schema.from_pyarrow_schema(
+ self.schema,
+ partition_keys=partition_keys,
+ options=options,
+ ),
+ False,
+ )
+ table = self.catalog.get_table(identifier)
+ write_builder = table.new_batch_write_builder()
+ writer = write_builder.new_write()
+ commit = write_builder.new_commit()
+ if sample_ids is None:
+ sample_ids = [
+ "sample_%d" % index for index in range(_ROW_COUNT)
+ ]
+ writer.write_arrow(pa.table({
+ "sample_id": sample_ids,
+ "payload": (
+ payloads if payloads is not None else
+ [bytes([index]) * 1024 for index in range(_ROW_COUNT)]
+ ),
+ "score": list(range(_ROW_COUNT)),
+ }, schema=self.schema))
+ commit.commit(writer.prepare_commit())
+ writer.close()
+ commit.close()
+ return self.catalog.get_table(identifier)
+
+ def _read(self, table, predicate, limit=None, blob_parallelism=None):
+ counting_file_io = _BlobCountingFileIO(table.file_io)
+ table.file_io = counting_file_io
+ read_builder = table.new_read_builder()
+ if predicate is not None:
+ read_builder = read_builder.with_filter(predicate)
+ read_builder = read_builder.with_projection(
+ ["sample_id", "payload", "score"])
+ if limit is not None:
+ read_builder = read_builder.with_limit(limit)
+ splits = read_builder.new_scan().plan().splits()
+ table_read = read_builder.new_read()
+ if blob_parallelism is None:
+ batch_reader = table_read.to_arrow_batch_reader(splits)
+ else:
+ batch_reader = table_read.to_arrow_batch_reader(
+ splits, blob_parallelism=blob_parallelism)
+ result = pa.Table.from_batches(batch_reader)
+ return result, counting_file_io
+
+ def test_fetches_payloads_only_for_filtered_rows(self):
+ table = self._create_table("defer_filtered")
+ predicate = table.new_read_builder().new_predicate_builder().less_than(
+ "score", 5)
+
+ result, counting_file_io = self._read(table, predicate)
+
+ self.assertEqual(5, result.num_rows)
+ self.assertEqual(5, counting_file_io.blobs_fetched)
+ self.assertEqual(0, counting_file_io.concurrent_read_calls)
+ self.assertEqual(
+ [bytes([index]) * 1024 for index in range(5)],
+ result.column("payload").to_pylist(),
+ )
+
+ def test_applies_limit_before_fetching_payloads(self):
+ table = self._create_table("defer_limit")
+ predicate = table.new_read_builder().new_predicate_builder().less_than(
+ "score", 8)
+
+ result, counting_file_io = self._read(table, predicate, limit=2)
+
+ self.assertEqual(2, result.num_rows)
+ self.assertEqual(2, counting_file_io.blobs_fetched)
+
+ def test_limit_without_predicate_defers_payloads(self):
+ table = self._create_table("defer_limit_only")
+
+ result, counting_file_io = self._read(table, None, limit=2)
+
+ self.assertEqual(2, result.num_rows)
+ self.assertEqual(2, counting_file_io.blobs_fetched)
+
+ def test_limit_does_not_prefetch_payloads_across_splits(self):
+ table = self._create_table(
+ "defer_limit_splits",
+ extra_options={"source.split.target-size": "1b"},
+ partition_keys=["sample_id"],
+ )
+ counting_file_io = _BlobCountingFileIO(table.file_io)
+ table.file_io = counting_file_io
+ splits = table.new_read_builder().new_scan().plan().splits()
+ read_builder = table.new_read_builder().with_limit(1)
+
+ table_read = read_builder.new_read()
+ with patch.object(
+ table_read,
+ "_to_arrow_parallel",
+ side_effect=AssertionError("deferred LIMIT must run serially"),
+ ) as parallel_read:
+ result = table_read.to_arrow(splits, parallelism=4)
+
+ self.assertGreater(len(splits), 1)
+ parallel_read.assert_not_called()
+ self.assertEqual(1, result.num_rows)
+ self.assertEqual(1, counting_file_io.blobs_fetched)
+
+ def test_limit_covering_all_rows_preserves_parallelism(self):
+ table = self._create_table(
+ "defer_limit_all_rows",
+ extra_options={"source.split.target-size": "1b"},
+ partition_keys=["sample_id"],
+ )
+ counting_file_io = _BlobCountingFileIO(table.file_io)
+ table.file_io = counting_file_io
+ read_builder = table.new_read_builder().with_limit(_ROW_COUNT)
+ splits = read_builder.new_scan().plan().splits()
+ table_read = read_builder.new_read()
+
+ with patch.object(
+ table_read,
+ "_to_arrow_parallel",
+ wraps=table_read._to_arrow_parallel,
+ ) as parallel_read:
+ result = table_read.to_arrow(splits, parallelism=4)
+
+ self.assertGreater(len(splits), 1)
+ parallel_read.assert_called_once()
+ self.assertEqual(_ROW_COUNT, result.num_rows)
+ self.assertEqual(_ROW_COUNT, counting_file_io.blobs_fetched)
+
+ def test_iterator_passes_remaining_limit_across_splits(self):
+ table = self._create_table(
+ "defer_iterator_limit_splits",
+ extra_options={"source.split.target-size": "1b"},
+ partition_keys=["sample_id"],
+ sample_ids=["a"] + ["b"] * (_ROW_COUNT - 1),
+ )
+ counting_file_io = _BlobCountingFileIO(table.file_io)
+ table.file_io = counting_file_io
+ read_builder = table.new_read_builder().with_projection(
+ ["sample_id", "payload", "score"]
+ ).with_limit(2)
+ splits = read_builder.new_scan().plan().splits()
+
+ rows = list(read_builder.new_read().to_iterator(splits))
+
+ self.assertEqual(2, len(splits))
+ self.assertEqual(2, len(rows))
+ self.assertEqual(2, counting_file_io.blobs_fetched)
+
+ def test_iterator_applies_limit_after_auth_filter(self):
+ table = self._create_table(
+ "defer_iterator_auth_limit_splits",
+ extra_options={"source.split.target-size": "1b"},
+ partition_keys=["sample_id"],
+ sample_ids=["a"] + ["b"] * (_ROW_COUNT - 1),
+ )
+ read_builder = table.new_read_builder().with_projection(
+ ["sample_id", "payload", "score"]
+ ).with_limit(2)
+ counting_file_io = _BlobCountingFileIO(table.file_io)
+ table.file_io = counting_file_io
+ auth_result = _RejectScoreOneAuthResult()
+ splits = [
+ QueryAuthSplit(split, auth_result)
+ for split in read_builder.new_scan().plan().splits()
+ ]
+
+ scores = [
+ row.get_field(2)
+ for row in read_builder.new_read().to_iterator(splits)
+ ]
+
+ self.assertEqual([0, 2], scores)
+ self.assertEqual(2, counting_file_io.blobs_fetched)
+
+ def test_auth_blob_filter_keeps_eager_resolution(self):
+ table = self._create_table("defer_auth_blob_filter")
+ expected_payload = bytes([3]) * 1024
+ counting_file_io = _BlobCountingFileIO(table.file_io)
+ table.file_io = counting_file_io
+ read_builder = table.new_read_builder().with_projection(
+ ["sample_id", "payload", "score"]
+ ).with_limit(1)
+ auth_result = _PayloadAuthResult(expected_payload)
+ splits = [
+ QueryAuthSplit(split, auth_result)
+ for split in read_builder.new_scan().plan().splits()
+ ]
+
+ result = pa.Table.from_batches(
+ read_builder.new_read().to_arrow_batch_reader(
+ splits, blob_parallelism=4)
+ )
+
+ self.assertEqual([expected_payload],
result.column("payload").to_pylist())
+ self.assertEqual(_ROW_COUNT, counting_file_io.blobs_fetched)
+
+ def test_auth_only_defers_non_auth_payloads(self):
+ # An auth filter with no predicate/limit still defers scalar BLOBs, so
payloads of
+ # the rows the auth filter drops are not read.
+ table = self._create_table("defer_auth_only")
+ counting_file_io = _BlobCountingFileIO(table.file_io)
+ table.file_io = counting_file_io
+ read_builder = table.new_read_builder().with_projection(
+ ["sample_id", "payload", "score"])
+ splits = [
+ QueryAuthSplit(split, _RejectScoreOneAuthResult())
+ for split in read_builder.new_scan().plan().splits()
+ ]
+
+ result = pa.Table.from_batches(
+ read_builder.new_read().to_arrow_batch_reader(
+ splits, blob_parallelism=4))
+
+ self.assertEqual(_ROW_COUNT - 1, result.num_rows)
+ self.assertEqual(_ROW_COUNT - 1, counting_file_io.blobs_fetched)
+
+ def test_auth_filter_preserves_file_io_for_blob_descriptors(self):
+ table = self._create_table(
+ "auth_blob_descriptor",
+ extra_options={"blob-as-descriptor": "true"},
+ )
+ read_builder = table.new_read_builder().with_projection(
+ ["sample_id", "payload", "score"])
+ splits = [
+ QueryAuthSplit(split, _RejectScoreOneAuthResult())
+ for split in read_builder.new_scan().plan().splits()
+ ]
+
+ payloads = [
+ row.get_blob(1).to_data()
+ for row in read_builder.new_read().to_iterator(splits)
+ ]
+
+ self.assertEqual(_ROW_COUNT - 1, len(payloads))
+ self.assertEqual(
+ [bytes([index]) * 1024 for index in range(_ROW_COUNT) if index !=
1],
+ payloads,
+ )
+
+ def test_hidden_auth_field_preserves_file_io_for_blob_descriptors(self):
+ table = self._create_table(
+ "hidden_auth_field_blob_descriptor",
+ extra_options={"blob-as-descriptor": "true"},
+ )
+ read_builder = table.new_read_builder().with_projection(
+ ["sample_id", "payload"])
+ auth_result = TableQueryAuthResult(
+ filter=_RejectScoreOneAuthResult.filter,
+ column_masking=None,
+ )
+ splits = [
+ QueryAuthSplit(split, auth_result)
+ for split in read_builder.new_scan().plan().splits()
+ ]
+
+ payloads = [
+ row.get_blob(1).to_data()
+ for row in read_builder.new_read().to_iterator(splits)
+ ]
+
+ self.assertEqual(
+ [bytes([index]) * 1024 for index in range(_ROW_COUNT) if index !=
1],
+ payloads,
+ )
+
+ def test_auth_masking_preserves_file_io_for_blob_descriptors(self):
+ table = self._create_table(
+ "auth_masking_blob_descriptor",
+ extra_options={"blob-as-descriptor": "true"},
+ )
+ read_builder = table.new_read_builder().with_projection(
+ ["sample_id", "payload"])
+ auth_result = TableQueryAuthResult(
+ filter=None,
+ column_masking={"sample_id": json.dumps({"name": "NULL"})},
+ )
+ splits = [
+ QueryAuthSplit(split, auth_result)
+ for split in read_builder.new_scan().plan().splits()
+ ]
+
+ rows = [
+ (row.get_field(0), row.get_blob(1).to_data())
+ for row in read_builder.new_read().to_iterator(splits)
+ ]
+
+ self.assertEqual([None] * _ROW_COUNT, [row[0] for row in rows])
+ self.assertEqual(
+ [bytes([index]) * 1024 for index in range(_ROW_COUNT)],
+ [row[1] for row in rows],
+ )
+
+ def test_preserves_null_payloads_after_filtering(self):
+ payloads = [
+ None if index == 1 else bytes([index]) * 1024
+ for index in range(_ROW_COUNT)
+ ]
+ table = self._create_table("defer_null", payloads=payloads)
+ predicate = table.new_read_builder().new_predicate_builder().less_than(
+ "score", 4)
+
+ result, counting_file_io = self._read(table, predicate)
+
+ self.assertEqual(4, result.num_rows)
+ self.assertEqual(3, counting_file_io.blobs_fetched)
+ self.assertEqual(payloads[:4], result.column("payload").to_pylist())
+
+ def test_blob_predicate_keeps_eager_resolution(self):
+ table = self._create_table("defer_blob_predicate")
+ expected_payload = bytes([3]) * 1024
+ predicate = table.new_read_builder().new_predicate_builder().equal(
+ "payload", expected_payload)
+
+ result, counting_file_io = self._read(
+ table, predicate, blob_parallelism=4)
+
+ self.assertEqual(1, result.num_rows)
+ self.assertEqual([expected_payload],
result.column("payload").to_pylist())
+ self.assertEqual(_ROW_COUNT, counting_file_io.blobs_fetched)
+
+ def test_defers_payloads_for_blob_fallback_reader(self):
+ table = self._create_table("defer_fallback")
+ update_builder = table.new_batch_write_builder()
+ table_update =
update_builder.new_update().with_update_type(["payload"])
+ updated_payload = b"updated-payload"
+ update_messages = table_update.update_by_arrow_with_row_id(pa.table({
+ "_ROW_ID": pa.array([3], type=pa.int64()),
+ "payload": pa.array([updated_payload], type=pa.large_binary()),
+ }))
+ update_builder.new_commit().commit(update_messages)
+
+ predicate = table.new_read_builder().new_predicate_builder().less_than(
+ "score", 5)
+ result, counting_file_io = self._read(table, predicate)
+
+ self.assertEqual(5, result.num_rows)
+ self.assertEqual(5, counting_file_io.blobs_fetched)
+ payload_by_score = dict(zip(
+ result.column("score").to_pylist(),
+ result.column("payload").to_pylist(),
+ ))
+ self.assertEqual(
+ updated_payload,
+ payload_by_score[3],
+ )
+
+
+if __name__ == "__main__":
+ unittest.main()
