{"record":{"id":"becc8cba7d83b189","repo":"apache/beam","slug":"cannot-read-state-written-iterable-without-state-reader","errorCode":null,"errorMessage":"Cannot read state-written iterable without state reader.","messagePattern":"Cannot read state-written iterable without state reader\\.","errorType":"exception","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/coders/coder_impl.py","lineNumber":1369,"sourceCode":"    # type: (create_InputStream, bool) -> Sequence\n    size = in_stream.read_bigendian_int32()\n\n    if size >= 0:\n      elements = [\n          self._elem_coder.decode_from_stream(in_stream, True)\n          for _ in range(size)\n      ]  # type: Iterable[Any]\n    else:\n      elements = []\n      count = in_stream.read_var_int64()\n      while count > 0:\n        for _ in range(count):\n          elements.append(self._elem_coder.decode_from_stream(in_stream, True))\n        count = in_stream.read_var_int64()\n\n      if count == -1:\n        if self._read_state is None:\n          raise ValueError(\n              'Cannot read state-written iterable without state reader.')\n\n        state_token = in_stream.read_all(True)\n        elements = _ConcatSequence(\n            elements, self._read_state(state_token, self._elem_coder))\n\n    return self._construct_from_sequence(elements)\n\n  def estimate_size(self, value, nested=False):\n    # type: (Any, bool) -> int\n\n    \"\"\"Estimates the encoded size of the given value, in bytes.\"\"\"\n    # TODO(ccy): This ignores element sizes.\n    estimated_size, _ = (self.get_estimated_size_and_observables(value))\n    return estimated_size\n\n  def get_estimated_size_and_observables(self, value, nested=False):\n    # type: (Any, bool) -> Tuple[int, Observables]","sourceCodeStart":1351,"sourceCodeEnd":1387,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/coders/coder_impl.py#L1351-L1387","documentation":"IterableCoderImpl.decode_from_stream supports a streaming encoding where a count of -1 means the remaining elements live in runner-managed state, accessed via a state reader supplied at construction (self._read_state). If a -1 marker appears in the byte stream but no state reader was configured, Beam raises ValueError because the state-backed portion of the iterable cannot be materialized.","triggerScenarios":"Decoding an iterable encoded with the state-written (count == -1) format using an IterableCoderImpl created without a state reader — e.g., decoding state-protocol bytes with a plain coder, or a test harness/direct pipeline path lacking the state reader wiring.","commonSituations":"Replaying or unit-testing encoded iterables captured from a streaming runner that uses the state API; decoding Fn Harness state protocol payloads outside the runner; Beam version/config mismatch where state-backed iterables are produced but the local decode path has no _read_state.","solutions":["Decode such iterables in the context that provides a state reader (runner/Fn Harness state API), not with a standalone coder","Use the non-state (materialized) iterable encoding for offline decoding — fully materialize elements on the writer side","If constructing IterableCoderImpl manually, pass a read_state callable that can resolve state tokens","Check pipeline options / runner support so state-backed coders are only used where the state client exists"],"exampleFix":"// before\ncoder_impl = IterableCoderImpl(VarIntCoder().get_impl())  # no state reader\ncoder_impl.decode_from_stream(stream_with_state_marker, True)  # ValueError\n// after\ncoder_impl = IterableCoderImpl(\n    VarIntCoder().get_impl(),\n    read_state=lambda token, elem_coder: fetch_state_elements(token, elem_coder))\ncoder_impl.decode_from_stream(stream_with_state_marker, True)","handlingStrategy":"try-catch","validationCode":"def is_state_marked_encoded(stream) -> bool:\n    # cannot peek portably; instead check construction: coder needs _read_state for state bytes\n    return getattr(coder_impl, '_read_state', None) is not None","typeGuard":"def supports_state_read(coder_impl) -> bool:\n    return getattr(coder_impl, '_read_state', None) is not None","tryCatchPattern":"try:\n    values = coder_impl.decode_from_stream(stream, True)\nexcept ValueError as e:\n    if 'state reader' in str(e):\n        log.error('Decode state-backed iterable in runner context, not standalone: %s', e)\n        values = []\n    else:\n        raise","preventionTips":["Only use state-backed iterable coders inside a runner/Fn Harness that supplies a state reader","For offline decoding, use fully materialized (count-prefixed) iterables","When constructing coder impls manually in tests, always wire a read_state callable if state bytes are possible"],"tags":["python","apache-beam","coding","iterable","state-api"],"backgroundTag":"missing-dependency","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"}