apache / apache/beam

[Bug]: Inconsistent types for tuple fields in Beam Schema and Row through serialization boundary

Open
#40,079 1 comment 0 reactions 0 assignees View on GitHub
bug P2 python schemas
Dominant language
Java
Stars
8.7k
Forks
4.7k
Avg merge
1d 20h
Merged PRs (30d)
196

Description

### 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

Contributor guide

Open the contributing guide

Research direction

Start by running the minimal reproducer, then read apache_beam/typehints/schemas.py around line 377 and apache_beam/coders/row_coder.py around line 165, including coder_impl.py around line 1466. Trace how tuple type constraints become ArrayType and are decoded, then verify that homogeneous tuples preserve tuple identity and heterogeneous tuples encode and decode without the reported errors.

Written by the indexing model from the issue text.

Assessment

Tech stack
python
Domain
data-engineering
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Active
Clarity
Mostly clear
Newbie friendliness
52/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.