[Feature Request]: Allow non-identifier column names in Python Row

Start in sdks/python/apache_beam/typehints/schemas.py at SchemaTranslation.named_tuple_from_schema, then read the RowCoder construction path in sdks/python/apache_beam/coders/coder_impl.py and the tests covering schema translation and row coding. Reproduce the failure with the supplied schema containing EXPR$0 and 31-60 days; done means Python can construct and encode or decode rows while preserving those field names.

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):

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

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