apache/beam · error · NotImplementedError

Unknown side input type: %r

Error message

Unknown side input type: %r

What it means

During DoFn setup, _read_side_inputs resolves each side input by tag and requires it to be a operation_specs.WorkerSideInputSource. Any other side input spec raises NotImplementedError 'Unknown side input type', because only materialized-PCollection side inputs are supported at this execution point.

Solutions

1. Upgrade Apache Beam to a version that supports the side input type you are using on your runner.
2. Replace unsupported side input usage with a supported pattern (e.g. a view of a bounded PCollection, or co-group/flatten instead).
3. If using a custom runner, fix its translation step to emit WorkerSideInputSource specs.
4. Ensure runner and SDK versions match so spec classes are identical on both sides.

Example fix

# before
side = p | beam.Create(large_stream)  # unsupported streaming side input
result = main | beam.Map(lambda x, s: x + s, beam.pvalue.AsIter(side))

# after
# materialize a bounded PCollection as the side input
side = p | beam.Create(list_of_items)
result = main | beam.Map(lambda x, s: x + s, beam.pvalue.AsIter(side))

unsupported = [si for si in side_inputs
               if not isinstance(si, operation_specs.WorkerSideInputSource)]
assert not unsupported, f'unsupported side input specs: {unsupported}'

def is_supported_side_input(si) -> bool:
    return isinstance(si, operation_specs.WorkerSideInputSource)

except NotImplementedError as e:
    if 'Unknown side input type' in str(e):
        switch_to_supported_side_input_pattern()  # e.g. AsIter on bounded PCollection
    raise

When it happens

Trigger: Running a pipeline whose ParDo side inputs include a spec type other than WorkerSideInputSource (e.g. streaming/iterable side inputs) on a runner/worker path that only implements the basic source type.

Common situations: Using side inputs with runners or modes lacking support (e.g. older Beam, streaming side inputs on unsupported backends); custom runners that fail to translate side inputs properly; SDK/runner version mismatch causing translated specs to differ.

Understand the failure class

Background: UnsupportedOperationException and "is not supported" errors: when a library deliberately refuses a call — this error's family across 30 libraries.

Related errors

• Assigning an index is not yet supported. Consider using…
• by
• collecting metrics will come later!
• concat(ignore_index)
• concat(levels)

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

Appendix: source

Thrown at sdks/python/apache_beam/runners/worker/operations.py:858

    # We will read the side inputs in the order prescribed by the
    # tags_and_types argument because this is exactly the order needed to
    # replace the ArgumentPlaceholder objects in the args/kwargs of the DoFn
    # getting the side inputs.
    #
    # Note that for each tag there could be several read operations in the
    # specification. This can happen for instance if the source has been
    # sharded into several files.
    for i, (side_tag, view_class, view_options) in enumerate(tags_and_types):
      sources = []
      # Using the side_tag in the lambda below will trigger a pylint warning.
      # However in this case it is fine because the lambda is used right away
      # while the variable has the value assigned by the current iteration of
      # the for loop.
      # pylint: disable=cell-var-from-loop
      for si in filter(lambda o: o.tag == side_tag, self.spec.side_inputs):
        if not isinstance(si, operation_specs.WorkerSideInputSource):
          raise NotImplementedError('Unknown side input type: %r' % si)
        sources.append(si.source)
      si_counter = opcounters.SideInputReadCounter(
          self.counter_factory,
          self.state_sampler,
          declaring_step=self.name_context.step_name,
          # Inputs are 1-indexed, so we add 1 to i in the side input id
          input_index=i + 1)
      element_counter = opcounters.OperationCounters(
          self.counter_factory,
          self.name_context.step_name,
          view_options['coder'],
          i,
          suffix='side-input')
      iterator_fn = sideinputs.get_iterator_fn_for_sources(
          sources, read_counter=si_counter, element_counter=element_counter)
      yield apache_sideinputs.SideInputMap(
          view_class, view_options, sideinputs.EmulatedIterable(iterator_fn))

View on GitHub (pinned to 12126d8942)