[Bug]: Inconsistent types for tuple fields in Beam Schema and Row through serialization boundary #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

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

• 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