apache/beam · error · TypeError

Combiner input type must be specified positionally.

Error message

Combiner input type must be specified positionally.

What it means

When wrapping a type-hinted callable as a CombineFn, Beam derives the input type from the function's type hints. If no positional input argument hint exists (only keyword-arg hints, and not exactly one of them), it cannot determine the combiner input type, so it raises this TypeError.

Solutions

Annotate the single input parameter positionally: def f(xs: Iterable[int]) -> int.
If using kwonly args, collapse to exactly one positional parameter.
Explicitly set input/output type hints via with_input_types/with_output_types on the PTransform instead of relying on inference.
Remove conflicting extra keyword hints or provide exactly one keyword hint.
Example fix: `def f(data: List[int]) -> int` (positional) instead of `def f(*, data: List[int]) -> int`.

Example fix

// before
def combiner(*, data: List[int]) -> int: ...
pcoll | beam.Combine(combiner)
// after
def combiner(data: Iterable[int]) -> int: ...
pcoll | beam.Combine(combiner)

Defensive patterns

Strategy: validation

Validate before calling

hints = typing.get_type_hints(fn)
positional = [k for k in hints if k != 'return']
if not positional:
    raise TypeError('combiner fn needs a positional input annotation')

Type guard

def has_positional_input_hint(fn) -> bool:
    params = inspect.signature(fn).parameters
    return any(p.kind in (p.POSITIONAL_ONLY, p.POSITIONAL_OR_KEYWORD)
               for p in params.values())

Prevention

Annotate the combine fn's single input positionally (e.g. Iterable[int]).
Avoid keyword-only parameters in combine functions.
Keep __annotations__ intact; avoid decorators that strip type hints.

When it happens

Trigger: Passing a function annotated only with keyword arguments (e.g. def f(*, data: List[int]) -> int) or with no input annotation at all plus multiple/zero kwarg hints, to CallableWrapperCombineFn / CombineFn.from_callable / beam.Combine.

Common situations: Adding type hints to an existing combine function using keyword-only parameters; using functools.wraps or wrappers that strip positional annotations; libraries that emit functions with kwonly args.

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

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

Appendix: source

Thrown at sdks/python/apache_beam/transforms/core.py:1344

    else:
      return [self._fn(accumulator, *args, **kwargs)]

  def extract_output(self, accumulator, *args, **kwargs):
    return self._fn(accumulator, *args, **kwargs)

  def default_type_hints(self):
    fn_type_hints = typehints.decorators.IOTypeHints.from_callable(self._fn)
    type_hints = get_type_hints(self._fn).with_defaults(fn_type_hints)
    if type_hints.input_types is None:
      return type_hints
    else:
      # fn(Iterable[V]) -> V becomes CombineFn(V) -> V
      input_args, input_kwargs = type_hints.input_types
      if not input_args:
        if len(input_kwargs) == 1:
          input_args, input_kwargs = tuple(input_kwargs.values()), {}
        else:
          raise TypeError('Combiner input type must be specified positionally.')
      if not is_consistent_with(input_args[0],
                                typehints.Iterable[typehints.Any]):
        raise TypeCheckError(
            'All functions for a Combine PTransform must accept a '
            'single argument compatible with: Iterable[Any]. '
            'Instead a function with input type: %s was received.' %
            input_args[0])
      input_args = (element_type(input_args[0]), ) + input_args[1:]
      # TODO(robertwb): Assert output type is consistent with input type?
      return type_hints.with_input_types(*input_args, **input_kwargs)

  def infer_output_type(self, input_type):
    return _strip_output_annotations(
        trivial_inference.infer_return_type(self._fn, [input_type]))

  def for_input_type(self, input_type):
    # Avoid circular imports.
    from apache_beam.transforms import cy_combiners

View on GitHub (pinned to 12126d8942)