Abacn opened a new issue, #40079:
URL: https://github.com/apache/beam/issues/40079
### What happened?
When Beam derives a schema (RowTypeConstraint / RowCoder) for user types
(such as dataclasses, NamedTuples, or dynamic keys generated by transforms like
beam.GroupBy), fields of type tuple (or Tuple[...], tuple[T, ...]) exhibit
inconsistent and breaking behavior depending on whether type hints are inferred:
* **Untyped / `Any`:** Falls back to `FastPrimitivesCoder` and preserves
`tuple` identity.
* **Typed as `tuple`:** Converted to `ArrayType` in Schema and deserialized
by `IterableCoder` as a `list`.
* **Impact:** Breaks downstream code expecting hashable/immutable objects
(e.g., `TypeError: cannot use 'list' as a dict key`).
* **Related bug (heterogeneous tuples):** `schemas.py` assumes all
`Sequence` types are homogeneous by taking only `arg_types[0]`. For `tuple[str,
int]`, it treats the entire tuple as `str` and crashes during encoding with
`AttributeError: 'int' object has no attribute 'encode'`.
#### Minimal Reproducer
```python
import apache_beam as beam
from apache_beam.testing.test_pipeline import TestPipeline
# Case 1: tuple deserializes as unhashable list
with TestPipeline() as p:
_ = (
p
| beam.Create([('a', 'b')])
| beam.Map(lambda x: beam.Row(spec=x))
| beam.Reshuffle() # forces RowCoder serialization
| beam.Map(lambda row: {row.spec: 1}) # TypeError: unhashable type:
'list'
| beam.Map(print)
)
# Case 1b: pipeline succeeded with tuple preserved if typehint get lost
def no_hint(x):
return x if isinstance(x, tuple) else str(x)
with TestPipeline() as p:
_ = (
p
| beam.Create([('a', 'b')])
| beam.Map(no_hint)
| beam.Map(lambda x: beam.Row(spec=x))
| beam.Reshuffle()
| beam.Map(lambda row: {row.spec: 1})
| beam.Map(print)
)
# Case 2: Heterogeneous tuple crashes on encode
with TestPipeline() as p:
_ = (
p
| beam.Create([('count', 42)])
| beam.Map(lambda x: beam.Row(spec=x))
| beam.Reshuffle() # AttributeError: 'int' object has no attribute
'encode'
| beam.Map(print)
)
# Case 2b: success if typehint get lost
with TestPipeline() as p:
_ = (
p
| beam.Create([('count', 42)])
| beam.Map(no_hint)
| beam.Map(lambda x: beam.Row(spec=x))
| beam.Reshuffle()
| beam.Map(print)
)
```
#### Root Cause
1. **`apache_beam/typehints/schemas.py:377`**: Maps all `Sequence` types
(including `tuple`) to `ArrayType`, taking only `arg_types[0]`.
2. **`apache_beam/coders/row_coder.py:165` & `coder_impl.py:1466`**: Uses
`IterableCoder`, which constructs a Python `list` upon decoding `ArrayType`.
#### Suggested Fix
* Reconstruct `tuple` (or use `TupleSequenceCoderImpl`) in `RowCoder` when
the original type constraint is a `TupleConstraint`.
* Do not treat fixed-length / heterogeneous `Tuple[T1, T2]` as homogeneous
`ArrayType(element_type=T1)`.
*
### 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
- [ ] 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]
