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 43a7e6d106 [python] Push down typed Variant projections (#10336)
43a7e6d106 is described below
commit 43a7e6d10619d4c2729817056f66bbfb2572f612
Author: XiaoHongbo <[email protected]>
AuthorDate: Mon Oct 5 19:12:21 2026 +0800
[python] Push down typed Variant projections (#10336)
---
docs/docs/pypaimon/data-types.md | 14 ++
.../data/variant/VariantMetadataUtilsTest.java | 61 +++++
paimon-python/pypaimon/ray/ray_paimon.py | 11 +-
.../pypaimon/read/datasource/ray_datasource.py | 8 +
.../pypaimon/read/datasource/split_provider.py | 19 +-
paimon-python/pypaimon/read/native_plan.py | 23 +-
paimon-python/pypaimon/read/read_builder.py | 184 +++++++++++++--
paimon-python/pypaimon/read/stream_read_builder.py | 20 +-
paimon-python/pypaimon/read/table_read.py | 167 +++++++++++---
paimon-python/pypaimon/read/variant_read_type.py | 80 +++++++
paimon-python/pypaimon/tests/native_plan_test.py | 92 ++++++++
paimon-python/pypaimon/tests/native_read_test.py | 72 +++++-
.../tests/py36/named_variant_projection_test.py | 53 +++++
.../pypaimon/tests/ray_integration_test.py | 72 ++++++
.../pypaimon/tests/split_provider_test.py | 14 +-
.../pypaimon/tests/stream_read_builder_test.py | 21 +-
.../tests/test_read_builder_nested_projection.py | 254 ++++++++++++++++++++-
paimon-python/pypaimon/tests/variant_test.py | 230 +++++++++++++++++++
18 files changed, 1312 insertions(+), 83 deletions(-)
diff --git a/docs/docs/pypaimon/data-types.md b/docs/docs/pypaimon/data-types.md
index a5bcf1e30e..63893d72fe 100644
--- a/docs/docs/pypaimon/data-types.md
+++ b/docs/docs/pypaimon/data-types.md
@@ -233,6 +233,20 @@ Field names are literal (`state.x` is not a nested path).
Missing fields are
omitted; VARIANT NULL is `None`. Values use natural Python types, even when
types differ between rows. This does not reduce Parquet I/O.
+To project VARIANT fields as named float32 columns:
+
+```python
+read_builder = table.new_read_builder().with_projection({
+ 'id': 'id',
+ 'x': "try_variant_get(payload, '$.x', 'float')",
+})
+```
+
+This requires `read.native.enabled=true` and a compatible `pypaimon-rust`; it
+does not fall back to Python. `try_variant_get` returns NULL on cast errors,
+while `variant_get` raises. Row iterators, Torch row format, and paths
containing
+`;` are unsupported.
+
### Read and update typed paths
Read existing paths as Arrow arrays, use Arrow compute, and replace them
diff --git
a/paimon-common/src/test/java/org/apache/paimon/data/variant/VariantMetadataUtilsTest.java
b/paimon-common/src/test/java/org/apache/paimon/data/variant/VariantMetadataUtilsTest.java
index cea9c27be0..79afd42fba 100644
---
a/paimon-common/src/test/java/org/apache/paimon/data/variant/VariantMetadataUtilsTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/data/variant/VariantMetadataUtilsTest.java
@@ -19,8 +19,10 @@
package org.apache.paimon.data.variant;
import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.JsonSerdeUtil;
import org.junit.jupiter.api.Test;
@@ -46,6 +48,65 @@ public class VariantMetadataUtilsTest {
.hasMessageContaining("must not contain ';'");
}
+ @Test
+ public void testReadTypeJsonInteropWithRustAndPython() {
+ // This is the read-type JSON emitted by Rust (which omits optional
nullable fields).
+ String rustJson =
+ "{\"type\":\"ROW\",\"fields\":[{\"id\":7,\"name\":\"payload\","
+ + "\"type\":{\"type\":\"ROW\",\"fields\":["
+ + "{\"id\":0,\"name\":\"0\",\"type\":\"FLOAT\","
+ +
"\"description\":\"__VARIANT_METADATA$.x;false;Asia/Shanghai\"},"
+ + "{\"id\":1,\"name\":\"1\",\"type\":\"FLOAT\","
+ +
"\"description\":\"__VARIANT_METADATA$.y;true;UTC\"}]}}]}";
+ RowType javaRow =
+ RowType.of(
+ new DataField(
+ 7,
+ "payload",
+ RowType.of(
+ new DataField(
+ 0,
+ "0",
+ DataTypes.FLOAT(),
+
VariantMetadataUtils.buildVariantMetadata(
+ "$.x", false,
"Asia/Shanghai")),
+ new DataField(
+ 1,
+ "1",
+ DataTypes.FLOAT(),
+
VariantMetadataUtils.buildVariantMetadata(
+ "$.y", true,
"UTC")))));
+ assertThat(JsonSerdeUtil.toFlatJson(javaRow)).isEqualTo(rustJson);
+ RowType parsedRustRow = (RowType) JsonSerdeUtil.fromJson(rustJson,
DataType.class);
+ RowType rustVariantRow = (RowType)
parsedRustRow.getFields().get(0).type();
+
assertThat(VariantMetadataUtils.path(rustVariantRow.getField(0).description()))
+ .isEqualTo("$.x");
+
assertThat(VariantMetadataUtils.failOnError(rustVariantRow.getField(0).description()))
+ .isFalse();
+
assertThat(VariantMetadataUtils.timeZoneId(rustVariantRow.getField(0).description()))
+ .isEqualTo(ZoneId.of("Asia/Shanghai"));
+
assertThat(VariantMetadataUtils.path(rustVariantRow.getField(1).description()))
+ .isEqualTo("$.y");
+
assertThat(VariantMetadataUtils.failOnError(rustVariantRow.getField(1).description()))
+ .isTrue();
+
assertThat(VariantMetadataUtils.timeZoneId(rustVariantRow.getField(1).description()))
+ .isEqualTo(ZoneId.of("UTC"));
+
+ // PyPaimon includes nullable in its RowType.to_dict representation.
+ String pythonJson =
+ "{\"type\":\"ROW\",\"fields\":[{\"id\":7,\"name\":\"payload\","
+ + "\"type\":{\"type\":\"ROW\",\"fields\":["
+ + "{\"id\":0,\"name\":\"0\",\"type\":\"FLOAT\","
+ +
"\"description\":\"__VARIANT_METADATA$.x;false;UTC\"}],"
+ + "\"nullable\":true}}],\"nullable\":true}";
+ DataType parsed = JsonSerdeUtil.fromJson(pythonJson, DataType.class);
+ RowType extracted = (RowType) ((RowType)
parsed).getFields().get(0).type();
+ String description = extracted.getField(0).description();
+ assertThat(VariantMetadataUtils.path(description)).isEqualTo("$.x");
+ assertThat(VariantMetadataUtils.failOnError(description)).isFalse();
+
assertThat(VariantMetadataUtils.timeZoneId(description)).isEqualTo(ZoneId.of("UTC"));
+ }
+
@Test
public void testIsVariantRow() {
// Create a row type with variant metadata
diff --git a/paimon-python/pypaimon/ray/ray_paimon.py
b/paimon-python/pypaimon/ray/ray_paimon.py
index 65380919d6..3ed118b401 100644
--- a/paimon-python/pypaimon/ray/ray_paimon.py
+++ b/paimon-python/pypaimon/ray/ray_paimon.py
@@ -29,7 +29,7 @@ Usage::
import hashlib
import importlib
import uuid
-from typing import Any, Callable, Dict, List, Optional, TYPE_CHECKING
+from typing import Any, Callable, Dict, List, Optional, TYPE_CHECKING, Union
from pypaimon.common.predicate import Predicate
@@ -55,7 +55,7 @@ def read_paimon(
catalog_options: Dict[str, str],
*,
filter: Optional[Predicate] = None,
- projection: Optional[List[str]] = None,
+ projection: Optional[Union[List[str], Dict[str, str]]] = None,
limit: Optional[int] = None,
snapshot_id: Optional[int] = None,
tag_name: Optional[str] = None,
@@ -73,7 +73,7 @@ def read_paimon(
catalog_options: Options passed to ``CatalogFactory.create()``,
e.g. ``{"warehouse": "/path/to/warehouse"}``.
filter: Optional predicate to push down into the scan.
- projection: Optional list of column names to read.
+ projection: Optional column names or output-name-to-expression mapping.
limit: Optional row limit for the scan.
snapshot_id: Optional snapshot id to time-travel to. Mutually
exclusive with ``tag_name``.
@@ -120,6 +120,11 @@ def read_paimon(
schema = PyarrowFieldParser.from_paimon_schema(
split_provider.read_type()
)
+ expression_projection = split_provider.expression_projection()
+ if expression_projection is not None:
+ from pypaimon.read.table_read import TableRead
+ schema = TableRead._apply_expression_projection_to_schema(
+ schema, expression_projection)
import pyarrow
empty_table = pyarrow.Table.from_arrays(
[pyarrow.array([], type=field.type) for field in schema],
diff --git a/paimon-python/pypaimon/read/datasource/ray_datasource.py
b/paimon-python/pypaimon/read/datasource/ray_datasource.py
index 4b7e9459da..ca06429766 100644
--- a/paimon-python/pypaimon/read/datasource/ray_datasource.py
+++ b/paimon-python/pypaimon/read/datasource/ray_datasource.py
@@ -125,6 +125,7 @@ class RayDatasource(Datasource):
predicate = self._split_provider.predicate()
read_type = self._split_provider.read_type()
nested_name_paths = self._split_provider.nested_name_paths()
+ expression_projection = self._split_provider.expression_projection()
splits = self._split_provider.splits()
limit = self._split_provider.limit()
include_row_kind = self._split_provider.include_row_kind()
@@ -133,6 +134,10 @@ class RayDatasource(Datasource):
if self._schema is None:
self._schema = PyarrowFieldParser.from_paimon_schema(read_type)
+ if expression_projection is not None:
+ from pypaimon.read.table_read import TableRead
+ self._schema =
TableRead._apply_expression_projection_to_schema(
+ self._schema, expression_projection)
if include_row_kind:
from pypaimon.read.table_read import TableRead
self._schema = TableRead._add_row_kind_to_schema(self._schema)
@@ -154,6 +159,7 @@ class RayDatasource(Datasource):
schema=schema,
limit=limit,
nested_name_paths=nested_name_paths,
+ expression_projection=expression_projection,
include_row_kind=include_row_kind,
) -> Iterable[pyarrow.Table]:
"""Read function that will be executed by Ray workers."""
@@ -165,6 +171,7 @@ class RayDatasource(Datasource):
worker_table_read = TableRead(
table, predicate, read_type, limit=limit,
nested_name_paths=nested_name_paths,
+ expression_projection=expression_projection,
include_row_kind=include_row_kind)
batch_reader = worker_table_read.to_arrow_batch_reader(splits)
@@ -193,6 +200,7 @@ class RayDatasource(Datasource):
schema=schema,
limit=limit,
nested_name_paths=nested_name_paths,
+ expression_projection=expression_projection,
include_row_kind=include_row_kind,
)
diff --git a/paimon-python/pypaimon/read/datasource/split_provider.py
b/paimon-python/pypaimon/read/datasource/split_provider.py
index 224a74fe54..fba5cc0df5 100644
--- a/paimon-python/pypaimon/read/datasource/split_provider.py
+++ b/paimon-python/pypaimon/read/datasource/split_provider.py
@@ -26,7 +26,7 @@ bridge (which already has a fully resolved ``TableRead``).
"""
from abc import ABC, abstractmethod
-from typing import Dict, List, Optional
+from typing import Dict, List, Optional, Union
from pypaimon.read.split import Split
@@ -82,6 +82,10 @@ class SplitProvider(ABC):
"""
return None
+ def expression_projection(self):
+ """Named output expressions for a pre-resolved read, or ``None``."""
+ return None
+
class CatalogSplitProvider(SplitProvider):
"""Plan splits from a fully-qualified table identifier and catalog options.
@@ -96,7 +100,7 @@ class CatalogSplitProvider(SplitProvider):
table_identifier: str,
catalog_options: Dict[str, str],
predicate=None,
- projection: Optional[List[str]] = None,
+ projection: Optional[Union[List[str], Dict[str, str]]] = None,
limit: Optional[int] = None,
snapshot_id: Optional[int] = None,
tag_name: Optional[str] = None,
@@ -142,6 +146,7 @@ class CatalogSplitProvider(SplitProvider):
self._splits_cached = None
self._read_type_cached = None
self._nested_name_paths_cached = None
+ self._expression_projection_cached = None
def _ensure_table(self):
if self._table_cached is None:
@@ -176,6 +181,7 @@ class CatalogSplitProvider(SplitProvider):
rb = rb.with_limit(self._limit)
self._read_type_cached = rb.read_type()
self._nested_name_paths_cached = rb._nested_name_paths()
+ self._expression_projection_cached = rb._expression_projection
self._splits_cached = rb.new_scan().plan().splits()
@property
@@ -197,6 +203,10 @@ class CatalogSplitProvider(SplitProvider):
self._ensure_planned()
return self._nested_name_paths_cached
+ def expression_projection(self):
+ self._ensure_planned()
+ return self._expression_projection_cached
+
def predicate(self):
return self._predicate
@@ -217,6 +227,7 @@ class PreResolvedSplitProvider(SplitProvider):
def __init__(self, table, splits: List[Split], read_type, predicate=None,
limit: Optional[int] = None, nested_name_paths=None,
+ expression_projection=None,
include_row_kind: bool = False):
self._table = table
self._splits = splits
@@ -224,6 +235,7 @@ class PreResolvedSplitProvider(SplitProvider):
self._predicate = predicate
self._limit = limit
self._nested_name_paths = nested_name_paths
+ self._expression_projection = expression_projection
self._include_row_kind = include_row_kind
def table(self):
@@ -238,6 +250,9 @@ class PreResolvedSplitProvider(SplitProvider):
def nested_name_paths(self) -> Optional[List[List[str]]]:
return self._nested_name_paths
+ def expression_projection(self):
+ return self._expression_projection
+
def include_row_kind(self) -> bool:
return self._include_row_kind
diff --git a/paimon-python/pypaimon/read/native_plan.py
b/paimon-python/pypaimon/read/native_plan.py
index c314a75e5c..8c7bd41de1 100644
--- a/paimon-python/pypaimon/read/native_plan.py
+++ b/paimon-python/pypaimon/read/native_plan.py
@@ -37,6 +37,7 @@ from pypaimon.read.plan import Plan
from pypaimon.read.split import Split
from pypaimon.read.split_serializer import (
deserialize_split_v1, serialize_split_v1)
+from pypaimon.schema.data_types import DataField, RowType
def native_runtime_available() -> bool:
@@ -350,10 +351,17 @@ def _native_read_builder(table):
def _configure_native_read_builder(builder, predicate, limit, projection,
nested_projection=None,
+ read_type: Optional[List[DataField]] = None,
include_row_kind=False):
+ if read_type is not None:
+ if nested_projection is not None:
+ raise ValueError(
+ "read_type cannot be combined with nested projection")
+ builder = builder.with_read_type(
+ json.dumps(RowType(True, read_type).to_dict()))
if nested_projection is not None:
builder = builder.with_nested_projection(nested_projection)
- elif projection is not None:
+ elif projection is not None and read_type is None:
builder = builder.with_projection(projection)
if predicate is not None:
builder = builder.with_filter(_predicate_to_native(predicate))
@@ -369,6 +377,7 @@ def _prepare_native_read(table, predicate:
Optional[Predicate] = None,
projection: Optional[List[str]] = None,
blob_parallelism: Optional[int] = None,
nested_projection: Optional[List[List[str]]] = None,
+ read_type: Optional[List[DataField]] = None,
include_row_kind: bool = False):
"""Create one Rust reader reusable across split groups."""
if not native_reader_available():
@@ -377,6 +386,7 @@ def _prepare_native_read(table, predicate:
Optional[Predicate] = None,
builder = _configure_native_read_builder(
_native_read_builder(table), predicate, limit, projection,
nested_projection=nested_projection,
+ read_type=read_type,
include_row_kind=include_row_kind)
if blob_parallelism is not None:
builder = builder.with_blob_parallelism(blob_parallelism)
@@ -390,11 +400,18 @@ def native_read(table, splits, predicate:
Optional[Predicate] = None,
projection: Optional[List[str]] = None,
blob_parallelism: Optional[int] = None,
nested_projection: Optional[List[List[str]]] = None,
+ read_type: Optional[List[DataField]] = None,
include_row_kind: bool = False):
"""Read Rust ``Split`` objects into PyArrow ``RecordBatch`` objects."""
read_splits = _prepare_native_read(
- table, predicate, limit, projection, blob_parallelism,
- nested_projection, include_row_kind)
+ table,
+ predicate=predicate,
+ limit=limit,
+ projection=projection,
+ blob_parallelism=blob_parallelism,
+ nested_projection=nested_projection,
+ read_type=read_type,
+ include_row_kind=include_row_kind)
return read_splits(splits)
diff --git a/paimon-python/pypaimon/read/read_builder.py
b/paimon-python/pypaimon/read/read_builder.py
index ccac24842b..39dc0d11b7 100644
--- a/paimon-python/pypaimon/read/read_builder.py
+++ b/paimon-python/pypaimon/read/read_builder.py
@@ -16,7 +16,9 @@
# under the License.
import ast
-from typing import List, Optional, Sequence, Union
+from typing import Dict, List, Optional, Sequence, Tuple, Union
+
+import pyarrow
from pypaimon.common.predicate import Predicate
from pypaimon.common.predicate_builder import PredicateBuilder
@@ -26,8 +28,9 @@ from pypaimon.read.push_down_utils import
predicate_field_names
from pypaimon.read.query_auth_split import QueryAuthSplit
from pypaimon.read.scan_stats import ScanStats
from pypaimon.read.split import Split
-from pypaimon.read.table_read import TableRead
+from pypaimon.read.table_read import ROW_KIND_COLUMN, TableRead
from pypaimon.read.table_scan import TableScan
+from pypaimon.read.variant_read_type import with_variant_extractions
from pypaimon.schema.data_types import AtomicType, DataField, MapType
from pypaimon.table.special_fields import SpecialFields
from pypaimon.utils.projection import MapKey, Projection, is_row_type
@@ -36,6 +39,50 @@ from pypaimon.utils.projection import MapKey, Projection,
is_row_type
ProjectionPath = Sequence[Union[int, MapKey]]
+def _string_literal(node: ast.AST) -> Optional[str]:
+ if isinstance(node, ast.Str):
+ return node.s
+ if isinstance(node, ast.Constant) and isinstance(node.value, str):
+ return node.value
+ return None
+
+
+def _normalize_sql_literals(expression: str) -> str:
+ """Decode SQL doubled quotes before validating the expression with AST."""
+ normalized = []
+ index = 0
+ previous_was_literal = False
+ while index < len(expression):
+ quote = expression[index]
+ if quote not in ("'", '"'):
+ normalized.append(quote)
+ if not quote.isspace():
+ previous_was_literal = False
+ index += 1
+ continue
+ if previous_was_literal:
+ raise ValueError("Adjacent string literals are not supported")
+ index += 1
+ literal = []
+ while index < len(expression):
+ char = expression[index]
+ if char == quote:
+ if index + 1 < len(expression) and expression[index + 1] ==
quote:
+ literal.append(quote)
+ index += 2
+ else:
+ index += 1
+ break
+ else:
+ literal.append(char)
+ index += 1
+ else:
+ raise ValueError("Unterminated string literal in projection
expression")
+ normalized.append(repr(''.join(literal)))
+ previous_was_literal = True
+ return ''.join(normalized)
+
+
class _ReadPredicateBuilder(PredicateBuilder):
def __init__(self, fields, unsupported_fields):
@@ -58,12 +105,13 @@ class ReadBuilder:
self.table: FileStoreTable = table
self._predicate: Optional[Predicate] = None
- # ``_projection`` stores the user-facing name list from
- # :meth:`with_projection`. When nested selectors are present,
- # ``_nested_paths`` is also populated and takes precedence
- # in ``read_type()`` and downstream consumers.
+ # ``_projection`` stores physical source columns. Expression aliases
+ # are kept separately for the Arrow output schema.
self._projection: Optional[List[str]] = None
self._nested_paths: Optional[List[ProjectionPath]] = None
+ self._resolved_read_type: Optional[List[DataField]] = None
+ self._expression_projection: Optional[
+ List[Tuple[str, str, Optional[int]]]] = None
self._partition_filter: Optional[Predicate] = None
self._limit: Optional[int] = None
@@ -75,23 +123,35 @@ class ReadBuilder:
self._partition_filter = partition_filter
return self
- def with_projection(self, projection: List[str]) -> 'ReadBuilder':
- """Project to the given column names.
-
- Names containing a dot (e.g. ``"struct.subfield"``) walk into ROW
- children. A quoted bracket selector on a top-level
- ``MAP<STRING, ...>`` selects one literal key (e.g.
- ``"attrs['key.with.dots']"``). Unknown names are silently skipped to
- preserve the pre-existing contract.
+ def with_projection(
+ self,
+ projection: Union[List[str], Dict[str, str]],
+ ) -> 'ReadBuilder':
+ """Project columns or named float32 VARIANT expressions.
- An exact top-level field match takes precedence over both forms.
+ Lists retain column, nested ROW and MAP-key projection semantics.
+ A mapping assigns output names to source columns or float32
+ ``variant_get`` / ``try_variant_get`` expressions. Variant
+ expressions require native reading.
"""
- self._projection = projection
+ if isinstance(projection, dict):
+ projection, variants, outputs = self._parse_expression_projection(
+ projection)
+ else:
+ variants, outputs = None, None
if projection and any(
'.' in name or '[' in name for name in projection):
- self._nested_paths = self._resolve_projection_paths(projection)
+ paths = self._resolve_projection_paths(projection)
+ nested_paths = (paths if any(len(path) > 1 for path in paths)
+ else None)
else:
- self._nested_paths = None
+ nested_paths = None
+ read_type = self._resolve_read_type(projection, nested_paths)
+ read_type = with_variant_extractions(read_type, variants)
+ self._projection = projection
+ self._nested_paths = nested_paths
+ self._resolved_read_type = read_type
+ self._expression_projection = outputs
return self
def with_limit(self, limit: int) -> 'ReadBuilder':
@@ -116,6 +176,7 @@ class ReadBuilder:
predicate=self._predicate,
read_type=self.read_type(),
nested_name_paths=self._nested_name_paths(),
+ expression_projection=self._expression_projection,
limit=self._limit,
)
@@ -168,24 +229,103 @@ class ReadBuilder:
)
def read_type(self) -> List[DataField]:
+ return (self._resolved_read_type if self._resolved_read_type is not
None
+ else self.table.fields)
+
+ def _resolve_read_type(
+ self, projection: Optional[List[str]],
+ nested_paths: Optional[List[ProjectionPath]]) -> List[DataField]:
table_fields = self.table.fields
- if not self._projection and not self._nested_paths:
+ if not projection and not nested_paths:
return table_fields
if self.table.options.row_tracking_enabled():
table_fields =
SpecialFields.row_type_with_row_tracking(table_fields)
- if self._nested_paths:
- return Projection.of(self._nested_paths).project(table_fields)
+ if nested_paths:
+ return Projection.of(nested_paths).project(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 [field_map[name] for name in projection if name in field_map]
# ------------------------------------------------------------------
# Helpers
# ------------------------------------------------------------------
+ def _parse_expression_projection(self, expressions: Dict[str, str]):
+ if not expressions:
+ raise ValueError("Projection expression mapping must not be empty")
+ table_fields = self.table.fields
+ 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}
+ projection = []
+ variants = {}
+ outputs = []
+ direct_columns = set()
+ for alias, expression in expressions.items():
+ if not isinstance(alias, str) or not alias:
+ raise TypeError("Projection output names must be non-empty
strings")
+ if alias == ROW_KIND_COLUMN:
+ raise ValueError("Projection output name %r is reserved" %
alias)
+ if not isinstance(expression, str) or not expression:
+ raise TypeError("Projection expressions must be non-empty
strings")
+ if expression in field_map:
+ source, child = expression, None
+ direct_columns.add(source)
+ else:
+ try:
+ call = ast.parse(
+ _normalize_sql_literals(expression), mode='eval').body
+ except SyntaxError as error:
+ raise ValueError(
+ "Unsupported projection expression %r" % expression
+ ) from error
+ if (not isinstance(call, ast.Call)
+ or not isinstance(call.func, ast.Name)
+ or call.func.id.lower() not in ('variant_get',
'try_variant_get')
+ or len(call.args) != 3 or call.keywords
+ or not (isinstance(call.args[0], ast.Name)
+ or _string_literal(call.args[0]) is not None)
+ or any(_string_literal(arg) is None
+ for arg in call.args[1:])):
+ raise ValueError(
+ "Unsupported projection expression %r" % expression)
+ source = (call.args[0].id
+ if isinstance(call.args[0], ast.Name)
+ else _string_literal(call.args[0]))
+ field = field_map.get(source)
+ if (field is None
+ or not isinstance(field.type, AtomicType)
+ or field.type.type.upper() != 'VARIANT'):
+ raise ValueError(
+ "Variant extraction requires a VARIANT column: %r"
+ % source)
+ path, target_type = (_string_literal(arg) for arg in
call.args[1:])
+ if ';' in path:
+ raise ValueError(
+ "Variant extraction path must not contain ';': %s"
+ % path)
+ if target_type.lower() != 'float':
+ raise ValueError(
+ "Only float32 Variant extractions are supported")
+ options = variants.setdefault(source, {
+ 'paths': [], 'target_type': pyarrow.float32(),
+ 'fail_on_error': [],
+ })
+ child = len(options['paths'])
+ options['paths'].append(path)
+ options['fail_on_error'].append(
+ call.func.id.lower() == 'variant_get')
+ if source not in projection:
+ projection.append(source)
+ outputs.append((alias, source, child))
+ if direct_columns & variants.keys():
+ raise ValueError(
+ "A VARIANT column cannot be read both whole and extracted")
+ return projection, variants or None, outputs
+
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
diff --git a/paimon-python/pypaimon/read/stream_read_builder.py
b/paimon-python/pypaimon/read/stream_read_builder.py
index 0a2753eebe..e28819be4e 100644
--- a/paimon-python/pypaimon/read/stream_read_builder.py
+++ b/paimon-python/pypaimon/read/stream_read_builder.py
@@ -22,7 +22,7 @@ This module provides a builder for configuring streaming
reads from Paimon
tables, similar to ReadBuilder but for continuous streaming use cases.
"""
-from typing import Callable, List, Optional, Set
+from typing import Callable, Dict, List, Optional, Set, Union
from pypaimon.common.predicate import Predicate
from pypaimon.common.predicate_builder import PredicateBuilder
@@ -54,7 +54,7 @@ class StreamReadBuilder:
self.table: FileStoreTable = table
self._predicate: Optional[Predicate] = None
- self._projection: Optional[List[str]] = None
+ self._read_builder = ReadBuilder(table)
self._poll_interval_ms: int = 1000
self._include_row_kind: bool = False
self._bucket_filter: Optional[Callable[[int], bool]] = None
@@ -63,11 +63,15 @@ class StreamReadBuilder:
def with_filter(self, predicate: Predicate) -> 'StreamReadBuilder':
"""Set a filter predicate for the streaming read."""
self._predicate = predicate
+ self._read_builder.with_filter(predicate)
return self
- def with_projection(self, projection: List[str]) -> 'StreamReadBuilder':
+ def with_projection(
+ self,
+ projection: Union[List[str], Dict[str, str]],
+ ) -> 'StreamReadBuilder':
"""Set column projection for the streaming read."""
- self._projection = projection
+ self._read_builder.with_projection(projection)
return self
def with_poll_interval_ms(self, poll_interval_ms: int) ->
'StreamReadBuilder':
@@ -136,6 +140,7 @@ class StreamReadBuilder:
predicate=self._predicate,
read_type=projection.read_type(),
nested_name_paths=projection._nested_name_paths(),
+ expression_projection=projection._expression_projection,
include_row_kind=self._include_row_kind
)
@@ -153,9 +158,4 @@ class StreamReadBuilder:
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
+ return self._read_builder
diff --git a/paimon-python/pypaimon/read/table_read.py
b/paimon-python/pypaimon/read/table_read.py
index 3ec796c68e..bef58ff0f5 100644
--- a/paimon-python/pypaimon/read/table_read.py
+++ b/paimon-python/pypaimon/read/table_read.py
@@ -20,7 +20,7 @@ import os
import queue
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
-from typing import Any, Callable, Dict, Iterator, List, Optional
+from typing import Any, Callable, Dict, Iterator, List, Optional, Tuple
import pandas
import pyarrow
@@ -38,6 +38,7 @@ from pypaimon.read.reader.auth_masking_reader import (
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.variant_read_type import has_variant_extractions
from pypaimon.read.split_read import (DataEvolutionSplitRead,
MergeFileSplitRead, RawFileSplitRead,
SplitRead, deferred_blob_field_names)
@@ -147,6 +148,8 @@ class TableRead:
read_type: List[DataField],
include_row_kind: bool = False,
nested_name_paths: Optional[List[List[str]]] = None,
+ expression_projection: Optional[
+ List[Tuple[str, str, Optional[int]]]] = None,
limit: Optional[int] = None,
):
from pypaimon.read.merge_engine_support import check_supported
@@ -193,11 +196,16 @@ class TableRead:
)
self.include_row_kind = include_row_kind
self.nested_name_paths = nested_name_paths
+ self.expression_projection = expression_projection
self.limit = limit
self._read_parallelism = self.table.options.read_parallelism()
self._parquet_row_group_cache = None
def to_iterator(self, splits: List[Split]) -> Iterator:
+ if self.expression_projection is not None:
+ raise RuntimeError(
+ "Projection expressions are not supported by to_iterator(); "
+ "use to_arrow() or to_arrow_batch_reader()")
self._check_python_merge_supported()
self._begin_auth_read(splits)
limit = self.limit
@@ -274,7 +282,7 @@ class TableRead:
splits: List[Split],
blob_parallelism: Optional[int] = None,
parallelism: Optional[int] = None):
- schema = PyarrowFieldParser.from_paimon_schema(self.read_type)
+ schema = self._output_arrow_schema()
if self.include_row_kind:
schema = self._add_row_kind_to_schema(schema)
effective = self._effective_parallelism(parallelism, len(splits))
@@ -372,7 +380,7 @@ class TableRead:
"""
effective_bp = self._resolve_blob_parallelism(blob_parallelism)
effective = self._effective_parallelism(parallelism, len(splits))
- schema = PyarrowFieldParser.from_paimon_schema(self.read_type)
+ schema = self._output_arrow_schema()
if self.include_row_kind:
schema = self._add_row_kind_to_schema(schema)
@@ -423,6 +431,48 @@ class TableRead:
schema=schema)
return pyarrow.Table.from_batches(batches)
+ def _output_arrow_schema(self) -> pyarrow.Schema:
+ schema = PyarrowFieldParser.from_paimon_schema(self.read_type)
+ return self._apply_expression_projection_to_schema(
+ schema, self.expression_projection)
+
+ @staticmethod
+ def _apply_expression_projection_to_schema(
+ schema: pyarrow.Schema,
+ expressions: Optional[List[Tuple[str, str, Optional[int]]]],
+ ) -> pyarrow.Schema:
+ if expressions is None:
+ return schema
+ fields = []
+ for alias, source, child in expressions:
+ index = schema.get_field_index(source)
+ if index < 0:
+ raise ValueError("Projection source %r is not in the read
type" % source)
+ field = schema.field(index)
+ if child is None:
+ fields.append(field.with_name(alias))
+ else:
+ if not pyarrow.types.is_struct(field.type):
+ raise TypeError("Variant projection %r is not a struct" %
source)
+ fields.append(pyarrow.field(alias, field.type[child].type))
+ return pyarrow.schema(fields, metadata=schema.metadata)
+
+ def _native_fallback(
+ self,
+ reason: str,
+ error: Optional[Exception] = None,
+ log: bool = False,
+ ):
+ if has_variant_extractions(self.read_type):
+ raise RuntimeError(
+ "Variant extraction requires a compatible native reader: %s"
+ % reason) from error
+ if log:
+ logger.warning(
+ "Native read failed, falling back to the Python reader: %s",
+ reason)
+ return None
+
def _try_native_batches(
self,
splits: List[Split],
@@ -432,9 +482,10 @@ class TableRead:
streaming: bool = False):
"""Return Rust-read batches, or ``None`` when this read must fall
back."""
if not self.table.options.native_read_enabled():
- return None
+ return self._native_fallback("read.native.enabled is false")
if self.table.options.file_format() not in _NATIVE_READ_FILE_FORMATS:
- return None
+ return self._native_fallback(
+ "the table file format is not supported")
if not splits:
return []
sequence_fields = self.table.options.sequence_field()
@@ -444,37 +495,40 @@ class TableRead:
# Native merge cannot extract floating sequence values: they
# would silently become missing sequence values.
if any(pyarrow.types.is_floating(field.type) for field in
sequence_schema):
- return None
+ return self._native_fallback(
+ "floating sequence fields are not supported")
if not self._native_blob_view_supported():
- return None
+ return self._native_fallback(
+ "the table uses an unsupported BLOB view")
if (self._deferred_blob_limit_may_prune(splits)
and not self.table.options.data_evolution_enabled()):
- return None
+ return self._native_fallback(
+ "deferred BLOB LIMIT pruning requires the Python reader")
# Query authorization has additional filtering, masking and projection
# semantics which are already implemented by the Python reader.
if any(isinstance(split, QueryAuthSplit) for split in splits):
- return None
+ return self._native_fallback(
+ "query authorization requires the Python reader")
try:
from pypaimon.read.native_plan import (
_prepare_native_read, native_read, native_split_from_python)
except Exception as e:
- logger.warning(
- "Native read failed, falling back to the Python reader: %s", e)
- return None
+ return self._native_fallback(str(e), e, log=True)
rust_splits = []
split_weights = []
for split in splits:
if not self._native_split_files_supported(split):
- return None
+ return self._native_fallback(
+ "a split contains a file unsupported by the native reader")
rust_split = getattr(split, '_native_split', None)
if rust_split is None:
try:
rust_split = native_split_from_python(split)
except Exception as e:
- logger.warning(
- "Native split conversion failed, falling back to the "
- "Python reader: %s", e)
- return None
+ return self._native_fallback(
+ "native split conversion failed: %s" % e,
+ e,
+ log=True)
rust_splits.append(rust_split)
split_weights.append(self._native_split_weight(split))
if (parallelism is not None
@@ -483,9 +537,7 @@ class TableRead:
try:
read_splits = _prepare_native_read(self.table, **read_kwargs)
except Exception as e:
- logger.warning(
- "Native read failed, falling back to the Python reader:
%s", e)
- return None
+ return self._native_fallback(str(e), e, log=True)
if streaming:
groups = self._native_split_groups(
rust_splits, parallelism, split_weights)
@@ -505,9 +557,7 @@ class TableRead:
"another reader failed to start",
exc_info=True,
)
- logger.warning(
- "Native read failed, falling back to the Python
reader: %s", e)
- return None
+ return self._native_fallback(str(e), e, log=True)
batches = self._native_batches_parallel_streaming(readers)
return self._convert_native_batches(batches, schema)
try:
@@ -515,28 +565,26 @@ class TableRead:
read_splits, rust_splits, schema, parallelism,
split_weights)
except _NativeReadSetupError as e:
- logger.warning(
- "Native read failed, falling back to the Python reader:
%s", e)
- return None
+ return self._native_fallback(str(e), e, log=True)
try:
read_kwargs = self._native_read_kwargs(blob_parallelism)
batches = native_read(self.table, rust_splits, **read_kwargs)
except Exception as e:
- logger.warning(
- "Native read failed, falling back to the Python reader: %s", e)
- return None
+ return self._native_fallback(str(e), e, log=True)
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:
+ # Nested and MAP-key selectors still need their source paths.
kwargs['nested_projection'] = self.nested_name_paths
+ else:
+ kwargs['read_type'] = self.read_type
if self.include_row_kind:
kwargs['include_row_kind'] = True
return kwargs
@@ -874,7 +922,7 @@ class TableRead:
break
if len(row_tuple_chunk) >= chunk_size:
- yield from
self._convert_rows_to_arrow_batches_with_row_kind(
+ yield from
self._convert_rows_to_output_batches(
row_tuple_chunk, row_kind_chunk, schema
)
row_tuple_chunk = []
@@ -883,7 +931,7 @@ class TableRead:
break
if row_tuple_chunk:
- yield from
self._convert_rows_to_arrow_batches_with_row_kind(
+ yield from self._convert_rows_to_output_batches(
row_tuple_chunk, row_kind_chunk, schema
)
finally:
@@ -1105,17 +1153,34 @@ class TableRead:
row_kind_chunk.append(row.get_row_kind().to_string())
if len(row_tuple_chunk) >= chunk_size:
-
out.extend(self._convert_rows_to_arrow_batches_with_row_kind(
+ out.extend(self._convert_rows_to_output_batches(
row_tuple_chunk, row_kind_chunk, schema))
row_tuple_chunk = []
row_kind_chunk = []
if row_tuple_chunk:
-
out.extend(self._convert_rows_to_arrow_batches_with_row_kind(
+ out.extend(self._convert_rows_to_output_batches(
row_tuple_chunk, row_kind_chunk, schema))
finally:
reader.close()
return out
+ def _convert_rows_to_output_batches(
+ self,
+ row_tuples: List[tuple],
+ row_kinds: List[str],
+ schema: pyarrow.Schema,
+ ) -> Iterator[pyarrow.RecordBatch]:
+ if self.expression_projection is None:
+ yield from self._convert_rows_to_arrow_batches_with_row_kind(
+ row_tuples, row_kinds, schema)
+ return
+ physical_schema = PyarrowFieldParser.from_paimon_schema(self.read_type)
+ if self.include_row_kind:
+ physical_schema = self._add_row_kind_to_schema(physical_schema)
+ for batch in self._convert_rows_to_arrow_batches_with_row_kind(
+ row_tuples, row_kinds, physical_schema):
+ yield self._project_batch_to_output(batch)
+
def _convert_rows_to_arrow_batches_with_row_kind(
self,
row_tuples: List[tuple],
@@ -1243,7 +1308,7 @@ class TableRead:
import ray
if not splits:
- schema = PyarrowFieldParser.from_paimon_schema(self.read_type)
+ schema = self._output_arrow_schema()
if self.include_row_kind:
schema = self._add_row_kind_to_schema(schema)
empty_table = pyarrow.Table.from_arrays(
@@ -1266,6 +1331,7 @@ class TableRead:
predicate=self.predicate,
limit=self.limit,
nested_name_paths=self.nested_name_paths,
+ expression_projection=self.expression_projection,
include_row_kind=self.include_row_kind,
)
)
@@ -1338,6 +1404,10 @@ class TableRead:
):
raise ValueError("batch_size must be a positive int or None")
if batch_format == "row":
+ if self.expression_projection is not None:
+ raise RuntimeError(
+ "Projection expressions are not supported by Torch row
format; "
+ "use streaming=True with batch_format='pyarrow' or
'torch'")
if batch_size is not None:
raise ValueError(
"batch_size requires batch_format='pyarrow' or 'torch'"
@@ -1597,6 +1667,31 @@ class TableRead:
return tuple(converted)
def _project_batch_to_output(self, batch: pyarrow.RecordBatch) ->
pyarrow.RecordBatch:
+ if self.expression_projection is not None:
+ fields = []
+ arrays = []
+ if self.include_row_kind and ROW_KIND_COLUMN in batch.schema.names:
+ fields.append(batch.schema.field(ROW_KIND_COLUMN))
+ arrays.append(batch.column(ROW_KIND_COLUMN))
+ output_schema = self._output_arrow_schema()
+ for alias, source, child in self.expression_projection:
+ index = batch.schema.get_field_index(source)
+ if index < 0:
+ raise ValueError("Projection source %r is missing" %
source)
+ array = batch.column(index)
+ if child is not None:
+ if not pyarrow.types.is_struct(array.type):
+ raise TypeError("Variant projection %r is not a
struct" % source)
+ parent = array
+ array = parent.field(child)
+ if parent.null_count:
+ array = pyarrow_compute.if_else(
+ parent.is_null(),
+ pyarrow.scalar(None, type=array.type), array)
+ arrays.append(array)
+ fields.append(output_schema.field(alias))
+ return pyarrow.RecordBatch.from_arrays(
+ arrays, schema=pyarrow.schema(fields))
if not self._needs_output_projection():
return batch
output_names = list(self._output_column_names)
@@ -1611,7 +1706,7 @@ class TableRead:
arrays, schema=pyarrow.schema(fields))
def _needs_output_projection(self) -> bool:
- return bool(self._predicate_extra_fields)
+ return bool(self._predicate_extra_fields or self.expression_projection)
def _output_extract_name_paths(self) -> List[List[str]]:
return [[f.name] for f in self.read_type]
diff --git a/paimon-python/pypaimon/read/variant_read_type.py
b/paimon-python/pypaimon/read/variant_read_type.py
new file mode 100644
index 0000000000..28cd081345
--- /dev/null
+++ b/paimon-python/pypaimon/read/variant_read_type.py
@@ -0,0 +1,80 @@
+# 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.
+
+"""Variant extractions encoded in the reader's Paimon ROW read type."""
+
+from typing import Any, Dict, List, Optional
+
+import pyarrow
+
+from pypaimon.schema.data_types import AtomicType, DataField, RowType
+
+
+_VARIANT_METADATA_PREFIX = '__VARIANT_METADATA'
+
+
+def with_variant_extractions(
+ read_type: List[DataField],
+ extractions: Optional[Dict[str, Dict[str, Any]]]) -> List[DataField]:
+ """Resolve expression paths once, using Java's DataField description
format."""
+ if not extractions:
+ return read_type
+ fields = []
+ projected = set()
+ for field in read_type:
+ options = extractions.get(field.name)
+ if options is None:
+ fields.append(field)
+ continue
+ if not isinstance(field.type, AtomicType) or field.type.type.upper()
!= 'VARIANT':
+ raise ValueError("Variant extraction column %r must be VARIANT" %
field.name)
+ if options['target_type'] != pyarrow.float32():
+ raise ValueError("Variant extraction target type must be float32")
+ error_policy = options['fail_on_error']
+ if isinstance(error_policy, bool):
+ error_policy = [error_policy] * len(options['paths'])
+ if len(error_policy) != len(options['paths']):
+ raise ValueError("Variant paths and error policies must match")
+ children = []
+ for index, (path, fail_on_error) in enumerate(
+ zip(options['paths'], error_policy)):
+ if ';' in path:
+ raise ValueError(
+ "Variant extraction path must not contain ';': %s" % path)
+ children.append(DataField(
+ index, str(index), AtomicType('FLOAT'),
+ '%s%s;%s;UTC' % (
+ _VARIANT_METADATA_PREFIX, path,
str(fail_on_error).lower())))
+ projected.add(field.name)
+ fields.append(DataField(
+ field.id, field.name, RowType(field.type.nullable, children),
+ field.description, field.default_value))
+ missing = set(extractions) - projected
+ if missing:
+ raise ValueError(
+ "Variant extraction column %r is not in the read type" %
min(missing))
+ return fields
+
+
+def has_variant_extractions(read_type: List[DataField]) -> bool:
+ """Recognize typed Variant children from their read-type metadata."""
+ return any(
+ isinstance(field.type, RowType)
+ and bool(field.type.fields)
+ and all(child.description is not None
+ and child.description.startswith(_VARIANT_METADATA_PREFIX)
+ for child in field.type.fields)
+ for field in read_type)
diff --git a/paimon-python/pypaimon/tests/native_plan_test.py
b/paimon-python/pypaimon/tests/native_plan_test.py
index ccbc92e160..e2dad2f191 100644
--- a/paimon-python/pypaimon/tests/native_plan_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_test.py
@@ -21,6 +21,7 @@ import unittest
from types import ModuleType, SimpleNamespace
from unittest.mock import Mock, call, patch
+import pyarrow
from pypaimon.catalog.catalog_context import CatalogContext
from pypaimon.catalog.filesystem_catalog_loader import FileSystemCatalogLoader
from pypaimon.catalog.jdbc_catalog_loader import JdbcCatalogLoader
@@ -33,6 +34,7 @@ from pypaimon.globalindex.global_index_result import
GlobalIndexResult
from pypaimon.globalindex.vector_search_result import ScoredGlobalIndexResult
from pypaimon.read.native_plan import (
_catalog_options,
+ _configure_native_read_builder,
_native_read_builder,
_predicate_to_native,
_resolved_schema_json,
@@ -43,12 +45,18 @@ from pypaimon.read.native_plan import (
)
from pypaimon.read.plan import Plan
from pypaimon.read.table_scan import TableScan
+from pypaimon.read.variant_read_type import with_variant_extractions
from pypaimon.schema.data_types import AtomicType, DataField, MapType,
MultisetType, RowType
from pypaimon.schema.table_schema import TableSchema
from pypaimon.table.bucket_mode import BucketMode
from pypaimon.utils.range import Range
+def _variant_read_type_json(read_type, options):
+ return json.dumps(RowType(True, with_variant_extractions(
+ read_type, options)).to_dict())
+
+
def _scan(native_enabled, file_scanner):
"""Build a TableScan without running its heavy __init__."""
scan = TableScan.__new__(TableScan)
@@ -92,6 +100,90 @@ def _scan(native_enabled, file_scanner):
class NativePlanTest(unittest.TestCase):
+ def test_variant_read_type_matches_java_interop_fixture(self):
+ actual = _variant_read_type_json(
+ [DataField(7, 'payload', AtomicType('VARIANT'))],
+ {'payload': {
+ 'paths': ['$.x'],
+ 'target_type': pyarrow.float32(),
+ 'fail_on_error': False,
+ }})
+ # Parsed by Java
VariantMetadataUtilsTest.testReadTypeJsonInteropWithRustAndPython.
+ java_fixture = (
+ '{"type":"ROW","fields":[{"id":7,"name":"payload",'
+ '"type":{"type":"ROW","fields":[{"id":0,"name":"0",'
+ '"type":"FLOAT","description":'
+ '"__VARIANT_METADATA$.x;false;UTC"}],"nullable":true}}],'
+ '"nullable":true}'
+ )
+ self.assertEqual(json.loads(actual), json.loads(java_fixture))
+
+ strict = _variant_read_type_json(
+ [DataField(7, 'payload', AtomicType('VARIANT'))],
+ {'payload': {
+ 'paths': ['$.x'],
+ 'target_type': pyarrow.float32(),
+ 'fail_on_error': True,
+ }})
+ self.assertEqual(
+
json.loads(strict)['fields'][0]['type']['fields'][0]['description'],
+ '__VARIANT_METADATA$.x;true;UTC')
+
+ mixed = _variant_read_type_json(
+ [DataField(7, 'payload', AtomicType('VARIANT'))],
+ {'payload': {
+ 'paths': ['$.x', '$.y'],
+ 'target_type': pyarrow.float32(),
+ 'fail_on_error': [False, True],
+ }})
+ children = json.loads(mixed)['fields'][0]['type']['fields']
+ self.assertEqual(
+ [child['description'] for child in children],
+ ['__VARIANT_METADATA$.x;false;UTC',
+ '__VARIANT_METADATA$.y;true;UTC'])
+
+ def test_variant_read_type_rejects_semicolon_path(self):
+ with self.assertRaisesRegex(ValueError, "must not contain ';'"):
+ _variant_read_type_json(
+ [DataField(7, 'payload', AtomicType('VARIANT'))],
+ {'payload': {
+ 'paths': ["$['a;b']"],
+ 'target_type': pyarrow.float32(),
+ 'fail_on_error': False,
+ }})
+
+ def test_native_builder_receives_variant_read_type(self):
+ builder = Mock()
+ builder.with_read_type.return_value = builder
+ variant_fields = {
+ 'payload': {
+ 'paths': ['$.ratio'],
+ 'target_type': pyarrow.float32(),
+ 'fail_on_error': False,
+ }
+ }
+ read_type = [
+ DataField(1, 'id', AtomicType('INT')),
+ DataField(2, 'payload', AtomicType('VARIANT')),
+ ]
+
+ result = _configure_native_read_builder(
+ builder,
+ predicate=None,
+ limit=None,
+ projection=['id', 'payload'],
+ read_type=with_variant_extractions(read_type, variant_fields),
+ )
+
+ self.assertIs(result, builder)
+ fields = json.loads(builder.with_read_type.call_args.args[0])['fields']
+ self.assertEqual([field['name'] for field in fields], ['id',
'payload'])
+ self.assertEqual(fields[1]['id'], 2)
+ self.assertEqual(fields[1]['type']['fields'][0]['type'], 'FLOAT')
+ self.assertEqual(fields[1]['type']['fields'][0]['description'],
+ '__VARIANT_METADATA$.ratio;false;UTC')
+ builder.with_projection.assert_not_called()
+
def setUp(self):
# Make the real capability probe see a split-API-capable pypaimon-rust
so
# gate tests route natively. Tests that call native_plan() directly
override
diff --git a/paimon-python/pypaimon/tests/native_read_test.py
b/paimon-python/pypaimon/tests/native_read_test.py
index f57c9deff0..ba466939c3 100644
--- a/paimon-python/pypaimon/tests/native_read_test.py
+++ b/paimon-python/pypaimon/tests/native_read_test.py
@@ -23,6 +23,7 @@ import pytest
from pypaimon.read.query_auth_split import QueryAuthSplit
from pypaimon.read.table_read import TableRead
+from pypaimon.read.variant_read_type import with_variant_extractions
from pypaimon.schema.data_types import AtomicType, DataField
@@ -47,6 +48,7 @@ def _table_read(limit=None):
read.read_type = [DataField(0, 'id', AtomicType('INT'))]
read.include_row_kind = False
read.nested_name_paths = None
+ read.expression_projection = None
read.limit = limit
read._read_parallelism = 1
read._deferred_blob_fields = set()
@@ -69,6 +71,72 @@ def _id_batch(values):
[pa.array(values, type=pa.int32())], names=['id'])
+def test_native_read_returns_named_variant_expression_columns():
+ read = _table_read()
+ variants = {
+ 'payload': {
+ 'paths': ['$.ratio', '$.missing'],
+ 'target_type': pa.float32(),
+ 'fail_on_error': [False, True],
+ }
+ }
+ read.read_type = with_variant_extractions([
+ DataField(0, 'id', AtomicType('INT')),
+ DataField(1, 'payload', AtomicType('VARIANT')),
+ ], variants)
+ read._scan_read_type = read.read_type
+ read._output_column_names = ['id', 'payload']
+ read.expression_projection = [
+ ('identifier', 'id', None),
+ ('ratio', 'payload', 0),
+ ('missing', 'payload', 1),
+ ]
+ split = _Split()
+ split._native_split = object()
+ payload_type = pa.struct([
+ pa.field('0', pa.float32()),
+ pa.field('1', pa.float32()),
+ ])
+ batch = pa.record_batch([
+ pa.array([1, 2, 3], type=pa.int32()),
+ pa.array([
+ {'0': 1.25, '1': None},
+ {'0': 2.5, '1': None},
+ None,
+ ], type=payload_type),
+ ], names=['id', 'payload'])
+
+ with patch('pypaimon.read.native_plan.native_read',
+ return_value=[batch]) as native:
+ result = read.to_arrow([split])
+
+ assert result.column_names == ['identifier', 'ratio', 'missing']
+ assert result.schema.field('ratio').type == pa.float32()
+ assert result.column('identifier').to_pylist() == [1, 2, 3]
+ assert result.column('ratio').to_pylist() == [1.25, 2.5, None]
+ assert result.column('missing').to_pylist() == [None, None, None]
+ assert 'variant_fields' not in native.call_args.kwargs
+ assert native.call_args.kwargs['read_type'] == read.read_type
+
+
+def test_variant_fields_never_silently_falls_back_to_python():
+ read = _table_read()
+ read.read_type = with_variant_extractions(
+ [DataField(0, 'payload', AtomicType('VARIANT'))],
+ {'payload': {
+ 'paths': ['$.ratio'],
+ 'target_type': pa.float32(),
+ 'fail_on_error': False,
+ }})
+ read._scan_read_type = read.read_type
+ read._output_column_names = ['payload']
+ read.expression_projection = [('ratio', 'payload', 0)]
+ read.table.options.native_read_enabled.return_value = False
+
+ with pytest.raises(RuntimeError, match='read.native.enabled is false'):
+ read.to_arrow([_Split()])
+
+
@pytest.mark.parametrize('type_', ['FLOAT', 'DOUBLE'])
def test_floating_sequence_falls_back_before_native_read(type_):
read = _table_read()
@@ -100,7 +168,7 @@ def
test_native_read_consumes_retained_rust_splits_and_enforces_limit():
[first._native_split, second._native_split],
predicate=None,
limit=2,
- projection=['id'],
+ read_type=read.read_type,
blob_parallelism=1,
)
@@ -196,7 +264,7 @@ def test_native_read_bridges_python_planned_split():
[converted],
predicate=None,
limit=None,
- projection=['id'],
+ read_type=read.read_type,
)
diff --git a/paimon-python/pypaimon/tests/py36/named_variant_projection_test.py
b/paimon-python/pypaimon/tests/py36/named_variant_projection_test.py
new file mode 100644
index 0000000000..9cf59a57c0
--- /dev/null
+++ b/paimon-python/pypaimon/tests/py36/named_variant_projection_test.py
@@ -0,0 +1,53 @@
+# 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 unittest
+from unittest.mock import Mock
+
+from pypaimon.read.read_builder import ReadBuilder
+from pypaimon.schema.data_types import AtomicType, DataField
+
+
+class NamedVariantProjectionTest(unittest.TestCase):
+
+ def test_string_literals_work_on_supported_python_versions(self):
+ table = Mock()
+ table.fields = [
+ DataField(0, 'id', AtomicType('INT')),
+ DataField(1, 'payload', AtomicType('VARIANT')),
+ ]
+ table.options.row_tracking_enabled.return_value = False
+
+ builder = ReadBuilder(table).with_projection({
+ 'identifier': 'id',
+ 'x': "try_variant_get(payload, '$.x', 'float')",
+ 'y': 'variant_get("payload", "$.y", "float")',
+ })
+
+ self.assertEqual(builder._projection, ['id', 'payload'])
+ self.assertEqual(
+ [field.description for field in
builder.read_type()[1].type.fields],
+ ['__VARIANT_METADATA$.x;false;UTC',
+ '__VARIANT_METADATA$.y;true;UTC'])
+ self.assertEqual(builder._expression_projection, [
+ ('identifier', 'id', None),
+ ('x', 'payload', 0),
+ ('y', 'payload', 1),
+ ])
+
+
+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 05aff4bc63..c3e47ae5f9 100644
--- a/paimon-python/pypaimon/tests/ray_integration_test.py
+++ b/paimon-python/pypaimon/tests/ray_integration_test.py
@@ -176,6 +176,78 @@ class RayIntegrationTest(unittest.TestCase):
self.assertEqual(set(df.columns), {'id', 'name'})
self.assertEqual(len(df), 2)
+ def test_read_paimon_with_named_projection(self):
+ from pypaimon.ray import read_paimon
+
+ pa_schema = pa.schema([('id', pa.int32()), ('name', pa.string())])
+ identifier = self._create_and_populate_table(
+ 'test_read_named_proj', pa_schema,
+ {'id': [1, 2], 'name': ['a', 'b']},
+ )
+ projection = {'identifier': 'id', 'label': 'name'}
+ ds = read_paimon(identifier, self.catalog_options,
+ projection=projection)
+ self.assertEqual(ds.schema().names, list(projection))
+ rows = sorted(ds.take_all(), key=lambda row: row['identifier'])
+ self.assertEqual(rows, [
+ {'identifier': 1, 'label': 'a'},
+ {'identifier': 2, 'label': 'b'},
+ ])
+
+ empty_schema = Schema.from_pyarrow_schema(pa_schema)
+ catalog = CatalogFactory.create(self.catalog_options)
+ catalog.create_table('default.test_empty_named_proj',
+ empty_schema, False)
+ empty = read_paimon('default.test_empty_named_proj',
+ self.catalog_options, projection=projection)
+ self.assertEqual(empty.schema().names, list(projection))
+ self.assertEqual(empty.count(), 0)
+
+ def test_ray_named_variant_projection_uses_resolved_read_type(self):
+ from pypaimon.data.generic_variant import GenericVariant
+ from pypaimon.read.datasource.ray_datasource import RayDatasource
+ from pypaimon.read.datasource.split_provider import
CatalogSplitProvider
+ from pypaimon.read.native_plan import (
+ native_method_available, native_split_bridge_available)
+ from pypaimon.ray import read_paimon
+
+ variant_type = pa.struct([
+ pa.field('value', pa.binary(), nullable=False),
+ pa.field('metadata', pa.binary(), nullable=False),
+ ])
+ schema = pa.schema([('id', pa.int32()), ('payload', variant_type)])
+ identifier = self._create_and_populate_table(
+ 'test_ray_named_variant_read_type', schema,
+ {'id': [1], 'payload': GenericVariant.to_arrow_array([
+ GenericVariant.from_python({'x': 1.25})])},
+ options={'read.native.enabled': 'true'},
+ )
+ expressions = {
+ 'identifier': 'id',
+ 'x': "try_variant_get(payload, '$.x', 'float')",
+ }
+ table =
CatalogFactory.create(self.catalog_options).get_table(identifier)
+ batch_type =
table.new_read_builder().with_projection(expressions).read_type()
+ stream_type = table.new_stream_read_builder().with_projection(
+ expressions).read_type()
+ provider = CatalogSplitProvider(
+ identifier, self.catalog_options, projection=expressions)
+ datasource = RayDatasource(provider)
+ self.assertEqual(batch_type, stream_type)
+ self.assertEqual(batch_type, provider.read_type())
+ self.assertEqual(
+ batch_type[1].type.fields[0].description,
+ '__VARIANT_METADATA$.x;false;UTC')
+ self.assertEqual(len(datasource.get_read_tasks(1)), 1)
+ self.assertEqual(datasource._schema.names, list(expressions))
+ self.assertEqual(datasource._schema.field('x').type, pa.float32())
+ if (native_split_bridge_available()
+ and native_method_available('ReadBuilder', 'with_read_type')):
+ rows = read_paimon(
+ identifier, self.catalog_options,
+ projection=expressions).take_all()
+ self.assertEqual(rows, [{'identifier': 1, 'x': 1.25}])
+
def test_read_paimon_count_with_query_auth(self):
from pypaimon.ray import read_paimon
diff --git a/paimon-python/pypaimon/tests/split_provider_test.py
b/paimon-python/pypaimon/tests/split_provider_test.py
index 7602a9978e..b14fce0889 100644
--- a/paimon-python/pypaimon/tests/split_provider_test.py
+++ b/paimon-python/pypaimon/tests/split_provider_test.py
@@ -100,6 +100,15 @@ class SplitProviderTest(unittest.TestCase):
field_names = [f.name for f in read_type]
self.assertEqual(field_names, ['id'])
+ named = CatalogSplitProvider(
+ table_identifier=self.identifier,
+ catalog_options=self.catalog_options,
+ projection={'identifier': 'id'},
+ )
+ self.assertEqual([field.name for field in named.read_type()], ['id'])
+ self.assertEqual(named.expression_projection(),
+ [('identifier', 'id', None)])
+
def test_catalog_provider_propagates_predicate(self):
"""``predicate`` is held on the provider and surfaced via
predicate()."""
catalog = CatalogFactory.create(self.catalog_options)
@@ -285,8 +294,11 @@ class SplitProviderTest(unittest.TestCase):
self.assertFalse(provider.include_row_kind())
provider = PreResolvedSplitProvider(
- table, splits, read_type, include_row_kind=True)
+ table, splits, read_type, include_row_kind=True,
+ expression_projection=[('alias', 'id', None)])
self.assertTrue(provider.include_row_kind())
+ self.assertEqual([('alias', 'id', None)],
+ provider.expression_projection())
if __name__ == '__main__':
diff --git a/paimon-python/pypaimon/tests/stream_read_builder_test.py
b/paimon-python/pypaimon/tests/stream_read_builder_test.py
index 459c896481..de0667c27a 100644
--- a/paimon-python/pypaimon/tests/stream_read_builder_test.py
+++ b/paimon-python/pypaimon/tests/stream_read_builder_test.py
@@ -17,12 +17,13 @@
"""Tests for StreamReadBuilder."""
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
import pytest
from pypaimon.read.stream_read_builder import StreamReadBuilder
from pypaimon.read.streaming_table_scan import AsyncStreamingTableScan
+from pypaimon.schema.data_types import AtomicType, DataField
class MockEntry:
@@ -99,6 +100,24 @@ class TestStreamReadBuilderValidation:
assert builder._bucket_filter is not None
assert builder._include_row_kind is True
+ def test_named_variant_projection_is_read_not_plan_schema(self,
mock_table):
+ mock_table.fields = [
+ DataField(0, 'id', AtomicType('INT')),
+ DataField(1, 'payload', AtomicType('VARIANT')),
+ ]
+ builder = StreamReadBuilder(mock_table).with_projection({
+ 'id_alias': 'id',
+ 'ratio': "try_variant_get(payload, '$.ratio', 'float')",
+ })
+ scan = builder.new_streaming_scan()
+ assert [field.name for field in scan._read_type] == ['id', 'payload']
+ with patch('pypaimon.read.stream_read_builder.TableRead') as read:
+ builder.new_read()
+ assert read.call_args.kwargs['expression_projection'] == [
+ ('id_alias', 'id', None), ('ratio', 'payload', 0)]
+ assert
(read.call_args.kwargs['read_type'][1].type.fields[0].description
+ == '__VARIANT_METADATA$.ratio;false;UTC')
+
class TestAsyncStreamingTableScanFiltering:
"""Test AsyncStreamingTableScan._filter_entries_for_shard()."""
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 4897bd96b1..df99ec907e 100644
--- a/paimon-python/pypaimon/tests/test_read_builder_nested_projection.py
+++ b/paimon-python/pypaimon/tests/test_read_builder_nested_projection.py
@@ -19,10 +19,17 @@ import os
import shutil
import tempfile
import unittest
+from unittest.mock import Mock, patch
import pyarrow as pa
from pypaimon import CatalogFactory, Schema
+from pypaimon.read.datasource.split_provider import (
+ CatalogSplitProvider, PreResolvedSplitProvider)
+from pypaimon.read.read_builder import ReadBuilder
+from pypaimon.read.stream_read_builder import StreamReadBuilder
+from pypaimon.read.table_read import _RemainingRows
+from pypaimon.schema.data_types import AtomicType, DataField, RowType
class _ReadBuilderTestBase(unittest.TestCase):
@@ -60,6 +67,196 @@ class _ReadBuilderTestBase(unittest.TestCase):
class ReadBuilderProjectionStateTest(_ReadBuilderTestBase):
+ @staticmethod
+ def _variant_descriptions(builder, column='payload'):
+ field = next(field for field in builder.read_type()
+ if field.name == column)
+ return [child.description for child in field.type.fields]
+
+ def test_named_variant_expressions_group_paths_and_error_policies(self):
+ table = Mock()
+ table.fields = [
+ DataField(0, 'id', AtomicType('INT')),
+ DataField(1, 'payload', AtomicType('VARIANT')),
+ ]
+ builder = ReadBuilder(table).with_projection(
+ {'identifier': 'id',
+ 'ratio': "try_variant_get(payload, '$.ratio', 'float')",
+ 'age': "variant_get(payload, '$.age', 'float')"})
+
+ self.assertEqual(['id', 'payload'], builder._projection)
+ self.assertEqual(
+ ['__VARIANT_METADATA$.ratio;false;UTC',
+ '__VARIANT_METADATA$.age;true;UTC'],
+ self._variant_descriptions(builder))
+ self.assertEqual(
+ [('identifier', 'id', None), ('ratio', 'payload', 0),
+ ('age', 'payload', 1)], builder._expression_projection)
+
+ builder.with_projection(['id'])
+ self.assertEqual(['id'], [field.name for field in builder.read_type()])
+ self.assertIsNone(builder._expression_projection)
+
+ def
test_variant_read_type_is_shared_by_batch_stream_and_ray_providers(self):
+ table = Mock()
+ table.fields = [
+ DataField(0, 'id', AtomicType('INT')),
+ DataField(1, 'payload', AtomicType('VARIANT')),
+ ]
+ table.options.row_tracking_enabled.return_value = False
+ expressions = {
+ 'identifier': 'id',
+ 'ratio': "try_variant_get(payload, '$.ratio', 'float')",
+ }
+ batch = ReadBuilder(table).with_projection(expressions)
+ stream = StreamReadBuilder(table).with_projection(expressions)
+ expected = batch.read_type()
+
+ with patch('pypaimon.read.read_builder.TableRead') as batch_read:
+ batch.new_read()
+ with patch('pypaimon.read.stream_read_builder.TableRead') as
stream_read:
+ stream.new_read()
+ self.assertEqual(expected, batch_read.call_args.kwargs['read_type'])
+ self.assertEqual(expected, stream_read.call_args.kwargs['read_type'])
+ self.assertEqual(expected, stream.new_streaming_scan()._read_type)
+
+ pre_resolved = PreResolvedSplitProvider(table, [], expected)
+ self.assertIs(expected, pre_resolved.read_type())
+ catalog = CatalogSplitProvider('default.t', {}, projection=expressions)
+ with patch.object(catalog, '_ensure_table', return_value=table), \
+ patch('pypaimon.read.read_builder.ReadBuilder.new_scan') as
scan:
+ scan.return_value.plan.return_value.splits.return_value = []
+ self.assertEqual(expected, catalog.read_type())
+
+ batch.with_projection(['id'])
+ stream.with_projection(['id'])
+ self.assertEqual([DataField(0, 'id', AtomicType('INT'))],
+ batch.read_type())
+ self.assertEqual(batch.read_type(), stream.read_type())
+
+ def test_variant_target_type_matches_native_float32_only(self):
+ table = Mock()
+ table.fields = [DataField(1, 'payload', AtomicType('VARIANT'))]
+
+ builder = ReadBuilder(table).with_projection(
+ {'x': "try_variant_get(payload, '$.x', 'float')"})
+ self.assertEqual(AtomicType('FLOAT'),
+ builder.read_type()[0].type.fields[0].type)
+ upper = ReadBuilder(table).with_projection(
+ {'x': "TRY_VARIANT_GET(payload, '$.x', 'FLOAT')"})
+ self.assertEqual(
+ ['__VARIANT_METADATA$.x;false;UTC'],
+ self._variant_descriptions(upper))
+
+ for target_type in ('double', 'int', 'string', 'timestamp', 'float32'):
+ with self.subTest(target_type=target_type):
+ with self.assertRaisesRegex(ValueError, 'Only float32'):
+ ReadBuilder(table).with_projection(
+ {'x': "try_variant_get(payload, '$.x', '%s')"
+ % target_type})
+
+ def test_variant_path_rejects_java_metadata_delimiter(self):
+ table = Mock()
+ table.fields = [DataField(1, 'payload', AtomicType('VARIANT'))]
+
+ with self.assertRaisesRegex(ValueError, "must not contain ';'"):
+ ReadBuilder(table).with_projection(
+ {'x': 'try_variant_get(payload, "$[\'a;b\']", "float")'})
+
+ def test_variant_path_preserves_sql_escaped_quotes(self):
+ table = Mock()
+ table.fields = [DataField(1, 'payload', AtomicType('VARIANT'))]
+ for function in ('variant_get', 'try_variant_get'):
+ with self.subTest(function=function):
+ builder = ReadBuilder(table).with_projection({
+ 'x': "%s(payload, '$[\"it''s\"]', 'float')" % function,
+ })
+ self.assertEqual(
+ ['__VARIANT_METADATA$["it\'s"];%s;UTC' %
+ ('true' if function == 'variant_get' else 'false')],
+ self._variant_descriptions(builder))
+ with self.assertRaisesRegex(ValueError, 'Adjacent string literals'):
+ ReadBuilder(table).with_projection({
+ 'x': "try_variant_get(payload, '$.it' 's', 'float')",
+ })
+
+ def test_named_projection_rejects_unknown_and_non_variant_sources(self):
+ table = Mock()
+ table.fields = [
+ DataField(0, 'id', AtomicType('INT')),
+ DataField(1, 'payload', AtomicType('VARIANT')),
+ ]
+
+ with self.assertRaisesRegex(ValueError, 'Unsupported projection'):
+ ReadBuilder(table).with_projection(
+ {'missing': 'not_a_column'})
+ with self.assertRaisesRegex(ValueError, 'requires a VARIANT'):
+ ReadBuilder(table).with_projection(
+ {'x': "try_variant_get(id, '$.x', 'float')"})
+ with self.assertRaisesRegex(ValueError, 'both whole and extracted'):
+ ReadBuilder(table).with_projection({
+ 'whole': 'payload',
+ 'x': "try_variant_get(payload, '$.x', 'float')"})
+
+ def test_named_projection_accepts_literal_top_level_punctuation(self):
+ table = Mock()
+ table.fields = [
+ DataField(0, 'id.dot', AtomicType('INT')),
+ DataField(1, 'payload', AtomicType('VARIANT')),
+ DataField(2, 'payload.dot', AtomicType('VARIANT')),
+ DataField(3, 'payload[raw]', AtomicType('VARIANT')),
+ ]
+ table.options.row_tracking_enabled.return_value = False
+
+ for projection, column, sources in (
+ ({'x': 'try_variant_get("payload.dot", "$.ratio", "float")'},
+ 'payload.dot', ['payload.dot']),
+ ({'x': 'try_variant_get("payload[raw]", "$.ratio", "float")'},
+ 'payload[raw]', ['payload[raw]']),
+ ({'id': 'id.dot',
+ 'x': 'try_variant_get(payload, "$.ratio", "float")'},
+ 'payload', ['id.dot', 'payload']),
+ ):
+ with self.subTest(projection=projection):
+ batch = ReadBuilder(table).with_projection(projection)
+ stream = StreamReadBuilder(table).with_projection(projection)
+ self.assertIsNone(batch._nested_paths)
+ self.assertIsNone(batch._nested_name_paths())
+ self.assertEqual(
+ sources, [field.name for field in batch.read_type()])
+ self.assertIsNone(stream._nested_name_paths())
+ self.assertEqual(
+ sources, [field.name for field in stream.read_type()])
+ with patch('pypaimon.read.read_builder.TableRead') as read:
+ batch.new_read()
+ self.assertIsNone(
+ read.call_args.kwargs['nested_name_paths'])
+ fields = read.call_args.kwargs['read_type']
+ self.assertIn(column, [field.name for field in fields])
+ with patch('pypaimon.read.stream_read_builder.TableRead') as
read:
+ stream.new_read()
+ self.assertIsNone(
+ read.call_args.kwargs['nested_name_paths'])
+ fields = read.call_args.kwargs['read_type']
+ self.assertIn(column, [field.name for field in fields])
+
+ def test_mapping_rejects_other_expressions(self):
+ table = Mock()
+ table.fields = [
+ DataField(0, 'row', RowType(True, [
+ DataField(1, 'value', AtomicType('INT'))])),
+ DataField(2, 'payload', AtomicType('VARIANT')),
+ ]
+ table.options.row_tracking_enabled.return_value = False
+ for expression in (
+ 'row.value', 'payload + 1',
+ "try_variant_get(payload, '$.ratio', 'float') + 1",
+ "try_variant_get(payload, '$.ratio', 'float', 1)",
+ ):
+ with self.subTest(expression=expression):
+ with self.assertRaisesRegex(ValueError, 'Unsupported
projection'):
+ ReadBuilder(table).with_projection({'x': expression})
+
def test_no_projection_returns_full_schema(self):
rb = self.table.new_read_builder()
fields = rb.read_type()
@@ -78,6 +275,57 @@ class ReadBuilderProjectionStateTest(_ReadBuilderTestBase):
# No nested paths derived; only names are stored.
self.assertIsNone(rb._nested_paths)
+ def test_named_ordinary_columns_keep_aliases_without_native_read(self):
+ read = self.table.new_read_builder().with_projection({
+ 'value': 'val', 'key': 'pk'}).new_read()
+ self.assertEqual(['value', 'key'], read._output_arrow_schema().names)
+ batch = pa.record_batch([
+ pa.array(['x']), pa.array([7], type=pa.int64()),
+ ], names=['val', 'pk'])
+ projected = read._project_batch_to_output(batch)
+ self.assertEqual(['value', 'key'], projected.schema.names)
+ self.assertEqual(['x'], projected.column('value').to_pylist())
+ self.assertEqual([7], projected.column('key').to_pylist())
+
+ def test_named_duplicate_columns_after_primary_key_merge(self):
+ pa_schema = pa.schema([
+ pa.field('pk', pa.int64(), nullable=False),
+ pa.field('val', pa.int64()),
+ ])
+ schema = Schema.from_pyarrow_schema(
+ pa_schema, primary_keys=['pk'], options={
+ 'bucket': '1', 'file.format': 'parquet',
+ 'read.native.enabled': 'false',
+ })
+ self.catalog.create_table('default.rb_named_duplicate_pk', schema,
False)
+ table = self.catalog.get_table('default.rb_named_duplicate_pk')
+ for value in (20, 30):
+ write_builder = table.new_batch_write_builder()
+ writer = write_builder.new_write()
+ commit = write_builder.new_commit()
+ writer.write_arrow(pa.Table.from_arrays([
+ pa.array([1], type=pa.int64()),
+ pa.array([value], type=pa.int64()),
+ ], schema=pa_schema))
+ commit.commit(writer.prepare_commit())
+ writer.close()
+ commit.close()
+
+ builder = table.new_read_builder().with_projection({
+ 'one': 'pk', 'two': 'pk', 'value': 'val',
+ })
+ splits = builder.new_scan().plan().splits()
+ read = builder.new_read()
+ expected = {'one': [1], 'two': [1], 'value': [30]}
+ self.assertEqual(
+ expected, read.to_arrow(splits, parallelism=1).to_pydict())
+ self.assertEqual(
+ expected, read.to_arrow_batch_reader(
+ splits, parallelism=1).read_all().to_pydict())
+ batches = read._read_one_split_to_batches(
+ splits[0], read._output_arrow_schema(), _RemainingRows(None))
+ self.assertEqual(expected, pa.Table.from_batches(batches).to_pydict())
+
def test_dotted_name_resolves_to_nested_path(self):
rb = self.table.new_read_builder().with_projection(
['mv.latest_version', 'pk'])
@@ -91,8 +339,8 @@ class ReadBuilderProjectionStateTest(_ReadBuilderTestBase):
def test_dotted_name_unknown_top_silently_skipped(self):
rb = self.table.new_read_builder().with_projection(
['nope.x', 'val'])
- # Only 'val' resolved; the dot trigger still populates _nested_paths.
- self.assertEqual(rb._nested_paths, [[2]])
+ # Only 'val' resolved, with no nested field actually selected.
+ self.assertIsNone(rb._nested_paths)
names = [f.name for f in rb.read_type()]
self.assertEqual(names, ['val'])
@@ -100,7 +348,7 @@ class ReadBuilderProjectionStateTest(_ReadBuilderTestBase):
rb = self.table.new_read_builder().with_projection(
['mv.no_such_subfield', 'pk'])
# The bad path drops out, the plain name survives.
- self.assertEqual(rb._nested_paths, [[0]])
+ self.assertIsNone(rb._nested_paths)
names = [f.name for f in rb.read_type()]
self.assertEqual(names, ['pk'])
diff --git a/paimon-python/pypaimon/tests/variant_test.py
b/paimon-python/pypaimon/tests/variant_test.py
index 6e85718d17..4a27543fe0 100644
--- a/paimon-python/pypaimon/tests/variant_test.py
+++ b/paimon-python/pypaimon/tests/variant_test.py
@@ -40,6 +40,7 @@ Sections
shredded VARIANT.
"""
+import asyncio
import io
import json
import os
@@ -74,6 +75,7 @@ from pypaimon.data.variant_shredding import (
rebuild_value,
shredding_schema_to_arrow_type,
)
+from pypaimon.read.native_plan import native_split_bridge_available
from pypaimon.schema.data_types import (
AtomicType,
DataField,
@@ -101,6 +103,16 @@ def _variant_arrow_type() -> pa.StructType:
])
+def _native_variant_projection_available() -> bool:
+ if not native_split_bridge_available():
+ return False
+ try:
+ from pypaimon_rust.datafusion import ReadBuilder as NativeReadBuilder
+ return callable(getattr(NativeReadBuilder, 'with_read_type', None))
+ except (ImportError, AttributeError, TypeError, ValueError):
+ return False
+
+
def _make_metadata(*keys: str) -> bytes:
"""Build variant metadata bytes. With no keys returns the minimal
header."""
if keys:
@@ -1194,6 +1206,224 @@ class TestVariantPaimonTable(unittest.TestCase):
pa.field('payload', _variant_arrow_type()),
])
+ def test_native_variant_projection_rejects_row_readers(self):
+ for mode, options in (
+ ('append', {}),
+ ('lazy', {
+ 'data-evolution.enabled': 'true',
+ 'row-tracking.enabled': 'true',
+ }),
+ ):
+ with self.subTest(mode=mode):
+ identifier = 'default.variant_projection_row_' + mode
+ schema = Schema.from_pyarrow_schema(
+ self._pa_schema(), options=options)
+ self.catalog.create_table(identifier, schema, False)
+ table = self.catalog.get_table(identifier)
+ data = pa.table({
+ 'id': [1],
+ 'payload': GenericVariant.to_arrow_array([
+ GenericVariant.from_python({'ratio': 1.25})]),
+ }, schema=self._pa_schema())
+ write_builder = table.new_batch_write_builder()
+ writer = write_builder.new_write()
+ commit = write_builder.new_commit()
+ writer.write_arrow(data)
+ commit.commit(writer.prepare_commit())
+ writer.close()
+ commit.close()
+
+ builder = table.new_read_builder().with_projection(
+ {'ratio': "try_variant_get(payload, '$.ratio', 'float')"})
+ splits = builder.new_scan().plan().splits()
+ self.assertTrue(splits)
+ read = builder.new_read()
+
+ with self.assertRaisesRegex(RuntimeError, 'to_iterator'):
+ read.to_iterator(splits)
+ with self.assertRaisesRegex(RuntimeError, 'Torch row format'):
+ read.to_torch(splits, streaming=True)
+ with self.assertRaisesRegex(RuntimeError, 'Torch row format'):
+ read.to_torch(splits, streaming=False)
+
+ def _check_variant_projection_literal_top_level_names(
+ self, suffix, native_read):
+ pa_schema = pa.schema([
+ pa.field('id.dot', pa.int64()),
+ pa.field('payload', _variant_arrow_type()),
+ pa.field('payload.dot', _variant_arrow_type()),
+ pa.field('payload[raw]', _variant_arrow_type()),
+ ])
+ schema = Schema.from_pyarrow_schema(
+ pa_schema, options={'read.native.enabled': 'true'})
+ identifier = 'default.variant_projection_literal_names_' + suffix
+ self.catalog.create_table(identifier, schema, False)
+ table = self.catalog.get_table(identifier)
+ values = GenericVariant.to_arrow_array([
+ GenericVariant.from_python({'ratio': 1.25})])
+ data = pa.Table.from_arrays([
+ pa.array([1], type=pa.int64()), values, values, values,
+ ], schema=pa_schema)
+ write_builder = table.new_batch_write_builder()
+ writer = write_builder.new_write()
+ commit = write_builder.new_commit()
+ writer.write_arrow(data)
+ commit.commit(writer.prepare_commit())
+ writer.close()
+ commit.close()
+
+ for expressions, projection, column in (
+ ({'ratio': 'try_variant_get("payload.dot", "$.ratio", "float")'},
+ ['payload.dot'], 'payload.dot'),
+ ({'ratio': 'try_variant_get("payload[raw]", "$.ratio", "float")'},
+ ['payload[raw]'], 'payload[raw]'),
+ ({'id': 'id.dot',
+ 'ratio': 'try_variant_get(payload, "$.ratio", "float")'},
+ ['id.dot', 'payload'], 'payload'),
+ ):
+ with self.subTest(expressions=expressions):
+ builder = table.new_read_builder().with_projection(
+ expressions)
+ splits = builder.new_scan().plan().splits()
+ self.assertTrue(splits)
+ read = builder.new_read()
+ kwargs = read._native_read_kwargs()
+ self.assertEqual(
+ [field.name for field in kwargs['read_type']], projection)
+ self.assertNotIn('nested_projection', kwargs)
+ field = next(field for field in kwargs['read_type']
+ if field.name == column)
+ self.assertEqual(
+ field.type.fields[0].description,
+ '__VARIANT_METADATA$.ratio;false;UTC')
+ self.assertEqual(
+ read._output_arrow_schema().field('ratio').type,
+ pa.float32())
+ if native_read:
+ projected = read.to_arrow(splits)
+ self.assertEqual(projected.column_names,
+ list(expressions))
+ self.assertEqual(
+ projected['ratio'].to_pylist(), [1.25])
+ plain = table.new_read_builder().with_projection(projection)
+ self.assertEqual(
+ plain.new_read().to_arrow(splits).column_names,
+ projection)
+
+ def test_variant_projection_literal_top_level_names_bridge(self):
+ self._check_variant_projection_literal_top_level_names(
+ 'bridge', native_read=False)
+
+ @unittest.skipUnless(_native_variant_projection_available(),
+ 'compatible native Variant projection API not
installed')
+ def test_variant_projection_literal_top_level_names_native(self):
+ self._check_variant_projection_literal_top_level_names(
+ 'native', native_read=True)
+
+ @unittest.skipUnless(_native_variant_projection_available(),
+ 'compatible native Variant projection API not
installed')
+ def test_named_variant_projection_sql_escaped_path(self):
+ schema = Schema.from_pyarrow_schema(
+ self._pa_schema(), options={'read.native.enabled': 'true'})
+ identifier = 'default.named_variant_escaped_path'
+ self.catalog.create_table(identifier, schema, False)
+ table = self.catalog.get_table(identifier)
+ data = pa.table({
+ 'id': [1],
+ 'payload': GenericVariant.to_arrow_array([
+ GenericVariant.from_python({"it's": 1.0, 'its': 2.0})]),
+ }, schema=self._pa_schema())
+ write_builder = table.new_batch_write_builder()
+ writer = write_builder.new_write()
+ commit = write_builder.new_commit()
+ writer.write_arrow(data)
+ commit.commit(writer.prepare_commit())
+ writer.close()
+ commit.close()
+
+ from pypaimon_rust.datafusion import SQLContext
+ context = SQLContext()
+ context.register_catalog(
+ 'paimon', {'warehouse': os.path.join(self.tmpdir, 'warehouse')})
+ for function in ('variant_get', 'try_variant_get'):
+ expression = "%s(payload, '$[\"it''s\"]', 'float')" % function
+ builder = table.new_read_builder().with_projection({'x':
expression})
+ splits = builder.new_scan().plan().splits()
+ actual = builder.new_read().to_arrow(splits)['x'].to_pylist()
+ sql = pa.Table.from_batches(context.sql(
+ 'SELECT %s AS x FROM %s' % (expression, identifier)))
+ self.assertEqual([1.0], actual)
+ self.assertEqual(sql['x'].to_pylist(), actual)
+
+ @unittest.skipUnless(_native_variant_projection_available(),
+ 'compatible native Variant projection API not
installed')
+ def test_named_variant_projection_batch_and_stream(self):
+ schema = Schema.from_pyarrow_schema(
+ self._pa_schema(), options={'read.native.enabled': 'true'})
+ identifier = 'default.named_variant_projection'
+ self.catalog.create_table(identifier, schema, False)
+ table = self.catalog.get_table(identifier)
+ data = pa.table({
+ 'id': [1, 2],
+ 'payload': GenericVariant.to_arrow_array([
+ GenericVariant.from_python({'x': 1.25, 'y': 2}), None]),
+ }, schema=self._pa_schema())
+ writer_builder = table.new_batch_write_builder()
+ writer = writer_builder.new_write()
+ commit = writer_builder.new_commit()
+ writer.write_arrow(data)
+ commit.commit(writer.prepare_commit())
+ writer.close()
+ commit.close()
+
+ expressions = {
+ 'identifier': 'id',
+ 'x': "try_variant_get(payload, '$.x', 'float')",
+ 'y': "variant_get(payload, '$.y', 'float')",
+ }
+ batch_builder = table.new_read_builder().with_projection(expressions)
+ expected = {'identifier': [1, 2], 'x': [1.25, None], 'y': [2.0, None]}
+ result = batch_builder.new_read().to_arrow(
+ batch_builder.new_scan().plan().splits())
+ self.assertEqual(result.column_names, list(expressions))
+ self.assertEqual(result.to_pydict(), expected)
+
+ stream_builder = table.new_stream_read_builder().with_projection(
+ expressions)
+
+ async def first_plan():
+ stream = stream_builder.new_streaming_scan().stream()
+ try:
+ return await stream.__anext__()
+ finally:
+ await stream.aclose()
+
+ stream_result = stream_builder.new_read().to_arrow(
+ asyncio.run(first_plan()).splits())
+ self.assertEqual(stream_result.to_pydict(), expected)
+
+ bad = pa.table({
+ 'id': [3],
+ 'payload': GenericVariant.to_arrow_array([
+ GenericVariant.from_python({'x': 'bad'})]),
+ }, schema=self._pa_schema())
+ writer = writer_builder.new_write()
+ commit = writer_builder.new_commit()
+ writer.write_arrow(bad)
+ commit.commit(writer.prepare_commit())
+ writer.close()
+ commit.close()
+
+ tolerant = table.new_read_builder().with_projection({
+ 'x': "try_variant_get(payload, '$.x', 'float')"})
+ result = tolerant.new_read().to_arrow(
+ tolerant.new_scan().plan().splits())
+ self.assertEqual(result['x'].to_pylist(), [1.25, None, None])
+ strict = table.new_read_builder().with_projection({
+ 'x': "variant_get(payload, '$.x', 'float')"})
+ with self.assertRaisesRegex(Exception, 'Cannot cast Variant'):
+ strict.new_read().to_arrow(strict.new_scan().plan().splits())
+
def test_plain_variant_write_and_read(self):
"""Plain VARIANT: GenericVariant → write_arrow → to_arrow →
GenericVariant."""
schema = Schema.from_pyarrow_schema(self._pa_schema())
