{"record":{"id":"1635d7f154b4b0c3","repo":"apache/beam","slug":"grouping-table-append-after-read","errorCode":null,"errorMessage":"Grouping table append after read.","messagePattern":"Grouping table append after read\\.","errorType":"exception","errorClass":"RuntimeError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/portability/fn_api_runner/execution.py","lineNumber":209,"sourceCode":"      windowing: core.Windowing) -> None:\n    self._key_coder = pre_grouped_coder.key_coder()\n    self._pre_grouped_coder = pre_grouped_coder\n    self._post_grouped_coder = post_grouped_coder\n    self._table: collections.defaultdict[bytes,\n                                         list[Any]] = collections.defaultdict(\n                                             list)\n    self._windowing = windowing\n    self._grouped_output: Optional[list[list[bytes]]] = None\n\n  def copy(self) -> 'GroupingBuffer':\n    # This is a silly temporary optimization. This class must be removed once\n    # full support for streaming is added (i.e. once we use trigger_manager for\n    # data grouping instead of GroupingBuffer).\n    return self\n\n  def append(self, elements_data: bytes) -> None:\n    if self._grouped_output:\n      raise RuntimeError('Grouping table append after read.')\n    input_stream = create_InputStream(elements_data)\n    coder_impl = self._pre_grouped_coder.get_impl()\n    key_coder_impl = self._key_coder.get_impl()\n    # TODO(robertwb): We could optimize this even more by using a\n    # window-dropping coder for the data plane.\n    is_trivial_windowing = self._windowing.is_default()\n    while input_stream.size() > 0:\n      windowed_key_value = coder_impl.decode_from_stream(input_stream, True)\n      key, value = windowed_key_value.value\n      self._table[key_coder_impl.encode(key)].append(\n          value if is_trivial_windowing else windowed_key_value.\n          with_value(value))\n\n  def extend(self, input_buffer: Buffer) -> None:\n    if isinstance(input_buffer, ListBuffer):\n      # TODO(pabloem): GroupingBuffer will be removed once shuffling is done\n      #  via state. Remove this workaround along with that.\n      return","sourceCodeStart":191,"sourceCodeEnd":227,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/portability/fn_api_runner/execution.py#L191-L227","documentation":"GroupingBuffer.append() raises RuntimeError if grouped output was already produced (self._grouped_output set). Once results have been read out of the grouping table, further raw inserts would bypass the grouping pipeline and corrupt consistency, so the append is rejected.","triggerScenarios":"Calling grouping_buffer.append(elements_data) after the grouped results have been read (e.g. after __iter__/output extraction set _grouped_output), such as late elements arriving for a finished grouping stage.","commonSituations":"Streaming-style feeds pushing data into a GroupingBuffer after it was flushed; runner code that reads grouped output and then processes remaining input bytes against the same buffer.","solutions":["Ensure all input data is appended before reading grouped output.","Create a new GroupingBuffer for data arriving after the read.","Restructure the runner loop to fully drain input before materializing grouped results."],"exampleFix":"# before\nresults = list(gbuf)\ngbuf.append(data)  # RuntimeError\n# after\ngbuf.append(data)\nresults = list(gbuf)","handlingStrategy":"type-guard","validationCode":"if gbuf._grouped_output is None:\n    gbuf.append(data)","typeGuard":"def accepts_input(gbuf):\n    return not gbuf._grouped_output","tryCatchPattern":"try:\n    gbuf.append(data)\nexcept RuntimeError:\n    new_gbuf = GroupingBuffer(...)\n    new_gbuf.append(data)","preventionTips":["Drain all input before reading grouped output.","Treat reading grouped results as finalizing the buffer.","Use a new GroupingBuffer for any post-read data."],"tags":["apache-beam","fn-api-runner","grouping","buffer-lifecycle","runtime-error"],"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-20T03:17:13.778Z"}