apache/beam · error · CompositeTypeHintError

hint type-constraint violated. The type of element # in the…

Error message

%s hint type-constraint violated. The type of element #%s in the passed %s is incorrect: %s

What it means

When an element inside a constrained sequence fails with a CompositeTypeHintError (e.g. a nested List[Tuple[int, str]] whose inner tuple is wrong), Beam re-raises it wrapped in this error, preserving the outer hint repr, the failing element index, the sequence type, and the nested error message. It exists to give a full path/context for deeply nested composite type failures.

Solutions

1. Read the nested message appended after 'incorrect:' to find the deepest failing element and fix its type.
2. Correct the nested structure construction (e.g. ensure tuples have exactly the hinted arity and types).
3. Update hints to match the real nested shape if the data shape is intentional.
4. Write a small unit test with beam.Create(sample) and the typed transform to reproduce and fix the nested mismatch.

Example fix

# before
List[Tuple[str, int]] data: [('a', '1'), ('b', 2)]  # '1' is str
# after
[('a', 1), ('b', 2)]  # every tuple is (str, int)

def check_nested(rows):
    for i, t in enumerate(rows):
        assert isinstance(t, tuple) and len(t) == 2 and isinstance(t[0], str) and isinstance(t[1], int), f'row {i} bad: {t!r}'
    return rows

def is_tuple2(t, a, b):
    return isinstance(t, tuple) and len(t) == 2 and isinstance(t[0], a) and isinstance(t[1], b)

try:
    expand(pcoll)
except CompositeTypeHintError as e:
    logger.error('nested element failure: %s', e)  # tail of message names the deepest failure
    raise

When it happens

Trigger: type_check on a SequenceTypeConstraint whose element fails check_constraint with CompositeTypeHintError instead of SimpleTypeHintError — i.e. the bad element is itself a composite (nested list, tuple, dict) violating its own constraint.

Common situations: Nested structures like List[Tuple[str, int]] where one tuple has swapped or wrong-typed fields; lists of dicts hinted as List[KV[str, int]]; recursively built data where one branch has the wrong shape.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Related errors

• An Option type-hint only accepts a single type parameter.
• bad type
• batch type must be List[T] for element type T
• batch type must be np.ndarray or…
• Cannot create Union without a sequence of types.

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/b7342f9af864cee4. Report an issue: GitHub.

Appendix: source

Thrown at sdks/python/apache_beam/typehints/typehints.py:301

              self._sequence_type.__name__.title(),
              self._sequence_type.__name__.lower(),
              sequence_instance.__class__.__name__))

    for index, elem in enumerate(sequence_instance):
      try:
        check_constraint(self.inner_type, elem)
      except SimpleTypeHintError:
        raise CompositeTypeHintError(
            '%s hint type-constraint violated. The type of element #%s in '
            'the passed %s is incorrect. Expected an instance of type %s, '
            'instead received an instance of type %s.' % (
                repr(self),
                index,
                repr(self._sequence_type),
                repr(self.inner_type),
                elem.__class__.__name__))
      except CompositeTypeHintError as e:
        raise CompositeTypeHintError(
            '%s hint type-constraint violated. The type of element #%s in '
            'the passed %s is incorrect: %s' %
            (repr(self), index, self._sequence_type.__name__, e))

  def match_type_variables(self, concrete_type):
    if isinstance(concrete_type, SequenceTypeConstraint):
      return match_type_variables(self.inner_type, concrete_type.inner_type)
    return {}

  def bind_type_variables(self, bindings):
    bound_inner_type = bind_type_variables(self.inner_type, bindings)
    if bound_inner_type == self.inner_type:
      return self
    bound_self = copy.copy(self)
    bound_self.inner_type = bound_inner_type
    return bound_self

View on GitHub (pinned to 12126d8942)