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 54545a9ae5 [python] Fix Ray read_paimon dropping nested projection
(reads nested leaves as NULL) (#8269)
54545a9ae5 is described below
commit 54545a9ae5f5e98eba1bbdf86415d979bfe2b539
Author: chaoyang <[email protected]>
AuthorDate: Fri Jun 19 08:56:51 2026 +0800
[python] Fix Ray read_paimon dropping nested projection (reads nested
leaves as NULL) (#8269)
---
.../pypaimon/read/datasource/ray_datasource.py | 10 ++++++-
.../pypaimon/read/datasource/split_provider.py | 23 +++++++++++++-
paimon-python/pypaimon/read/table_read.py | 1 +
paimon-python/pypaimon/tests/ray_data_test.py | 35 ++++++++++++++++++++++
.../pypaimon/tests/ray_integration_test.py | 25 ++++++++++++++++
5 files changed, 92 insertions(+), 2 deletions(-)
diff --git a/paimon-python/pypaimon/read/datasource/ray_datasource.py
b/paimon-python/pypaimon/read/datasource/ray_datasource.py
index 25c6109259..a5ac10bc10 100644
--- a/paimon-python/pypaimon/read/datasource/ray_datasource.py
+++ b/paimon-python/pypaimon/read/datasource/ray_datasource.py
@@ -124,6 +124,7 @@ class RayDatasource(Datasource):
table = self._split_provider.table()
predicate = self._split_provider.predicate()
read_type = self._split_provider.read_type()
+ nested_name_paths = self._split_provider.nested_name_paths()
splits = self._split_provider.splits()
limit = self._split_provider.limit()
if not splits:
@@ -148,11 +149,17 @@ class RayDatasource(Datasource):
read_type=read_type,
schema=schema,
limit=limit,
+ nested_name_paths=nested_name_paths,
) -> Iterable[pyarrow.Table]:
"""Read function that will be executed by Ray workers."""
from pypaimon.read.table_read import TableRead
+ # nested_name_paths must be forwarded so a nested-leaf projection
+ # widens to the parent struct and extracts the leaves; without it
+ # the worker treats the flattened leaf names as missing top-level
+ # columns and reads every projected leaf as NULL.
worker_table_read = TableRead(
- table, predicate, read_type, limit=limit)
+ table, predicate, read_type, limit=limit,
+ nested_name_paths=nested_name_paths)
batch_reader = worker_table_read.to_arrow_batch_reader(splits)
has_data = False
@@ -179,6 +186,7 @@ class RayDatasource(Datasource):
read_type=read_type,
schema=schema,
limit=limit,
+ nested_name_paths=nested_name_paths,
)
read_tasks = []
diff --git a/paimon-python/pypaimon/read/datasource/split_provider.py
b/paimon-python/pypaimon/read/datasource/split_provider.py
index 430eeb7cac..eb953c573e 100644
--- a/paimon-python/pypaimon/read/datasource/split_provider.py
+++ b/paimon-python/pypaimon/read/datasource/split_provider.py
@@ -67,6 +67,17 @@ class SplitProvider(ABC):
"""
return None
+ def nested_name_paths(self) -> Optional[List[List[str]]]:
+ """Parallel name paths for a nested-leaf projection, or ``None``.
+
+ Forwarded to the per-task ``TableRead`` so a projection like
+ ``['mv.latest_value.x']`` is read by widening to the parent struct and
+ extracting the requested leaves. Without it the worker treats the
+ flattened leaf names as missing top-level columns and reads every
+ projected leaf as NULL.
+ """
+ return None
+
class CatalogSplitProvider(SplitProvider):
"""Plan splits from a fully-qualified table identifier and catalog options.
@@ -124,6 +135,7 @@ class CatalogSplitProvider(SplitProvider):
self._table_cached = None
self._splits_cached = None
self._read_type_cached = None
+ self._nested_name_paths_cached = None
def _ensure_table(self):
if self._table_cached is None:
@@ -154,6 +166,7 @@ class CatalogSplitProvider(SplitProvider):
if self._limit is not None:
rb = rb.with_limit(self._limit)
self._read_type_cached = rb.read_type()
+ self._nested_name_paths_cached = rb._nested_name_paths()
self._splits_cached = rb.new_scan().plan().splits()
@property
@@ -171,6 +184,10 @@ class CatalogSplitProvider(SplitProvider):
self._ensure_planned()
return self._read_type_cached
+ def nested_name_paths(self) -> Optional[List[List[str]]]:
+ self._ensure_planned()
+ return self._nested_name_paths_cached
+
def predicate(self):
return self._predicate
@@ -190,12 +207,13 @@ class PreResolvedSplitProvider(SplitProvider):
"""
def __init__(self, table, splits: List[Split], read_type, predicate=None,
- limit: Optional[int] = None):
+ limit: Optional[int] = None, nested_name_paths=None):
self._table = table
self._splits = splits
self._read_type = read_type
self._predicate = predicate
self._limit = limit
+ self._nested_name_paths = nested_name_paths
def table(self):
return self._table
@@ -206,6 +224,9 @@ class PreResolvedSplitProvider(SplitProvider):
def read_type(self):
return self._read_type
+ def nested_name_paths(self) -> Optional[List[List[str]]]:
+ return self._nested_name_paths
+
def predicate(self):
return self._predicate
diff --git a/paimon-python/pypaimon/read/table_read.py
b/paimon-python/pypaimon/read/table_read.py
index 3731280577..67159a69c7 100644
--- a/paimon-python/pypaimon/read/table_read.py
+++ b/paimon-python/pypaimon/read/table_read.py
@@ -530,6 +530,7 @@ class TableRead:
read_type=self.read_type,
predicate=self.predicate,
limit=self.limit,
+ nested_name_paths=self.nested_name_paths,
)
)
ds = ray.data.read_datasource(
diff --git a/paimon-python/pypaimon/tests/ray_data_test.py
b/paimon-python/pypaimon/tests/ray_data_test.py
index 990f23a5ca..6d00d13ff9 100644
--- a/paimon-python/pypaimon/tests/ray_data_test.py
+++ b/paimon-python/pypaimon/tests/ray_data_test.py
@@ -834,6 +834,41 @@ class RayDataTest(unittest.TestCase):
"Blob data column should match"
)
+ def test_to_ray_with_nested_projection(self):
+ """to_ray() respects a nested-leaf projection.
+
+ Sibling of the read_paimon() nested-projection test: this exercises
+ the PreResolvedSplitProvider entry point (TableRead.to_ray), which
+ must also forward nested_name_paths to the worker TableRead. Without
+ it the worker treats the flattened leaf name as a missing top-level
+ column and reads the projected leaf as NULL.
+ """
+ inner = pa.struct([('a', pa.int64()), ('b', pa.string())])
+ pa_schema = pa.schema([('id', pa.int64()), ('payload', inner)])
+ schema = Schema.from_pyarrow_schema(pa_schema)
+ self.catalog.create_table('default.test_ray_nested_proj', schema,
False)
+ table = self.catalog.get_table('default.test_ray_nested_proj')
+
+ write_builder = table.new_batch_write_builder()
+ writer = write_builder.new_write()
+ writer.write_arrow(pa.Table.from_pylist(
+ [{'id': 1, 'payload': {'a': 10, 'b': 'x'}},
+ {'id': 2, 'payload': {'a': 20, 'b': 'y'}}],
+ schema=pa_schema))
+ commit = write_builder.new_commit()
+ commit.commit(writer.prepare_commit())
+ writer.close()
+
+ read_builder = table.new_read_builder().with_projection(['id',
'payload.a'])
+ table_read = read_builder.new_read()
+ splits = read_builder.new_scan().plan().splits()
+
+ ray_dataset = table_read.to_ray(splits, override_num_blocks=1)
+ rows = {r['id']: r for r in ray_dataset.take_all()}
+ self.assertEqual(set(rows.keys()), {1, 2})
+ self.assertEqual(rows[1]['payload_a'], 10)
+ self.assertEqual(rows[2]['payload_a'], 20)
+
if __name__ == '__main__':
unittest.main()
diff --git a/paimon-python/pypaimon/tests/ray_integration_test.py
b/paimon-python/pypaimon/tests/ray_integration_test.py
index 225dc710d6..c79b62e9fb 100644
--- a/paimon-python/pypaimon/tests/ray_integration_test.py
+++ b/paimon-python/pypaimon/tests/ray_integration_test.py
@@ -119,6 +119,31 @@ class RayIntegrationTest(unittest.TestCase):
self.assertEqual(set(df.columns), {'id', 'name'})
self.assertEqual(len(df), 2)
+ def test_read_paimon_with_nested_projection(self):
+ """read_paimon() respects a nested-leaf projection.
+
+ Regression for the worker-side TableRead being rebuilt without
+ nested_name_paths: a projection like ['payload.a'] used to read every
+ nested leaf as NULL because the worker treated the flattened leaf name
+ as a missing top-level column.
+ """
+ from pypaimon.ray import read_paimon
+
+ inner = pa.struct([('a', pa.int64()), ('b', pa.string())])
+ pa_schema = pa.schema([('id', pa.int32()), ('payload', inner)])
+ identifier = self._create_and_populate_table(
+ 'test_read_nested_proj', pa_schema,
+ {'id': [1, 2],
+ 'payload': [{'a': 10, 'b': 'x'}, {'a': 20, 'b': 'y'}]},
+ )
+
+ ds = read_paimon(identifier, self.catalog_options,
+ projection=['id', 'payload.a'])
+ rows = {r['id']: r for r in ds.take_all()}
+ self.assertEqual(set(rows.keys()), {1, 2})
+ self.assertEqual(rows[1]['payload_a'], 10)
+ self.assertEqual(rows[2]['payload_a'], 20)
+
def test_read_paimon_with_filter(self):
"""read_paimon() pushes down a predicate filter."""
from pypaimon.ray import read_paimon
