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 3d1caec797 [python] Fix MAP key resolution and ROW projection
nullability (#10054)
3d1caec797 is described below
commit 3d1caec7979e7c399b0ff82400a8f9cacbc2874b
Author: zhigang <[email protected]>
AuthorDate: Tue Sep 22 10:45:40 2026 +0800
[python] Fix MAP key resolution and ROW projection nullability (#10054)
---
paimon-python/pypaimon/read/read_builder.py | 95 ++++++++++++----------
.../read/reader/nested_leaf_batch_reader.py | 5 +-
.../tests/map_selected_key_projection_test.py | 85 +++++++++++++++++++
.../pypaimon/tests/test_nested_projection_e2e.py | 33 ++++++++
.../pypaimon/tests/test_projection_utility.py | 20 +++++
paimon-python/pypaimon/utils/projection.py | 22 +++--
6 files changed, 207 insertions(+), 53 deletions(-)
diff --git a/paimon-python/pypaimon/read/read_builder.py
b/paimon-python/pypaimon/read/read_builder.py
index 7a18e1e85f..ccac24842b 100644
--- a/paimon-python/pypaimon/read/read_builder.py
+++ b/paimon-python/pypaimon/read/read_builder.py
@@ -16,7 +16,7 @@
# under the License.
import ast
-from typing import List, Optional
+from typing import List, Optional, Sequence, Union
from pypaimon.common.predicate import Predicate
from pypaimon.common.predicate_builder import PredicateBuilder
@@ -33,6 +33,9 @@ from pypaimon.table.special_fields import SpecialFields
from pypaimon.utils.projection import MapKey, Projection, is_row_type
+ProjectionPath = Sequence[Union[int, MapKey]]
+
+
class _ReadPredicateBuilder(PredicateBuilder):
def __init__(self, fields, unsupported_fields):
@@ -60,7 +63,7 @@ class ReadBuilder:
# ``_nested_paths`` is also populated and takes precedence
# in ``read_type()`` and downstream consumers.
self._projection: Optional[List[str]] = None
- self._nested_paths: Optional[List[List[int]]] = None
+ self._nested_paths: Optional[List[ProjectionPath]] = None
self._partition_filter: Optional[Predicate] = None
self._limit: Optional[int] = None
@@ -183,31 +186,14 @@ class ReadBuilder:
# Helpers
# ------------------------------------------------------------------
- def _resolve_projection_paths(self, names: List[str]) -> List[List[int]]:
+ def _resolve_projection_paths(self, names: List[str]) ->
List[ProjectionPath]:
"""Translate ROW paths and MAP-key selectors into internal paths."""
table_fields = self.table.fields
if self.table.options.row_tracking_enabled():
table_fields =
SpecialFields.row_type_with_row_tracking(table_fields)
top_index = {f.name: i for i, f in enumerate(table_fields)}
- def resolve_row_path(top, parts):
- path = [top_index[top]]
- current_field = table_fields[path[0]]
- for part in parts:
- if not is_row_type(current_field.type):
- return None
- child_fields = current_field.type.fields
- child_idx = next(
- (i for i, f in enumerate(child_fields)
- if f.name == part),
- -1)
- if child_idx < 0:
- return None
- path.append(child_idx)
- current_field = child_fields[child_idx]
- return path
-
- paths: List[List[int]] = []
+ paths: List[ProjectionPath] = []
for name in names:
# Dot can be part of a top-level field name, not only a struct path
# separator. Top-level match takes precedence over struct walk.
@@ -221,29 +207,31 @@ class ReadBuilder:
paths.append([top_index[top], MapKey(key)])
continue
- if '.' not in name:
+ if "." not in name:
continue
# Preserve the original ROW-path semantics before considering a
# dotted top-level field name as the path prefix.
- parts = name.split('.')
+ parts = name.split(".")
top = parts[0]
if top in top_index:
- path = resolve_row_path(top, parts[1:])
+ path = _resolve_row_path(table_fields, top_index[top],
parts[1:])
if path is not None:
paths.append(path)
continue
candidates = [
- field_name for field_name in top_index
- if name.startswith(field_name + '.')
+ field_name
+ for field_name in top_index
+ if name.startswith(field_name + ".")
and is_row_type(table_fields[top_index[field_name]].type)
]
if not candidates:
continue
top = max(candidates, key=len)
- parts = name[len(top) + 1:].split('.')
- path = resolve_row_path(top, parts)
+ prefix_length = len(top) + 1
+ parts = name[prefix_length:].split(".")
+ path = _resolve_row_path(table_fields, top_index[top], parts)
if path is not None:
paths.append(path)
return paths
@@ -284,25 +272,44 @@ class ReadBuilder:
return fields
+def _resolve_row_path(
+ table_fields: List[DataField], top_index: int, parts: List[str]
+) -> Optional[List[int]]:
+ """Walk ROW children from a top-level field; return None for invalid
paths."""
+ path = [top_index]
+ current_field = table_fields[top_index]
+ for part in parts:
+ if not is_row_type(current_field.type):
+ return None
+ child_fields = current_field.type.fields
+ child_idx = next((i for i, f in enumerate(child_fields) if f.name ==
part), -1)
+ if child_idx < 0:
+ return None
+ path.append(child_idx)
+ current_field = child_fields[child_idx]
+ return path
+
+
def _map_key_selector(name, table_fields):
+ if not name.endswith("]"):
+ return None
candidates = [
- field for field in table_fields
- if _is_string_key_map(field.type)
- and name.startswith(field.name + '[')
+ field
+ for field in table_fields
+ if _is_string_key_map(field.type) and name.startswith(field.name + "[")
]
- if not candidates:
- return None
- field = max(candidates, key=lambda candidate: len(candidate.name))
- selector = name[len(field.name):]
- if not selector.endswith(']'):
- return None
- try:
- key = ast.literal_eval(selector[1:-1])
- except (SyntaxError, ValueError):
- return None
- if not isinstance(key, str):
- return None
- return field.name, key
+ for field in sorted(
+ candidates, key=lambda candidate: len(candidate.name), reverse=True
+ ):
+ prefix_length = len(field.name)
+ selector = name[prefix_length:]
+ try:
+ key = ast.literal_eval(selector[1:-1])
+ except (SyntaxError, ValueError):
+ continue
+ if isinstance(key, str):
+ return field.name, key
+ return None
def _is_string_key_map(data_type) -> bool:
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 6bb1d6450d..f85c818081 100644
--- a/paimon-python/pypaimon/read/reader/nested_leaf_batch_reader.py
+++ b/paimon-python/pypaimon/read/reader/nested_leaf_batch_reader.py
@@ -30,11 +30,12 @@ from pypaimon.schema.data_types import DataField,
PyarrowFieldParser
def _struct_field(column, name):
+ # Resolve literal names before Arrow can interpret them as field paths.
+ field_index = [field.name for field in column.type].index(name)
struct_field = getattr(pc, "struct_field", None)
if struct_field is not None:
- return struct_field(column, name)
+ return struct_field(column, field_index)
- field_index = [field.name for field in column.type].index(name)
return column.flatten()[field_index]
diff --git a/paimon-python/pypaimon/tests/map_selected_key_projection_test.py
b/paimon-python/pypaimon/tests/map_selected_key_projection_test.py
index 63670f25da..aa770564f0 100644
--- a/paimon-python/pypaimon/tests/map_selected_key_projection_test.py
+++ b/paimon-python/pypaimon/tests/map_selected_key_projection_test.py
@@ -61,6 +61,91 @@ class MapSelectedKeyProjectionTest(unittest.TestCase):
'attributes_missing': [None, None, None],
}, result.to_pydict())
+ def test_dot_prefixed_map_keys_are_literal(self):
+ for file_format in ("parquet", "row"):
+ with self.subTest(file_format=file_format):
+ schema = pa.schema([("attrs", pa.map_(pa.string(),
pa.int64()))])
+ identifier = "default.literal_dot_" + file_format
+ self.catalog.create_table(
+ identifier,
+ Schema.from_pyarrow_schema(
+ schema,
+ options={
+ "bucket": "-1",
+ "file.format": file_format,
+ },
+ ),
+ False,
+ )
+ table = self.catalog.get_table(identifier)
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ try:
+ writer.write_arrow(
+ pa.Table.from_pylist(
+ [
+ {"attrs": [("foo", 100), (".foo", 107), (".",
108)]},
+ {"attrs": []},
+ {"attrs": None},
+ ],
+ schema=schema,
+ )
+ )
+ builder.new_commit().commit(writer.prepare_commit())
+ finally:
+ writer.close()
+
+ result = self._read(table, ["attrs['foo']", "attrs['.foo']"])
+ self.assertEqual(
+ {
+ "attrs_foo": [100, None, None],
+ "attrs__foo": [107, None, None],
+ },
+ result.to_pydict(),
+ )
+ for key, expected in ((".foo", 107), (".", 108)):
+ result = self._read(table, [f"attrs[{key!r}]"])
+ self.assertEqual(
+ [expected, None, None], result.column(0).to_pylist()
+ )
+
+ def test_map_selector_skips_invalid_longer_prefix(self):
+ map_type = pa.map_(pa.string(), pa.int64())
+ schema = pa.schema([("attrs", map_type), ("attrs['x", map_type)])
+ identifier = "default.map_prefix"
+ self.catalog.create_table(
+ identifier,
+ Schema.from_pyarrow_schema(schema, options={"bucket": "-1"}),
+ False,
+ )
+ table = self.catalog.get_table(identifier)
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ try:
+ writer.write_arrow(
+ pa.Table.from_pylist(
+ [
+ {"attrs": [("x[0]", 42)], "attrs['x": [("other", 99)]},
+ ],
+ schema=schema,
+ )
+ )
+ builder.new_commit().commit(writer.prepare_commit())
+ finally:
+ writer.close()
+
+ for selector in ("attrs['x[0]']", 'attrs["x[0]"]'):
+ with self.subTest(selector=selector):
+ self.assertEqual(
+ {"attrs_x[0]": [42]}, self._read(table,
[selector]).to_pydict()
+ )
+ self.assertEqual(
+ {"attrs['x_other": [99]},
+ self._read(
+ table, ["attrs['x['other']", "attrs[123]", "attrs[other]"]
+ ).to_pydict(),
+ )
+
def test_projects_map_key_from_data_evolution_table(self):
table = self._write_table('data_evolution', {
'row-tracking.enabled': 'true',
diff --git a/paimon-python/pypaimon/tests/test_nested_projection_e2e.py
b/paimon-python/pypaimon/tests/test_nested_projection_e2e.py
index 40e53b672b..934cc4e4ed 100644
--- a/paimon-python/pypaimon/tests/test_nested_projection_e2e.py
+++ b/paimon-python/pypaimon/tests/test_nested_projection_e2e.py
@@ -21,6 +21,7 @@ import tempfile
import unittest
import pyarrow as pa
+import pyarrow.parquet as pq
from pypaimon import CatalogFactory, Schema
@@ -98,6 +99,38 @@ class AppendOnlyNestedParquetTest(_AppendOnlyNestedBase):
self.assertEqual([100, None], got.column(0).to_pylist())
+ def test_required_leaf_under_nullable_row_can_be_exported(self):
+ schema = pa.schema(
+ [
+ ("r", pa.struct([pa.field("x", pa.int64(), nullable=False)])),
+ ]
+ )
+ identifier = "default.nullable_row_required_leaf"
+ self.catalog.create_table(
+ identifier,
+ Schema.from_pyarrow_schema(schema, options={"bucket": "-1"}),
+ False,
+ )
+ table = self.catalog.get_table(identifier)
+ wb = table.new_batch_write_builder()
+ writer = wb.new_write()
+ try:
+ writer.write_arrow(
+ pa.Table.from_pylist([{"r": None}, {"r": {"x": 7}}],
schema=schema)
+ )
+ wb.new_commit().commit(writer.prepare_commit())
+ finally:
+ writer.close()
+
+ rb = table.new_read_builder().with_projection(["r.x"])
+ result = rb.new_read().to_arrow(rb.new_scan().plan().splits())
+ self.assertEqual([None, 7], result.column(0).to_pylist())
+ sink = pa.BufferOutputStream()
+ pq.write_table(result, sink)
+ restored = pq.read_table(pa.BufferReader(sink.getvalue()))
+ self.assertTrue(restored.schema.field("r_x").nullable)
+ self.assertEqual(result.to_pydict(), restored.to_pydict())
+
def test_mixed_nested_and_top_level_preserves_order(self):
table = self._create_table('ao_mixed_order')
rb = table.new_read_builder().with_projection(
diff --git a/paimon-python/pypaimon/tests/test_projection_utility.py
b/paimon-python/pypaimon/tests/test_projection_utility.py
index d3d37d10a3..87b455e0fb 100644
--- a/paimon-python/pypaimon/tests/test_projection_utility.py
+++ b/paimon-python/pypaimon/tests/test_projection_utility.py
@@ -16,6 +16,7 @@
# under the License.
import unittest
+from itertools import product
from pypaimon.schema.data_types import AtomicType, DataField, RowType
from pypaimon.utils.projection import (NestedProjection, Projection,
@@ -77,6 +78,25 @@ class TopLevelProjectionTest(unittest.TestCase):
class NestedProjectionTest(unittest.TestCase):
+ def test_nullable_is_inherited_from_all_ancestors(self):
+ for outer_nullable, inner_nullable, leaf_nullable in product(
+ (False, True), repeat=3
+ ):
+ with self.subTest(
+ outer=outer_nullable, inner=inner_nullable, leaf=leaf_nullable
+ ):
+ leaf = DataField(3, "x", AtomicType("BIGINT", leaf_nullable))
+ inner = DataField(2, "inner", RowType(inner_nullable, [leaf]))
+ outer = DataField(1, "outer", RowType(outer_nullable, [inner]))
+ projected = Projection.of([[0, 0, 0]]).project([outer])
+ self.assertEqual(
+ outer_nullable or inner_nullable or leaf_nullable,
+ projected[0].type.nullable,
+ )
+ # Projection must not alter the source schema's constraints.
+ self.assertEqual(leaf_nullable, leaf.type.nullable)
+ self.assertEqual(inner_nullable, inner.type.nullable)
+ self.assertEqual(outer_nullable, outer.type.nullable)
def test_factory_produces_nested(self):
p = Projection.of([[1, 0], [1, 1]])
diff --git a/paimon-python/pypaimon/utils/projection.py
b/paimon-python/pypaimon/utils/projection.py
index 9be66dad10..95111a96e1 100644
--- a/paimon-python/pypaimon/utils/projection.py
+++ b/paimon-python/pypaimon/utils/projection.py
@@ -226,6 +226,7 @@ class NestedProjection(Projection):
for path in self.paths:
field = fields[path[0]]
name_parts = [field.name]
+ nullable = field.type.nullable
is_map_key = False
for idx in path[1:]:
child_type = field.type
@@ -250,6 +251,7 @@ class NestedProjection(Projection):
"for field '%s'" % (child_type, field.name))
child_fields = _row_fields(child_type)
field = child_fields[idx]
+ nullable = nullable or field.type.nullable
name_parts.append(field.name)
base_name = "_".join(name_parts)
final_name = base_name
@@ -258,15 +260,21 @@ class NestedProjection(Projection):
final_name = "%s__%d" % (base_name, dup_count)
dup_count += 1
seen_names.add(final_name)
+ field_type = field.type
+ if nullable and not field_type.nullable:
+ field_type = copy(field_type)
+ field_type.nullable = True
# Keep the leaf field's ID so downstream schema-evolution
# remapping by field ID still works after rename.
- out.append(DataField(
- id=field.id,
- name=final_name,
- type=field.type,
- description=getattr(field, 'description', None),
- default_value=getattr(field, 'default_value', None),
- ))
+ out.append(
+ DataField(
+ id=field.id,
+ name=final_name,
+ type=field_type,
+ description=getattr(field, "description", None),
+ default_value=getattr(field, "default_value", None),
+ )
+ )
return out
