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
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
apache_beam/typehints/schemas.py:377: Maps all Sequence types (including tuple) to ArrayType, taking only arg_types[0].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
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:
Any: Falls back toFastPrimitivesCoderand preservestupleidentity.tuple: Converted toArrayTypein Schema and deserialized byIterableCoderas alist.TypeError: cannot use 'list' as a dict key).schemas.pyassumes allSequencetypes are homogeneous by taking onlyarg_types[0]. Fortuple[str, int], it treats the entire tuple asstrand crashes during encoding withAttributeError: 'int' object has no attribute 'encode'.Minimal Reproducer
Root Cause
apache_beam/typehints/schemas.py:377: Maps allSequencetypes (includingtuple) toArrayType, taking onlyarg_types[0].apache_beam/coders/row_coder.py:165&coder_impl.py:1466: UsesIterableCoder, which constructs a Pythonlistupon decodingArrayType.Suggested Fix
tuple(or useTupleSequenceCoderImpl) inRowCoderwhen the original type constraint is aTupleConstraint.Tuple[T1, T2]as homogeneousArrayType(element_type=T1).Issue Priority
Priority: 2 (default / most bugs should be filed as P2)
Issue Components