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 1da14460c1 [python] Expand native read coverage for nested and
streaming reads (#10009)
1da14460c1 is described below
commit 1da14460c1dfe90051e518c33eddbaf32c255ce4
Author: Jingsong Lee <[email protected]>
AuthorDate: Sun Sep 20 19:00:19 2026 +0800
[python] Expand native read coverage for nested and streaming reads (#10009)
---
paimon-python/pypaimon/read/native_plan.py | 18 +-
paimon-python/pypaimon/read/stream_read_builder.py | 36 ++-
.../pypaimon/read/streaming_table_scan.py | 45 ++-
paimon-python/pypaimon/read/table_read.py | 101 +++++--
.../pypaimon/tests/native_plan_incremental_test.py | 85 +++++-
.../pypaimon/tests/native_plan_integration_test.py | 323 ++++++++++++++++++++-
paimon-python/pypaimon/tests/native_read_test.py | 60 ++++
.../pypaimon/tests/streaming_table_scan_test.py | 36 +++
.../tests/test_read_builder_nested_projection.py | 29 ++
9 files changed, 686 insertions(+), 47 deletions(-)
diff --git a/paimon-python/pypaimon/read/native_plan.py
b/paimon-python/pypaimon/read/native_plan.py
index 6a15b41939..89ba3616cc 100644
--- a/paimon-python/pypaimon/read/native_plan.py
+++ b/paimon-python/pypaimon/read/native_plan.py
@@ -305,26 +305,36 @@ def _native_read_builder(table):
return builder
-def _configure_native_read_builder(builder, predicate, limit, projection):
- if projection is not None:
+def _configure_native_read_builder(builder, predicate, limit, projection,
+ nested_projection=None,
+ include_row_kind=False):
+ if nested_projection is not None:
+ builder = builder.with_nested_projection(nested_projection)
+ elif projection is not None:
builder = builder.with_projection(projection)
if predicate is not None:
builder = builder.with_filter(_predicate_to_native(predicate))
if limit is not None:
builder = builder.with_limit(limit)
+ if include_row_kind:
+ builder = builder.with_include_row_kind(True)
return builder
def native_read(table, splits, predicate: Optional[Predicate] = None,
limit: Optional[int] = None,
projection: Optional[List[str]] = None,
- blob_parallelism: Optional[int] = None):
+ blob_parallelism: Optional[int] = None,
+ nested_projection: Optional[List[List[str]]] = None,
+ include_row_kind: bool = False):
"""Read Rust ``Split`` objects into PyArrow ``RecordBatch`` objects."""
if not native_reader_available():
raise RuntimeError(
"read.native.enabled needs the pypaimon-rust native reader API")
builder = _configure_native_read_builder(
- _native_read_builder(table), predicate, limit, projection)
+ _native_read_builder(table), predicate, limit, projection,
+ nested_projection=nested_projection,
+ include_row_kind=include_row_kind)
if blob_parallelism is not None:
builder = builder.with_blob_parallelism(blob_parallelism)
reader = builder.new_read()
diff --git a/paimon-python/pypaimon/read/stream_read_builder.py
b/paimon-python/pypaimon/read/stream_read_builder.py
index 4ea4fbe694..0a2753eebe 100644
--- a/paimon-python/pypaimon/read/stream_read_builder.py
+++ b/paimon-python/pypaimon/read/stream_read_builder.py
@@ -26,10 +26,10 @@ from typing import Callable, List, Optional, Set
from pypaimon.common.predicate import Predicate
from pypaimon.common.predicate_builder import PredicateBuilder
+from pypaimon.read.read_builder import ReadBuilder
from pypaimon.read.streaming_table_scan import AsyncStreamingTableScan
from pypaimon.read.table_read import TableRead
from pypaimon.schema.data_types import DataField
-from pypaimon.table.special_fields import SpecialFields
class StreamReadBuilder:
@@ -115,6 +115,8 @@ class StreamReadBuilder:
def new_streaming_scan(self) -> AsyncStreamingTableScan:
"""Create a new AsyncStreamingTableScan with this builder's
settings."""
+ projection = self._projection_builder()
+ projection._validate_map_key_filter()
scan = AsyncStreamingTableScan(
table=self.table,
predicate=self._predicate,
@@ -122,30 +124,38 @@ class StreamReadBuilder:
bucket_filter=self._bucket_filter,
consumer_id=self._consumer_id
)
- scan._read_type = self.read_type()
+ scan._read_type = projection._scan_read_type()
return scan
def new_read(self) -> TableRead:
"""Create a new TableRead with this builder's settings."""
+ projection = self._projection_builder()
+ projection._validate_map_key_filter()
return TableRead(
table=self.table,
predicate=self._predicate,
- read_type=self.read_type(),
+ read_type=projection.read_type(),
+ nested_name_paths=projection._nested_name_paths(),
include_row_kind=self._include_row_kind
)
def new_predicate_builder(self) -> PredicateBuilder:
"""Create a PredicateBuilder for building filter predicates."""
- return PredicateBuilder(self.read_type())
+ return self._projection_builder().new_predicate_builder()
def read_type(self) -> List[DataField]:
"""Get the read schema fields, applying projection if set."""
- table_fields = self.table.fields
-
- if not self._projection:
- return table_fields
- else:
- if self.table.options.row_tracking_enabled():
- table_fields =
SpecialFields.row_type_with_row_tracking(table_fields)
- field_map = {field.name: field for field in table_fields}
- return [field_map[name] for name in self._projection if name in
field_map]
+ return self._projection_builder().read_type()
+
+ def _nested_name_paths(self) -> Optional[List[List[str]]]:
+ """Return nested paths resolved with the batch read-builder rules."""
+ return self._projection_builder()._nested_name_paths()
+
+ def _projection_builder(self) -> ReadBuilder:
+ """Share projection and validation semantics with batch reads."""
+ builder = ReadBuilder(self.table)
+ if self._projection is not None:
+ builder.with_projection(self._projection)
+ if self._predicate is not None:
+ builder.with_filter(self._predicate)
+ return builder
diff --git a/paimon-python/pypaimon/read/streaming_table_scan.py
b/paimon-python/pypaimon/read/streaming_table_scan.py
index ee0cb669ee..7cdc137e01 100644
--- a/paimon-python/pypaimon/read/streaming_table_scan.py
+++ b/paimon-python/pypaimon/read/streaming_table_scan.py
@@ -348,14 +348,57 @@ class AsyncStreamingTableScan:
def _create_initial_plan(self, snapshot: Snapshot) -> Plan:
"""Create a Plan for the initial full scan of the latest snapshot."""
auth_result = self.__auth_query()
- plan = self.__create_initial_plan_raw(snapshot, auth_result)
+ plan = None
+ if auth_result is None:
+ plan = self._try_native_plan(snapshot.id)
+ if plan is None:
+ plan = self.__create_initial_plan_raw(snapshot, auth_result)
return wrap_plan_with_auth(auth_result, plan)
def _create_delta_plan(self, snapshot: Snapshot) -> Plan:
"""Read new files from delta_manifest_list
(changelog-producer=none)."""
+ plan = self._try_native_plan(
+ snapshot.id, incremental_range=(snapshot.id - 1, snapshot.id))
+ if plan is not None:
+ return plan
manifest_files = self._manifest_list_manager.read_delta(snapshot)
return self._create_plan_from_manifests(manifest_files, snapshot.id)
+ def _try_native_plan(self, expected_snapshot_id: int,
+ incremental_range=None) -> Optional[Plan]:
+ """Plan an initial or delta streaming frame with pypaimon-rust.
+
+ An arbitrary Python bucket predicate cannot be represented by the Rust
+ planner, so sharded stream consumers retain the Python plan. Any native
+ setup or planning error is an optimization miss and falls back before
+ a plan is returned.
+ """
+ if (not self.table.options.native_plan_enabled()
+ or self._bucket_filter is not None):
+ return None
+ try:
+ from pypaimon.read.native_plan import native_plan
+ plan = native_plan(
+ self.table,
+ predicate=self.predicate,
+ projection=(
+ [field.name for field in self._read_type]
+ if self._read_type is not None else None),
+ incremental_range=incremental_range,
+ )
+ if plan.snapshot_id != expected_snapshot_id:
+ logging.warning(
+ "Native streaming plan resolved snapshot %s, expected %s; "
+ "falling back to Python planning",
+ plan.snapshot_id, expected_snapshot_id)
+ return None
+ return plan
+ except Exception as error:
+ logging.warning(
+ "Native streaming plan failed, falling back to Python
planning: %s",
+ error)
+ return None
+
def _create_changelog_plan(self, snapshot: Snapshot) -> Plan:
"""Read from changelog_manifest_list
(changelog-producer=input/full-compaction/lookup)."""
manifest_files = self._manifest_list_manager.read_changelog(snapshot)
diff --git a/paimon-python/pypaimon/read/table_read.py
b/paimon-python/pypaimon/read/table_read.py
index e697b51d2f..5595298422 100644
--- a/paimon-python/pypaimon/read/table_read.py
+++ b/paimon-python/pypaimon/read/table_read.py
@@ -24,6 +24,7 @@ from typing import Any, Callable, Dict, Iterator, List,
Optional
import pandas
import pyarrow
+import pyarrow.compute as pyarrow_compute
from pypaimon.common.options.core_options import CoreOptions
from pypaimon.common.predicate import Predicate
@@ -414,9 +415,6 @@ class TableRead:
"""Return Rust-read batches, or ``None`` when this read must fall
back."""
if not self.table.options.native_read_enabled():
return None
- # These Python-only output controls do not yet have native equivalents.
- if self.include_row_kind or self.nested_name_paths:
- return None
if self.table.options.file_format() not in _NATIVE_READ_FILE_FORMATS:
return None
if not splits:
@@ -446,13 +444,7 @@ class TableRead:
if streaming:
groups = self._native_split_groups(
rust_splits, parallelism, split_weights)
- read_kwargs = {
- 'predicate': self.predicate,
- 'limit': self.limit,
- 'projection': [field.name for field in self.read_type],
- }
- if blob_parallelism is not None:
- read_kwargs['blob_parallelism'] = blob_parallelism
+ read_kwargs = self._native_read_kwargs(blob_parallelism)
readers = []
try:
for group in groups:
@@ -484,13 +476,7 @@ class TableRead:
"Native read failed, falling back to the Python reader:
%s", e)
return None
try:
- read_kwargs = {
- 'predicate': self.predicate,
- 'limit': self.limit,
- 'projection': [field.name for field in self.read_type],
- }
- if blob_parallelism is not None:
- read_kwargs['blob_parallelism'] = blob_parallelism
+ read_kwargs = self._native_read_kwargs(blob_parallelism)
batches = native_read(self.table, rust_splits, **read_kwargs)
except Exception as e:
logger.warning(
@@ -498,6 +484,20 @@ class TableRead:
return None
return self._convert_native_batches(batches, schema)
+ def _native_read_kwargs(self, blob_parallelism=None):
+ kwargs = {
+ 'predicate': self.predicate,
+ 'limit': self.limit,
+ 'projection': [field.name for field in self.read_type],
+ }
+ if blob_parallelism is not None:
+ kwargs['blob_parallelism'] = blob_parallelism
+ if self.nested_name_paths:
+ kwargs['nested_projection'] = self.nested_name_paths
+ if self.include_row_kind:
+ kwargs['include_row_kind'] = True
+ return kwargs
+
@staticmethod
def _native_split_weight(split):
"""Estimate native split work from its physical data-file bytes."""
@@ -682,13 +682,7 @@ class TableRead:
if remaining_state.exhausted():
return []
try:
- read_kwargs = {
- 'predicate': self.predicate,
- 'limit': self.limit,
- 'projection': [field.name for field in self.read_type],
- }
- if blob_parallelism is not None:
- read_kwargs['blob_parallelism'] = blob_parallelism
+ read_kwargs = self._native_read_kwargs(blob_parallelism)
batches = native_read(self.table, rust_splits, **read_kwargs)
except Exception as e:
raise _NativeReadSetupError(str(e)) from e
@@ -701,6 +695,7 @@ class TableRead:
break
if allowed < batch.num_rows:
batch = batch.slice(0, allowed)
+ batch = self._prepare_native_batch(batch, schema)
batch = self._project_batch_to_output(batch)
result.append(self._try_to_pad_batch_by_schema(batch, schema))
if remaining_state.exhausted():
@@ -726,6 +721,7 @@ class TableRead:
continue
if remaining is not None and batch.num_rows > remaining:
batch = batch.slice(0, remaining)
+ batch = self._prepare_native_batch(batch, schema)
batch = self._project_batch_to_output(batch)
yield self._try_to_pad_batch_by_schema(batch, schema)
if remaining is not None:
@@ -737,6 +733,54 @@ class TableRead:
if close is not None:
close()
+ def _prepare_native_batch(self, batch, schema):
+ if self.include_row_kind and 'rowkind' in batch.schema.names:
+ index = batch.schema.get_field_index('rowkind')
+ batch = batch.rename_columns([
+ ROW_KIND_COLUMN if position == index else name
+ for position, name in enumerate(batch.schema.names)
+ ])
+ if self.nested_name_paths:
+ batch = self._flatten_native_nested_batch(batch, schema)
+ return batch
+
+ def _flatten_native_nested_batch(self, batch, schema):
+ arrays = []
+ fields = []
+ if self.include_row_kind:
+ index = batch.schema.get_field_index(ROW_KIND_COLUMN)
+ if index < 0:
+ raise ValueError('Native row-kind read did not return rowkind')
+ arrays.append(batch.column(index))
+ fields.append(schema.field(ROW_KIND_COLUMN))
+
+ for path, output in zip(self.nested_name_paths, self.read_type):
+ index = batch.schema.get_field_index(path[0])
+ if index < 0:
+ raise ValueError(
+ "Native nested read did not return top-level field %r"
+ % path[0])
+ array = batch.column(index)
+ parent_nulls = []
+ for name in path[1:]:
+ if pyarrow.types.is_struct(array.type):
+ parent_nulls.append(pyarrow_compute.is_null(array))
+ array = array.field(name)
+ elif pyarrow.types.is_map(array.type):
+ array = pyarrow_compute.map_lookup(array, name, 'first')
+ else:
+ raise ValueError(
+ "Native nested path %r cannot descend through %s"
+ % (path, array.type))
+ for nulls in reversed(parent_nulls):
+ array = pyarrow_compute.if_else(
+ nulls, pyarrow.scalar(None, type=array.type), array)
+ arrays.append(array)
+ fields.append(schema.field(output.name))
+
+ return pyarrow.RecordBatch.from_arrays(
+ arrays, schema=pyarrow.schema(fields))
+
def _arrow_batch_generator(self, splits: List[Split], schema:
pyarrow.Schema,
blob_parallelism: int = 1) ->
Iterator[pyarrow.RecordBatch]:
chunk_size = 65536
@@ -1452,11 +1496,14 @@ class TableRead:
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:
+ output_names = list(self._output_column_names)
+ if self.include_row_kind and ROW_KIND_COLUMN in batch.schema.names:
+ output_names.insert(0, ROW_KIND_COLUMN)
+ if batch.schema.names == output_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]
+ arrays = [batch.column(name_to_pos[name]) for name in output_names]
+ fields = [batch.schema.field(name_to_pos[name]) for name in
output_names]
return pyarrow.RecordBatch.from_arrays(
arrays, schema=pyarrow.schema(fields))
diff --git a/paimon-python/pypaimon/tests/native_plan_incremental_test.py
b/paimon-python/pypaimon/tests/native_plan_incremental_test.py
index 9dde912e44..afc46d5608 100644
--- a/paimon-python/pypaimon/tests/native_plan_incremental_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_incremental_test.py
@@ -18,6 +18,7 @@
"""Compare committed timestamp windows, including cross-snapshot change
events."""
import json
+import os
from contextlib import ExitStack
from unittest.mock import patch
@@ -321,21 +322,103 @@ def
test_incremental_reader_preserves_all_physical_row_kinds(catalog, native):
assert plan.snapshot_id == snapshot_id
expected.append((str(kind), kind))
builder = table.copy({'scan.native-plan.enabled': str(native).lower(),
+ 'read.native.enabled': str(native).lower(),
'incremental-between-timestamp':
'0,400'}).new_read_builder()
batch_scan = builder.new_scan()
with ExitStack() as stack:
if native:
stack.enter_context(patch.object(batch_scan.file_scanner, 'scan',
side_effect=AssertionError('native fallback')))
- plan = batch_scan.plan()
+ plan = batch_scan.plan()
+ if native:
+ # The fixture rewrites Parquet bytes after commit to manufacture all
+ # four physical kinds. Refresh the opaque split's file-size metadata;
+ # real committed files already have matching manifest sizes.
+ from pypaimon_rust.datafusion import Split as NativeSplit
+ for split in plan.splits():
+ _, (state,) = split._native_split.__reduce__()
+ native_state = json.loads(bytes(state))
+ for native_file, data_file in zip(
+ native_state['data_files'], split.files):
+ native_file['_FILE_SIZE'] = os.path.getsize(
+ data_file.file_path)
+ split._native_split = NativeSplit(
+ json.dumps(native_state).encode())
rows = list(builder.new_read().to_iterator(plan.splits()))
assert sorted((row.get_field(1), row.get_row_kind().value) for row in
rows) == expected
+ if native:
+ read = builder.new_read()
+ read.include_row_kind = True
+ with patch(
+ 'pypaimon.read.table_read.TableRead._create_split_read',
+ side_effect=AssertionError('native row-kind read fell back')):
+ arrow = read.to_arrow(plan.splits(), parallelism=2)
+ assert sorted(zip(
+ arrow.column('v').to_pylist(),
+ arrow.column('_row_kind').to_pylist(),
+ )) == [('0', '+I'), ('1', '-U'), ('2', '+U'), ('3', '-D')]
# Initial streaming bootstrap remains a merged snapshot, where the last -D
removes the key.
initial =
scan._create_initial_plan(table.snapshot_manager().get_snapshot_by_id(4))
assert all(not split.is_streaming for split in initial.splits())
assert
list(table.new_read_builder().new_read().to_iterator(initial.splits())) == []
[email protected]_plan
[email protected](
+ not native_method_available('ReadBuilder', 'with_nested_projection'),
+ reason='pypaimon_rust nested native reader API required')
+def test_stream_read_builder_combines_native_nested_projection_and_row_kind(
+ catalog):
+ from pypaimon.read.streaming_table_scan import AsyncStreamingTableScan
+
+ schema = pa.schema([
+ ('k', pa.int64()),
+ ('payload', pa.struct([
+ ('score', pa.int32()),
+ ('ignored', pa.string()),
+ ])),
+ ])
+ catalog.create_table(
+ 'default.stream_nested',
+ Schema.from_pyarrow_schema(schema, options={'bucket': '-1'}),
+ False,
+ )
+ table = catalog.get_table('default.stream_nested')
+ 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([
+ {'k': 1, 'payload': {'score': 10, 'ignored': 'a'}},
+ {'k': 2, 'payload': None},
+ ], schema=schema))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+
+ native_table = table.copy({
+ 'scan.native-plan.enabled': 'true',
+ 'read.native.enabled': 'true',
+ })
+ builder = (native_table.new_stream_read_builder()
+ .with_projection(['payload.score', 'k'])
+ .with_include_row_kind())
+ # Use the same delta-plan primitive as the streaming loop without polling.
+ scan = builder.new_streaming_scan()
+ assert isinstance(scan, AsyncStreamingTableScan)
+ snapshot = native_table.snapshot_manager().get_latest_snapshot()
+ plan = scan._create_delta_plan(snapshot)
+ with patch(
+ 'pypaimon.read.table_read.TableRead._create_split_read',
+ side_effect=AssertionError('streaming nested native read fell
back')):
+ rows = builder.new_read().to_arrow(plan.splits()).to_pylist()
+
+ assert rows == [
+ {'_row_kind': '+I', 'payload_score': 10, 'k': 1},
+ {'_row_kind': '+I', 'payload_score': None, 'k': 2},
+ ]
+
+
def test_streaming_reader_honors_explicit_split_deletion_vector(catalog,
native, tmp_path):
from pypaimon.deletionvectors.bitmap_deletion_vector import
BitmapDeletionVector
from pypaimon.table.source.deletion_file import DeletionFile
diff --git a/paimon-python/pypaimon/tests/native_plan_integration_test.py
b/paimon-python/pypaimon/tests/native_plan_integration_test.py
index 41858d3997..35eeb9edbc 100644
--- a/paimon-python/pypaimon/tests/native_plan_integration_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_integration_test.py
@@ -16,6 +16,8 @@
# under the License.
import datetime
+import importlib.util
+import io
import os
import pickle
import tempfile
@@ -31,6 +33,7 @@ from pypaimon.read.native_plan import (
native_family_search_modes_available, native_method_available, native_read,
native_reader_available,
)
+from pypaimon.schema.schema_change import SchemaChange
from pypaimon.table.row.blob import BlobDescriptor, BlobRef
from pypaimon.utils.range import Range
@@ -51,6 +54,23 @@ def _has_native_row_ranges():
return hasattr(ReadBuilder, 'with_row_ranges')
+def _mosaic_supports_nested_row():
+ if importlib.util.find_spec('mosaic') is None:
+ return False
+
+ import mosaic
+
+ schema = pa.schema([('nested', pa.struct([('value', pa.int32())]))])
+ table = pa.Table.from_pylist([{'nested': {'value': 1}}], schema=schema)
+ try:
+ mosaic.write_table(table, io.BytesIO())
+ except RuntimeError as error:
+ if 'unsupported DataType: Struct' in str(error):
+ return False
+ raise
+ return True
+
+
@pytest.mark.native_plan
@unittest.skipUnless(_has_native_planner(),
"pypaimon_rust with split-planning API not installed")
@@ -262,11 +282,312 @@ class NativePlanIntegrationTest(unittest.TestCase):
with patch(
'pypaimon.read.table_read.TableRead._create_split_read',
side_effect=AssertionError('Python reader was used')):
+ native_read_instance = builder.new_read()
native = sorted(
- builder.new_read().to_arrow(plan.splits()).to_pylist(),
+ native_read_instance.to_arrow(plan.splits()).to_pylist(),
key=lambda row: row['k'])
+ native_read_instance.include_row_kind = True
+ with_row_kind = native_read_instance.to_arrow(
+ plan.splits()).to_pylist()
self.assertEqual(native, normal)
+ self.assertEqual(
+ {row['_row_kind'] for row in with_row_kind}, {'+I'})
+ self.assertEqual(
+ sorted(({key: value for key, value in row.items()
+ if key != '_row_kind'} for row in with_row_kind),
+ key=lambda row: row['k']),
+ normal,
+ )
+
+ @unittest.skipUnless(
+ native_method_available('ReadBuilder', 'with_nested_projection'),
+ "pypaimon-rust nested native reader API not installed")
+ def test_native_read_nested_rows_and_map_keys_across_formats(self):
+ base_fields = [
+ ('id', pa.int64()),
+ ('payload', pa.struct([
+ ('details', pa.struct([
+ ('score', pa.int32()),
+ ('label', pa.string()),
+ ])),
+ ('ignored', pa.string()),
+ ])),
+ ]
+ rows = [
+ {'id': 1,
+ 'payload': {'details': {'score': 7, 'label': 'a'}, 'ignored':
'x'},
+ 'attrs': {'selected': 10, 'other': 11},
+ 'top.with.dot': 'first'},
+ {'id': 2, 'payload': None, 'attrs': None, 'top.with.dot':
'second'},
+ {'id': 3,
+ 'payload': {'details': None, 'ignored': 'z'},
+ 'attrs': {},
+ 'top.with.dot': 'third'},
+ ]
+ formats = ['parquet', 'orc', 'avro', 'row']
+ if _mosaic_supports_nested_row():
+ formats.append('mosaic')
+ for file_format in formats:
+ with self.subTest(file_format=file_format):
+ # PyPaimon's Avro writer has no MAP conversion; it still covers
+ # recursive ROW pruning. Other formats also exercise literal
+ # MAP-key extraction from the native batch.
+ include_map = file_format != 'avro'
+ include_dotted_name = file_format != 'avro'
+ nested_schema = pa.schema(
+ base_fields
+ + ([('attrs', pa.map_(pa.string(), pa.int32()))]
+ if include_map else [])
+ + ([('top.with.dot', pa.string())]
+ if include_dotted_name else []))
+ projection = ['payload.details.score']
+ if include_map:
+ projection.append("attrs['selected']")
+ if include_dotted_name:
+ projection.append('top.with.dot')
+ projection.append('id')
+ expected = [
+ {'payload_details_score': 7,
+ 'top.with.dot': 'first', 'id': 1},
+ {'payload_details_score': None,
+ 'top.with.dot': 'second', 'id': 2},
+ {'payload_details_score': None,
+ 'top.with.dot': 'third', 'id': 3},
+ ]
+ if include_map:
+ for expected_row, value in zip(expected, [10, None, None]):
+ expected_row['attrs_selected'] = value
+ if not include_dotted_name:
+ for expected_row in expected:
+ expected_row.pop('top.with.dot')
+ name = 'native_nested_' + file_format
+ options = {'file.format': file_format, 'bucket': '-1'}
+ if file_format == 'parquet':
+ options.update({
+ 'fields.attrs.map.storage-layout': 'shared-shredding',
+ 'fields.attrs.map.shared-shredding.max-columns': '2',
+ })
+ self.cat.create_table(
+ 'default.' + name,
+ Schema.from_pyarrow_schema(
+ nested_schema, options=options),
+ False,
+ )
+ table = self.cat.get_table('default.' + name)
+ write_builder = table.new_batch_write_builder()
+ write = write_builder.new_write()
+ write.write_arrow(pa.Table.from_pylist(rows,
schema=nested_schema))
+ write_builder.new_commit().commit(write.prepare_commit())
+ write.close()
+
+ native_table = table.copy({
+ 'scan.native-plan.enabled': 'true',
+ 'read.native.enabled': 'true',
+ })
+ builder = native_table.new_read_builder().with_projection(
+ projection)
+ plan = builder.new_scan().plan()
+ with patch(
+
'pypaimon.read.table_read.TableRead._create_split_read',
+ side_effect=AssertionError(
+ 'nested native read fell back to Python')):
+ actual = builder.new_read().to_arrow(
+ plan.splits(), parallelism=2).to_pylist()
+
+ self.assertEqual(actual, expected)
+
+ @unittest.skipUnless(
+ native_method_available('ReadBuilder', 'with_nested_projection'),
+ "pypaimon-rust nested native reader API not installed")
+ def test_native_read_nested_projection_after_primary_key_merge(self):
+ nested_schema = pa.schema([
+ ('id', pa.int64()),
+ ('payload', pa.struct([
+ ('score', pa.int32()),
+ ('ignored', pa.string()),
+ ])),
+ ])
+ self.cat.create_table(
+ 'default.native_nested_pk',
+ Schema.from_pyarrow_schema(
+ nested_schema, primary_keys=['id'], options={'bucket': '1'}),
+ False,
+ )
+ table = self.cat.get_table('default.native_nested_pk')
+ for rows in (
+ [{'id': 1, 'payload': {'score': 1, 'ignored': 'old'}},
+ {'id': 2, 'payload': {'score': 2, 'ignored': 'keep'}}],
+ [{'id': 1, 'payload': {'score': 10, 'ignored': 'new'}}],
+ ):
+ write_builder = table.new_batch_write_builder()
+ write = write_builder.new_write()
+ write.write_arrow(pa.Table.from_pylist(rows, schema=nested_schema))
+ write_builder.new_commit().commit(write.prepare_commit())
+ write.close()
+
+ native_table = table.copy({
+ 'scan.native-plan.enabled': 'true',
+ 'read.native.enabled': 'true',
+ })
+ builder = native_table.new_read_builder().with_projection(
+ ['id', 'payload.score'])
+ plan = builder.new_scan().plan()
+ with patch(
+ 'pypaimon.read.table_read.TableRead._create_split_read',
+ side_effect=AssertionError('nested PK native read fell back')):
+ actual = sorted(
+ builder.new_read().to_arrow(plan.splits()).to_pylist(),
+ key=lambda row: row['id'])
+
+ self.assertEqual(actual, [
+ {'id': 1, 'payload_score': 10},
+ {'id': 2, 'payload_score': 2},
+ ])
+
+ @unittest.skipUnless(
+ native_method_available('ReadBuilder', 'with_nested_projection'),
+ "pypaimon-rust nested native reader API not installed")
+ def test_native_nested_projection_after_partial_update_merge(self):
+ nested_schema = pa.schema([
+ ('id', pa.int64()),
+ ('payload', pa.struct([
+ ('score', pa.int32()),
+ ('ignored', pa.string()),
+ ])),
+ ])
+ self.cat.create_table(
+ 'default.native_nested_partial_update',
+ Schema.from_pyarrow_schema(
+ nested_schema,
+ primary_keys=['id'],
+ options={'bucket': '1', 'merge-engine': 'partial-update'},
+ ),
+ False,
+ )
+ table = self.cat.get_table('default.native_nested_partial_update')
+ for rows in (
+ [{'id': 1, 'payload': {'score': 1, 'ignored': 'old'}},
+ {'id': 2, 'payload': {'score': 2, 'ignored': 'keep'}}],
+ [{'id': 1, 'payload': {'score': 10, 'ignored': 'new'}}],
+ ):
+ write_builder = table.new_batch_write_builder()
+ table_write = write_builder.new_write()
+ table_write.write_arrow(pa.Table.from_pylist(rows,
schema=nested_schema))
+ write_builder.new_commit().commit(table_write.prepare_commit())
+ table_write.close()
+
+ native_table = table.copy({
+ 'scan.native-plan.enabled': 'true',
+ 'read.native.enabled': 'true',
+ })
+ builder = native_table.new_read_builder().with_projection(
+ ['id', 'payload.score'])
+ plan = builder.new_scan().plan()
+ with patch(
+ 'pypaimon.read.table_read.TableRead._create_split_read',
+ side_effect=AssertionError(
+ 'nested partial-update native read fell back')):
+ actual = sorted(
+ builder.new_read().to_arrow(plan.splits()).to_pylist(),
+ key=lambda row: row['id'])
+
+ self.assertEqual(actual, [
+ {'id': 1, 'payload_score': 10},
+ {'id': 2, 'payload_score': 2},
+ ])
+
+ @unittest.skipUnless(
+ native_method_available('ReadBuilder', 'with_nested_projection'),
+ "pypaimon-rust nested native reader API not installed")
+ def
test_native_nested_projection_across_schema_rename_filter_and_limit(self):
+ name = 'default.native_nested_evolution'
+ old_schema = pa.schema([
+ ('id', pa.int64()),
+ ('payload', pa.struct([
+ ('old_score', pa.int32()),
+ ('ignored', pa.string()),
+ ])),
+ ])
+ self.cat.create_table(
+ name,
+ Schema.from_pyarrow_schema(old_schema, options={
+ 'bucket': '-1',
+ 'data-evolution.enabled': 'true',
+ 'row-tracking.enabled': 'true',
+ 'file.format': 'parquet',
+ }),
+ False,
+ )
+
+ def write(table, schema, rows):
+ write_builder = table.new_batch_write_builder()
+ table_write = write_builder.new_write()
+ table_write.write_arrow(pa.Table.from_pylist(rows, schema=schema))
+ write_builder.new_commit().commit(table_write.prepare_commit())
+ table_write.close()
+
+ table = self.cat.get_table(name)
+ write(table, old_schema, [
+ {'id': 1, 'payload': {'old_score': 10, 'ignored': 'old-1'}},
+ {'id': 2, 'payload': {'old_score': 20, 'ignored': 'old-2'}},
+ ])
+ self.cat.alter_table(
+ name,
+ [SchemaChange.rename_column(['payload', 'old_score'], 'score')],
+ False,
+ )
+ new_schema = pa.schema([
+ ('id', pa.int64()),
+ ('payload', pa.struct([
+ ('score', pa.int32()),
+ ('ignored', pa.string()),
+ ])),
+ ])
+ table = self.cat.get_table(name)
+ write(table, new_schema, [
+ {'id': 3, 'payload': {'score': 30, 'ignored': 'new-3'}},
+ {'id': 4, 'payload': None},
+ ])
+
+ native_table = table.copy({
+ 'scan.native-plan.enabled': 'true',
+ 'read.native.enabled': 'true',
+ 'read.parallelism': '3',
+ })
+ builder = native_table.new_read_builder().with_projection(
+ ['payload.score', 'id'])
+ predicate = builder.new_predicate_builder().greater_or_equal('id', 2)
+ builder.with_filter(predicate)
+ plan = builder.new_scan().plan()
+ with patch(
+ 'pypaimon.read.table_read.TableRead._create_split_read',
+ side_effect=AssertionError(
+ 'evolved nested native read fell back to Python')):
+ actual = sorted(
+ builder.new_read().to_arrow(
+ plan.splits(), parallelism=3).to_pylist(),
+ key=lambda row: row['id'],
+ )
+
+ self.assertEqual(actual, [
+ {'payload_score': 20, 'id': 2},
+ {'payload_score': 30, 'id': 3},
+ {'payload_score': None, 'id': 4},
+ ])
+
+ limited = native_table.new_read_builder().with_projection(
+ ['payload.score', 'id']).with_limit(2)
+ limited_plan = limited.new_scan().plan()
+ with patch(
+ 'pypaimon.read.table_read.TableRead._create_split_read',
+ side_effect=AssertionError(
+ 'limited nested native read fell back to Python')):
+ limited_rows = limited.new_read().to_arrow(
+ limited_plan.splits(), parallelism=3).to_pylist()
+ self.assertEqual(len(limited_rows), 2)
+ self.assertTrue(all(set(row) == {'payload_score', 'id'}
+ for row in limited_rows))
def test_append_distribution_matches_interleaved_partition_buckets(self):
self.schema = pa.schema([('k', pa.int64()), ('v', pa.string()), ('p',
pa.string())])
diff --git a/paimon-python/pypaimon/tests/native_read_test.py
b/paimon-python/pypaimon/tests/native_read_test.py
index cae8c4deb5..8376a878d7 100644
--- a/paimon-python/pypaimon/tests/native_read_test.py
+++ b/paimon-python/pypaimon/tests/native_read_test.py
@@ -85,6 +85,66 @@ def
test_native_read_consumes_retained_rust_splits_and_enforces_limit():
)
+def test_native_read_flattens_nested_rows_and_map_keys_with_parent_nulls():
+ read = _table_read()
+ read.read_type = [
+ DataField(2, 'payload_score', AtomicType('INT')),
+ DataField(3, 'attrs_selected', AtomicType('INT')),
+ ]
+ read._output_column_names = [field.name for field in read.read_type]
+ read.nested_name_paths = [
+ ['payload', 'details', 'score'],
+ ['attrs', 'selected'],
+ ]
+ split = _Split()
+ split._native_split = object()
+ payload_type = pa.struct([
+ ('details', pa.struct([('score', pa.int32()), ('ignored',
pa.string())])),
+ ('ignored', pa.string()),
+ ])
+ batch = pa.record_batch([
+ pa.array([
+ {'details': {'score': 7, 'ignored': 'x'}, 'ignored': 'x'},
+ None,
+ {'details': None, 'ignored': 'z'},
+ ], type=payload_type),
+ pa.array([
+ [('selected', 10), ('other', 11)],
+ None,
+ [],
+ ], type=pa.map_(pa.string(), pa.int32())),
+ ], names=['payload', 'attrs'])
+
+ with patch('pypaimon.read.native_plan.native_read',
+ return_value=[batch]) as native:
+ result = read.to_arrow([split])
+
+ assert result.to_pydict() == {
+ 'payload_score': [7, None, None],
+ 'attrs_selected': [10, None, None],
+ }
+ assert native.call_args.kwargs['nested_projection'] ==
read.nested_name_paths
+
+
+def test_native_read_preserves_physical_row_kinds():
+ read = _table_read()
+ read.include_row_kind = True
+ split = _Split()
+ split._native_split = object()
+ batch = pa.record_batch([
+ pa.array(['+I', '-U', '+U', '-D']),
+ pa.array([1, 2, 3, 4], type=pa.int32()),
+ ], names=['rowkind', 'id'])
+
+ with patch('pypaimon.read.native_plan.native_read',
+ return_value=[batch]) as native:
+ result = read.to_arrow([split])
+
+ assert result.schema.names == ['_row_kind', 'id']
+ assert result.column('_row_kind').to_pylist() == ['+I', '-U', '+U', '-D']
+ assert native.call_args.kwargs['include_row_kind'] is True
+
+
def test_native_read_falls_back_for_transformed_python_split():
read = _table_read()
schema = pa.schema([('id', pa.int32())])
diff --git a/paimon-python/pypaimon/tests/streaming_table_scan_test.py
b/paimon-python/pypaimon/tests/streaming_table_scan_test.py
index a1b0ffe58e..630f1324af 100644
--- a/paimon-python/pypaimon/tests/streaming_table_scan_test.py
+++ b/paimon-python/pypaimon/tests/streaming_table_scan_test.py
@@ -51,6 +51,7 @@ def _create_mock_table(latest_snapshot_id: int = 5):
table.options.data_evolution_enabled.return_value = False
table.options.deletion_vectors_enabled.return_value = False
table.options.changelog_producer.return_value = ChangelogProducer.NONE
+ table.options.native_plan_enabled.return_value = False
table.field_names = ['col1', 'col2']
table.trimmed_primary_keys = []
table.partition_keys = []
@@ -93,6 +94,41 @@ class AsyncStreamingTableScanTest(unittest.TestCase):
self.assertIsInstance(plan, Plan)
self.assertEqual(scan.next_snapshot_id, 6)
+ @patch('pypaimon.read.streaming_table_scan.ManifestListManager')
+ @patch('pypaimon.read.streaming_table_scan.ManifestFileManager')
+ @patch('pypaimon.read.native_plan.native_plan')
+ def test_delta_scan_uses_native_plan_with_projection(
+ self, native_plan, _manifest_files, _manifest_lists):
+ table, _ = _create_mock_table()
+ table.options.native_plan_enabled.return_value = True
+ native_plan.return_value = Plan([], snapshot_id=5)
+ scan = AsyncStreamingTableScan(table, predicate=Mock())
+ scan._read_type = [Mock(name='payload'), Mock(name='id')]
+ scan._read_type[0].name = 'payload'
+ scan._read_type[1].name = 'id'
+
+ plan = scan._create_delta_plan(_create_mock_snapshot(5))
+
+ self.assertIs(plan, native_plan.return_value)
+ self.assertEqual(native_plan.call_args.kwargs['incremental_range'],
(4, 5))
+ self.assertEqual(
+ native_plan.call_args.kwargs['projection'], ['payload', 'id'])
+
+ @patch('pypaimon.read.streaming_table_scan.ManifestListManager')
+ @patch('pypaimon.read.streaming_table_scan.ManifestFileManager')
+ @patch('pypaimon.read.native_plan.native_plan')
+ def test_bucket_filtered_delta_scan_keeps_python_planner(
+ self, native_plan, _manifest_files, manifest_lists):
+ table, _ = _create_mock_table()
+ table.options.native_plan_enabled.return_value = True
+ manifest_lists.return_value.read_delta.return_value = []
+ scan = AsyncStreamingTableScan(table, bucket_filter=lambda bucket:
True)
+
+ plan = scan._create_delta_plan(_create_mock_snapshot(5))
+
+ self.assertEqual(plan.snapshot_id, 5)
+ native_plan.assert_not_called()
+
@patch('pypaimon.read.streaming_table_scan.ManifestListManager')
@patch('pypaimon.read.streaming_table_scan.FileScanner')
def test_stream_skips_non_append_commits(self, MockStartingScanner,
MockManifestListManager):
diff --git
a/paimon-python/pypaimon/tests/test_read_builder_nested_projection.py
b/paimon-python/pypaimon/tests/test_read_builder_nested_projection.py
index 61e7396e40..4897bd96b1 100644
--- a/paimon-python/pypaimon/tests/test_read_builder_nested_projection.py
+++ b/paimon-python/pypaimon/tests/test_read_builder_nested_projection.py
@@ -141,5 +141,34 @@ class
ReadBuilderProjectionFieldIdTest(_ReadBuilderTestBase):
self.assertEqual(leaf_ids, [sub_v.id, sub_x.id])
+class StreamReadBuilderNestedProjectionTest(_ReadBuilderTestBase):
+
+ def test_stream_builder_matches_batch_nested_projection(self):
+ projection = [
+ 'mv.latest_version',
+ "attrs['key.with.dots']",
+ 'pk',
+ ]
+ batch = self.table.new_read_builder().with_projection(projection)
+ stream =
self.table.new_stream_read_builder().with_projection(projection)
+
+ self.assertEqual(
+ [field.name for field in batch.read_type()],
+ [field.name for field in stream.read_type()],
+ )
+ self.assertEqual(
+ batch._nested_name_paths(),
+ stream._nested_name_paths(),
+ )
+ self.assertEqual(
+ [field.name for field in batch.new_scan()._read_type],
+ [field.name for field in stream.new_streaming_scan()._read_type],
+ )
+
+ table_read = stream.with_include_row_kind().new_read()
+ self.assertEqual(batch._nested_name_paths(),
table_read.nested_name_paths)
+ self.assertTrue(table_read.include_row_kind)
+
+
if __name__ == '__main__':
unittest.main()
