{"record":{"id":"e7e5ed78e35d9b37","repo":"apache/beam","slug":"processelement-uses-a-stateprovider-but-is-not-keyed-all","errorCode":null,"errorMessage":"ProcessElement uses a StateProvider, but is not keyed. All stateful DoFns must take a key/value pair as an input.","messagePattern":"ProcessElement uses a StateProvider, but is not keyed\\. All stateful DoFns must take a key/value pair as an input\\.","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/graph/fn.go","lineNumber":1352,"sourceCode":"\t\t\tif method.Ret[0].T != watermarkStateT {\n\t\t\t\terr := errors.Errorf(\"mismatched output type in method %v, return %v. got: %v, want: %v\",\n\t\t\t\t\twatermarkEstimatorStateName, 0, method.Ret[0].T, watermarkStateT)\n\t\t\t\treturn errors.SetTopLevelMsgf(err, \"mismatched output type in method %v, \"+\n\t\t\t\t\t\"return value at index %v got: %v, want: %v (from method %v). \"+\n\t\t\t\t\t\"Ensure that all watermark estimators in an SDF are the same type.\",\n\t\t\t\t\twatermarkEstimatorStateName, 0, method.Ret[0].T, watermarkStateT, watermarkEstimatorStateName)\n\t\t\t}\n\t\t}\n\t}\n\treturn nil\n}\n\nfunc validateState(fn *DoFn, numIn mainInputs) error {\n\tps := fn.PipelineState()\n\n\tif _, hasSp := fn.methods[processElementName].StateProvider(); hasSp {\n\t\tif numIn == MainSingle {\n\t\t\terr := errors.Errorf(\"ProcessElement uses a StateProvider, but is not keyed\")\n\t\t\treturn errors.SetTopLevelMsgf(err, \"ProcessElement uses a StateProvider, but is not keyed. \"+\n\t\t\t\t\"All stateful DoFns must take a key/value pair as an input.\")\n\t\t}\n\t\tif len(ps) == 0 {\n\t\t\terr := errors.Errorf(\"ProcessElement uses a StateProvider, but noState structs are attached to the DoFn\")\n\t\t\treturn errors.SetTopLevelMsgf(err, \"ProcessElement uses a StateProvider, but no State structs are \"+\n\t\t\t\t\"attached to the DoFn. Ensure that you are including the State structs you're using to read/write\"+\n\t\t\t\t\"global state as public uppercase member variables.\")\n\t\t}\n\t\tstateKeys := make(map[string]state.PipelineState)\n\t\tfor _, s := range ps {\n\t\t\tk := s.StateKey()\n\t\t\tif orig, ok := stateKeys[k]; ok {\n\t\t\t\terr := errors.Errorf(\"Duplicate state key %v\", k)\n\t\t\t\treturn errors.SetTopLevelMsgf(err, \"Duplicate state key %v used by %v and %v. Ensure that state keys are\"+\n\t\t\t\t\t\"unique per DoFn\", k, orig, s)\n\t\t\t}\n\t\t\tt := s.StateType()","sourceCodeStart":1334,"sourceCodeEnd":1370,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/graph/fn.go#L1334-L1370","documentation":"Beam Go stateful DoFns must operate on keyed data (KV pairs) so state can be scoped per key. This error is thrown during DoFn validation when ProcessElement declares a StateProvider parameter but the DoFn's input is a single element rather than a key/value pair. Without a key, Beam has no way to partition the state.","triggerScenarios":"Declaring a state.Provider (or StateProvider) parameter in ProcessElement while the DoFn's input element is not a KV (two-element struct with K/V fields), e.g. ProcessElement(ctx, sp state.Provider, value string) where input is a plain string.","commonSituations":"Adding state to an existing stateless DoFn without changing the input to KV; copying stateful example code into a DoFn that consumes unkeyed PCollections.","solutions":["Key the input: change ProcessElement to accept a KV[K, V] and apply beam.ParDo over a KVs PCollection (e.g. via beam.KV or beam.AddFixedKey).","Remove the StateProvider parameter if state isn't actually needed.","Verify the upstream PCollection is keyed (KV[K,V]) before the stateful DoFn."],"exampleFix":"// before\nfunc (fn *CountFn) ProcessElement(ctx context.Context, sp state.Provider, word string) error { ... }\n// after\nfunc (fn *CountFn) ProcessElement(ctx context.Context, sp state.Provider, w beam.KV[string, int]) error { ... }\n// and pipe a keyed PCollection: beam.ParDo(s, &CountFn{}, beam.AddFixedKey(s, input))","handlingStrategy":"validation","validationCode":"// before running: ensure input is keyed\nif _, ok := input.(beam.KV[string, string]); !ok {\n    input = beam.AddFixedKey(s, input)\n}","typeGuard":"func isKV(v interface{}) bool { _, ok := v.(beam.KV[interface{}, interface{}]); return ok }","tryCatchPattern":"err := beam.Run(ctx, p)\nif err != nil && strings.Contains(err.Error(), \"uses a StateProvider, but is not keyed\") {\n    log.Fatalf(\"stateful DoFn %T needs a KV input: %v\", dofn, err)\n}","preventionTips":["Always key PCollections feeding stateful DoFns (beam.AddFixedKey or explicit KV)","Keep stateful DoFn ProcessElement signatures in a team convention doc","Validate DoFns in unit tests before pipeline runs"],"tags":["go","apache-beam","dofn","state","keyed-input"],"backgroundTag":"missing-required-argument","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"}