{"record":{"id":"62e826d4ca5cc681","repo":"apache/beam","slug":"invalid-state-spec-s","errorCode":null,"errorMessage":"Invalid state spec: %s","messagePattern":"Invalid state spec: (.+?)","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/direct/direct_userstate.py","lineNumber":49,"sourceCode":"  def __init__(self, state_spec, state_tag, current_value_accessor):\n    self._state_spec = state_spec\n    self._state_tag = state_tag\n    self._current_value_accessor = current_value_accessor\n\n  @staticmethod\n  def for_spec(state_spec, state_tag, current_value_accessor):\n    if isinstance(state_spec, userstate.ReadModifyWriteStateSpec):\n      return ReadModifyWriteRuntimeState(\n          state_spec, state_tag, current_value_accessor)\n    elif isinstance(state_spec, userstate.BagStateSpec):\n      return BagRuntimeState(state_spec, state_tag, current_value_accessor)\n    elif isinstance(state_spec, userstate.CombiningValueStateSpec):\n      return CombiningValueRuntimeState(\n          state_spec, state_tag, current_value_accessor)\n    elif isinstance(state_spec, userstate.SetStateSpec):\n      return SetRuntimeState(state_spec, state_tag, current_value_accessor)\n    else:\n      raise ValueError('Invalid state spec: %s' % state_spec)\n\n  def _encode(self, value):\n    return self._state_spec.coder.encode(value)\n\n  def _decode(self, value):\n    return self._state_spec.coder.decode(value)\n\n\n# Sentinel designating an unread value.\nUNREAD_VALUE = object()\n\n\nclass ReadModifyWriteRuntimeState(DirectRuntimeState,\n                                  userstate.ReadModifyWriteRuntimeState):\n  def __init__(self, state_spec, state_tag, current_value_accessor):\n    super().__init__(state_spec, state_tag, current_value_accessor)\n    self._value = UNREAD_VALUE\n    self._cleared = False","sourceCodeStart":31,"sourceCodeEnd":67,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/direct/direct_userstate.py#L31-L67","documentation":"RuntimeStateFactory.for_spec maps userstate state specs (BagStateSpec, CombiningValueStateSpec, SetStateSpec, etc.) to direct-runner runtime state implementations. A state_spec of an unsupported type reaches the final else branch and raises ValueError. It usually means a new/unsupported state spec kind was used with the direct runner's streaming (TimelyQueue) execution.","triggerScenarios":"Using DoFn state (e.g. ReadModifyWriteStateSpec, OrderedListStateSpec, or a custom UserStateSpec subclass) in a streaming DirectRunner pipeline where for_spec has no mapping for that spec type.","commonSituations":"Adopting a newer state API before the direct runner implements it; custom state spec implementations; running a pipeline locally that was built for a different runner.","solutions":["Use a supported state spec: BagStateSpec, CombiningValueStateSpec, or SetStateSpec.","Switch to a runner that supports the state spec you need (e.g. DataflowRunner or FlinkRunner).","Upgrade apache-beam in case support for the spec was added in a newer release."],"exampleFix":"// before\nspec = ReadModifyWriteStateSpec('cache', coder)  # unmapped -> ValueError\n// after\nspec = userstate.BagStateSpec('cache', coder)  # or another supported spec","handlingStrategy":"validation","validationCode":"SUPPORTED_SPECS = (userstate.BagStateSpec, userstate.CombiningValueStateSpec, userstate.SetStateSpec)\nif not isinstance(state_spec, SUPPORTED_SPECS):\n    raise ValueError(f'DirectRunner does not support state spec: {state_spec!r}')","typeGuard":"def is_supported_state_spec(spec) -> bool:\n    import apache_beam.transforms.userstate as us\n    return isinstance(spec, (us.BagStateSpec, us.CombiningValueStateSpec, us.SetStateSpec))","tryCatchPattern":"try:\n    run_streaming_pipeline()\nexcept ValueError as e:\n    if 'Invalid state spec' in str(e):\n        sys.exit('Restrict DoFn state to bag/combining/set specs on DirectRunner')\n    raise","preventionTips":["Limit DoFn state APIs to bag/combining/set when targeting the direct runner.","Test stateful DoFns on the target runner early.","Keep custom state spec usage behind a runner-conditional path."],"tags":["python","apache-beam","direct-runner","user-state"],"backgroundTag":"invalid-argument-value","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"}