Abacn opened a new issue, #40698:
URL: https://github.com/apache/beam/issues/40698
### What would you like to happen?
### What happened?
In the Beam Portability API (`schema.proto`) and the Java SDK
(`org.apache.beam.sdk.schemas.Schema` / `org.apache.beam.sdk.values.Row`),
schema field names can be arbitrary strings. For example, Calcite SQL
(`SqlTransform`) routinely produces output schemas with column names such as:
- Unaliased expression names: `EXPR$0`, `EXPR$1`
- Quoted SQL column aliases containing spaces, punctuation, or leading
digits: `>120 days`, `31-60 days`, `91-120 days` (e.g., TPC-DS Q62, Q99)
- Python reserved keywords: `from`, `class`, `return`
However, when a schema is decoded in the Python SDK (for instance, when
receiving the output of cross-language `SqlTransform` or any portable
`RowCoder`), `SchemaTranslation.named_tuple_from_schema`
(`sdks/python/apache_beam/typehints/schemas.py`) attempts to construct a
`typing.NamedTuple` (or `dataclasses.make_dataclass` when a field starts with
`_`, thanks to a recent patch in #40542):
```python
if any(name.startswith('_') for name, _ in subfields):
user_type = dataclasses.make_dataclass(
type_name,
subfields,
namespace={'_asdict': _dataclass_asdict},
frozen=True)
else:
user_type = NamedTuple(type_name, subfields)
```
Both `typing.NamedTuple` and `dataclasses.make_dataclass` enforce strict
Python identifier rules (`name.isidentifier() and not
keyword.iskeyword(name)`), causing `RowCoder` initialization to fail with:
```
ValueError: Type names and field names must be valid identifiers: 'EXPR$0'
```
or:
```
TypeError: Field names must be valid identifiers: '31-60 days'
```
### Why is this a feature gap?
1. **Cross-language incompatibility:** A valid `PCollection<Row>` produced
by a Java transform (such as `SqlTransform` or xlang I/O connectors) cannot be
consumed in Python if any column name is not a strict Python identifier.
2. **Incremental workarounds in `named_tuple_from_schema`:** Previously,
`dataclasses.make_dataclass` was added as a fallback specifically for
leading-underscore column names like `_change_type` (CDC metadata), because
`NamedTuple` rejected them. However, `make_dataclass` still rejects
non-identifier characters (`$`, spaces, `-`, `>`), leading digits, dunder
names, and Python keywords.
3. **Python runtime already supports arbitrary attribute names:**
`RowCoderImpl` (`sdks/python/apache_beam/coders/coder_impl.py`) encodes and
decodes fields using `getattr(value, name)` and positional construction
`self.constructor(*values)`. In Python, `getattr` and class namespaces
(`type(name, (tuple,), namespace)` with `property(operator.itemgetter(i))`)
work with arbitrary string keys, not just valid Python identifiers.
### Steps to Reproduce
```python
from apache_beam.coders import RowCoder
from apache_beam.portability.api import schema_pb2
schema = schema_pb2.Schema(
id="sql_expr_schema",
fields=[
schema_pb2.Field(
name="EXPR$0",
type=schema_pb2.FieldType(atomic_type=schema_pb2.INT64)),
schema_pb2.Field(
name="31-60 days",
type=schema_pb2.FieldType(atomic_type=schema_pb2.INT64)),
])
coder = RowCoder(schema) # Raises ValueError: Type names and field names
must be valid identifiers: 'EXPR$0'
```
### Proposed Solution
In `SchemaTranslation.named_tuple_from_schema`
(`sdks/python/apache_beam/typehints/schemas.py`), when any field name in
`subfields` is not a valid Python identifier or is a Python keyword (i.e.,
rejected by both `NamedTuple` and `make_dataclass`), construct a dynamic
`tuple`-backed row class via `type(type_name, (tuple,), namespace)` that:
- Sets `_fields = tuple(name for name, _ in subfields)` and `__annotations__
= dict(subfields)` so `match_is_named_tuple()` and
`RowTypeConstraint.from_user_type()` continue to work seamlessly.
- Binds `property(operator.itemgetter(i))` in the class dictionary for each
field name (which Python allows for arbitrary string keys, so `getattr(row,
"EXPR$0")` and `getattr(row, "31-60 days")` work in O(1) time).
- Implements `__new__(cls, *args, **kwargs)`, `_asdict()`, and `__repr__()`
matching `NamedTuple` semantics while preserving the exact portable schema
field names.
### Issue Priority
Priority: 2 (default / most feature requests should be filed as P2)
### Issue Components
- [ ] Component: Python SDK
- [ ] Component: Java SDK
- [ ] Component: Go SDK
- [ ] Component: Typescript SDK
- [ ] Component: IO connector
- [ ] Component: Beam YAML
- [ ] Component: Beam examples
- [ ] Component: Beam playground
- [ ] Component: Beam katas
- [ ] Component: Website
- [ ] Component: Infrastructure
- [ ] Component: Spark Runner
- [ ] Component: Flink Runner
- [ ] Component: Prism Runner
- [ ] Component: Twister2 Runner
- [ ] Component: Hazelcast Jet Runner
- [ ] Component: Google Cloud Dataflow Runner
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
