[Bug]: [Python] ReadFromParquet(as_rows=True) yields dicts for struct fields declared as Row #41100

What happened?

For a Parquet file with a struct column, ReadFromParquet(..., as_rows=True) (the mode Beam YAML's ReadFromParquet uses) declares the field as a nested row in element_type, but each element holds a plain dict for that field. Code that relies on the declared schema, such as row.address.city in a Map or a Beam YAML MapToFields expression, fails at runtime with AttributeError: 'dict' object has no attribute 'city'.

Minimal reproducer:

import os, tempfile
import pyarrow as pa
import pyarrow.parquet as pq
import apache_beam as beam

with tempfile.TemporaryDirectory() as d:
  path = os.path.join(d, 'people.parquet')
  pq.write_table(
      pa.table({
          'name': ['a', 'b'],
          'address': [{'city': 'SF', 'zip': 1}, {'city': 'NY', 'zip': 2}],
      }), path)
  with beam.Pipeline() as p:
    rows = p | beam.io.ReadFromParquet(path, as_rows=True)
    print(rows.element_type.__annotations__['address'])
    # Row(city=str | None, zip=numpy.int64 | None) | None
    rows | beam.Map(lambda r: print(type(r.address).__name__, r.address))
    # dict {'city': 'SF', 'zip': 1}

Cause

The schema for as_rows=True is derived from the Arrow schema, so struct columns become nested row types, but the row values are built from Arrow data without converting nested struct values (which to_pylist() returns as dicts) into the corresponding row type. Nested struct values should be converted to the declared row type (recursively, including structs inside lists) so elements match element_type.

Issue Priority

Priority: 2 (default / most bugs 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: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner