jk-shah opened a new issue, #41099:
URL: https://github.com/apache/beam/issues/41099
### What happened?
Accessing a nested field of a schema-aware PCollection is inferred as `Any`.
In Beam YAML this makes a common pipeline fail at construction: a `MapToFields`
that projects a field of a Parquet struct column feeds `WriteToParquet`, which
raises `NotImplementedError: Beam logical types are not currently supported in
arrow_type_compatibility` (#23817) because the inferred field type is `Any`.
Reproducer (Parquet file with `name: string, address: struct<city: string,
zip: int64>`):
```yaml
pipeline:
type: chain
transforms:
- type: ReadFromParquet
config:
path: people_struct.parquet
- type: MapToFields
config:
language: python
fields:
name: name
city: address.city
- type: WriteToParquet
config:
path: out/people
```
`ValueError: Error applying transform "WriteToParquet" at line 13: Beam
logical types are not currently supported in arrow_type_compatibility.`
The same happens with `trivial_inference` directly:
```python
import typing
from apache_beam.typehints import trivial_inference
class Address(typing.NamedTuple):
city: str
class Person(typing.NamedTuple):
name: str
address: typing.Optional[Address]
trivial_inference.infer_return_type(lambda r: r.name, [Person]) #
Any (expected str)
trivial_inference.infer_return_type(lambda r: r.address.city, [Person]) #
Any (expected Optional[str])
```
### Cause
`opcodes._getattr` (`sdks/python/apache_beam/typehints/opcodes.py`) resolves
`row.field` only when the type is a `RowTypeConstraint`. Schema-aware
PCollections are usually typed with the NamedTuple class generated for their
schema (for example `ReadFromParquet(as_rows=True)` produces a `BeamSchema_...`
class), and nested struct fields are `Optional[Row]` / `Optional[NamedTuple]`,
so both cases fall through to `Any`.
Resolving NamedTuple classes through `RowTypeConstraint.from_user_type` and
unwrapping `Optional[row]` (keeping the result nullable) fixes inference. I
have a small fix with tests in `trivial_inference_test` and can open a PR.
Note: with that fix the Parquet pipeline above builds, but then fails at
runtime due to a separate issue where `ReadFromParquet(as_rows=True)` yields
dicts for struct fields (filed separately).
### Issue Priority
Priority: 2 (default / most bugs should be filed as P2)
### Issue Components
- [x] Component: Python SDK
- [ ] Component: Java SDK
- [ ] Component: Go SDK
- [ ] Component: Typescript SDK
- [ ] Component: IO connector
- [x] 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
--
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]
