{"record":{"id":"12cff868b09272f6","repo":"apache/beam","slug":"can-only-iterate-once-over-prefetchingsourcesetiterable","errorCode":null,"errorMessage":"Can only iterate once over PrefetchingSourceSetIterable instance.","messagePattern":"Can only iterate once over PrefetchingSourceSetIterable instance\\.","errorType":"exception","errorClass":"RuntimeError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/worker/sideinputs.py","lineNumber":151,"sourceCode":"                  self.element_queue.put(value)\n                else:\n                  self.element_queue.put(_globally_windowed(value))\n        except queue.Empty:\n          return\n    except Exception as e:  # pylint: disable=broad-except\n      _LOGGER.error(\n          'Encountered exception in PrefetchingSourceSetIterable '\n          'reader thread: %s',\n          traceback.format_exc())\n      self.reader_exceptions.put(e)\n      self.has_errored = True\n    finally:\n      self.element_queue.put(READER_THREAD_IS_DONE_SENTINEL)\n\n  def __iter__(self):\n    # pylint: disable=too-many-nested-blocks\n    if self.already_iterated:\n      raise RuntimeError(\n          'Can only iterate once over PrefetchingSourceSetIterable instance.')\n    self.already_iterated = True\n\n    # The invariants during execution are:\n    # 1) A worker thread always posts the sentinel as the last thing it does\n    #    before exiting.\n    # 2) We always wait for all sentinels and then join all threads.\n    num_readers_finished = 0\n    try:\n      while True:\n        try:\n          with self.read_counter:\n            element = self.element_queue.get()\n          if element is READER_THREAD_IS_DONE_SENTINEL:\n            num_readers_finished += 1\n            if num_readers_finished == self.num_reader_threads:\n              return\n          else:","sourceCodeStart":133,"sourceCodeEnd":169,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/worker/sideinputs.py#L133-L169","documentation":"PrefetchingSourceSetIterable streams a side-input source into an element queue via a reader thread, and is designed to be consumed exactly once. Calling __iter__ again after the first full iteration has started would yield an empty or partially-consumed view, so the class raises RuntimeError to protect this single-pass invariant.","triggerScenarios":"Calling iter()/for-loops twice on the same PrefetchingSourceSetIterable, or re-iterating it after it was already consumed (self.already_iterated is True). Typically happens when a DoFn's start_bundle/process accesses the side input view multiple times without re-materializing it, or Beam SDK code reuses a cached iterable.","commonSituations":"User DoFns that iterate a side-input iterable twice (e.g., once to count, once to process); custom side-input access patterns in the Fn API worker; retry/replay logic in worker code that re-reads the same view.","solutions":["Materialize the iterable into a list (list(iterable)) before consuming if you need multiple passes","Call .read() or obtain a fresh side input view for each access instead of reusing the iterable","Refactor the DoFn so the side input is iterated exactly once, accumulating results in one pass"],"exampleFix":"# before\nfor x in side_input_iterable:\n  process(x)\nfor x in side_input_iterable:  # RuntimeError\n  audit(x)\n# after\nitems = list(side_input_iterable)\nfor x in items:\n  process(x)\nfor x in items:\n  audit(x)","handlingStrategy":"validation","validationCode":"# before consuming a side input multiple times\nif isinstance(view, PrefetchingSourceSetIterable) and view.already_iterated:\n    raise RuntimeError('side input view already consumed; materialize it instead')\nview = list(view)  # single materialization, reusable passes","typeGuard":"def is_consumable(it) -> bool:\n    return not getattr(it, 'already_iterated', False)","tryCatchPattern":"try:\n    for x in side_input:\n        process(x)\nexcept RuntimeError as e:\n    if 'Can only iterate once' in str(e):\n        side_input = get_fresh_side_input_view()\n    else:\n        raise","preventionTips":["Materialize side-input iterables to list() if more than one pass is needed","Never cache and reuse PrefetchingSourceSetIterable across bundle phases","Access side inputs via the standard view interfaces (AsList/AsDict) rather than raw iterables"],"tags":["python","apache-beam","side-input","single-pass-iterable"],"backgroundTag":"invalid-state-transition","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}