{"record":{"id":"1ab6969c838b8c68","repo":"apache/beam","slug":"unknown-side-input-type-r","errorCode":null,"errorMessage":"Unknown side input type: %r","messagePattern":"Unknown side input type: %r","errorType":"exception","errorClass":"NotImplementedError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/worker/operations.py","lineNumber":858,"sourceCode":"\n    # We will read the side inputs in the order prescribed by the\n    # tags_and_types argument because this is exactly the order needed to\n    # replace the ArgumentPlaceholder objects in the args/kwargs of the DoFn\n    # getting the side inputs.\n    #\n    # Note that for each tag there could be several read operations in the\n    # specification. This can happen for instance if the source has been\n    # sharded into several files.\n    for i, (side_tag, view_class, view_options) in enumerate(tags_and_types):\n      sources = []\n      # Using the side_tag in the lambda below will trigger a pylint warning.\n      # However in this case it is fine because the lambda is used right away\n      # while the variable has the value assigned by the current iteration of\n      # the for loop.\n      # pylint: disable=cell-var-from-loop\n      for si in filter(lambda o: o.tag == side_tag, self.spec.side_inputs):\n        if not isinstance(si, operation_specs.WorkerSideInputSource):\n          raise NotImplementedError('Unknown side input type: %r' % si)\n        sources.append(si.source)\n      si_counter = opcounters.SideInputReadCounter(\n          self.counter_factory,\n          self.state_sampler,\n          declaring_step=self.name_context.step_name,\n          # Inputs are 1-indexed, so we add 1 to i in the side input id\n          input_index=i + 1)\n      element_counter = opcounters.OperationCounters(\n          self.counter_factory,\n          self.name_context.step_name,\n          view_options['coder'],\n          i,\n          suffix='side-input')\n      iterator_fn = sideinputs.get_iterator_fn_for_sources(\n          sources, read_counter=si_counter, element_counter=element_counter)\n      yield apache_sideinputs.SideInputMap(\n          view_class, view_options, sideinputs.EmulatedIterable(iterator_fn))\n","sourceCodeStart":840,"sourceCodeEnd":876,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/worker/operations.py#L840-L876","documentation":" 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.","triggerScenarios":" 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.","commonSituations":" 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.","solutions":["Upgrade Apache Beam to a version that supports the side input type you are using on your runner.","Replace unsupported side input usage with a supported pattern (e.g. a view of a bounded PCollection, or co-group/flatten instead).","If using a custom runner, fix its translation step to emit WorkerSideInputSource specs.","Ensure runner and SDK versions match so spec classes are identical on both sides."],"exampleFix":"# before\nside = p | beam.Create(large_stream)  # unsupported streaming side input\nresult = main | beam.Map(lambda x, s: x + s, beam.pvalue.AsIter(side))\n\n# after\n# materialize a bounded PCollection as the side input\nside = p | beam.Create(list_of_items)\nresult = main | beam.Map(lambda x, s: x + s, beam.pvalue.AsIter(side))","handlingStrategy":"validation","validationCode":"unsupported = [si for si in side_inputs\n               if not isinstance(si, operation_specs.WorkerSideInputSource)]\nassert not unsupported, f'unsupported side input specs: {unsupported}'","typeGuard":"def is_supported_side_input(si) -> bool:\n    return isinstance(si, operation_specs.WorkerSideInputSource)","tryCatchPattern":"except NotImplementedError as e:\n    if 'Unknown side input type' in str(e):\n        switch_to_supported_side_input_pattern()  # e.g. AsIter on bounded PCollection\n    raise","preventionTips":["Use bounded, materialized PCollections for side inputs on runners lacking streaming side input support.","Keep runner and SDK versions aligned.","Prefer CoGroupByKey when joining large/streamed data instead of side inputs."],"tags":["apache-beam","side-inputs","not-implemented"],"backgroundTag":"unsupported-operation","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}