{"record":{"id":"84622060f207532b","repo":"apache/beam","slug":"pcollection-must-be-of-non-composite-type-v","errorCode":null,"errorMessage":"pcollection must be of non-composite type: %v","messagePattern":"pcollection must be of non-composite type: (.+?)","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/validate.go","lineNumber":41,"sourceCode":"\t\"github.com/apache/beam/sdks/v2/go/pkg/beam/internal/errors\"\n)\n\n// ValidateKVType panics if the type of the PCollection is not KV<A,B>.\n// It returns (A,B).\nfunc ValidateKVType(col PCollection) (typex.FullType, typex.FullType) {\n\tt := col.Type()\n\tif !typex.IsKV(t) {\n\t\tpanic(fmt.Sprintf(\"pcollection must be of KV type: %v\", col))\n\t}\n\treturn t.Components()[0], t.Components()[1]\n}\n\n// ValidateNonCompositeType panics if the type of the PCollection is not a\n// composite type. It returns the type.\nfunc ValidateNonCompositeType(col PCollection) typex.FullType {\n\tt := col.Type()\n\tif typex.IsComposite(t.Type()) {\n\t\tpanic(fmt.Sprintf(\"pcollection must be of non-composite type: %v\", col))\n\t}\n\treturn t\n}\n\n// validate validates and processes the input collection and options. Private convenience\n// function.\nfunc validate(s Scope, col PCollection, opts []Option) ([]SideInput, map[string]reflect.Type, error) {\n\tif !s.IsValid() {\n\t\treturn nil, nil, errors.New(\"invalid scope\")\n\t}\n\tif !col.IsValid() {\n\t\treturn nil, nil, errors.New(\"invalid main pcollection\")\n\t}\n\tside, defs := parseOpts(opts)\n\tfor i, in := range side {\n\t\tif !in.Input.IsValid() {\n\t\t\treturn nil, nil, errors.Errorf(\"invalid side pcollection: index %v\", i)\n\t\t}","sourceCodeStart":23,"sourceCodeEnd":59,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/validate.go#L23-L59","documentation":"beam.ValidateNonCompositeType asserts that a PCollection's type is NOT composite (i.e. not KV, CoGBK, or windowed/composite constructs) and panics otherwise. Non-composite transforms like Mean, Diff, Largest, combine, and equality checks require single-component element types. The message includes the offending collection.","triggerScenarios":"Calling beam.Mean, beam.Largest, beam.Diff, beam.AllWithinBounds, TryEqualsFloat, or combine with a PCollection whose element type is a composite (e.g. KV or CoGBK) instead of a scalar/struct value.","commonSituations":"Applying a global aggregation like Mean to keyed data without extracting values first; chaining a keyed transform's output into a non-keyed aggregation without a Map step.","solutions":["Extract the value component first: beam.DropKey or a ParDo that maps KV{k,v} -> v.","Use the keyed variants (MeanPerKey, LargestPerKey) if you actually want per-key aggregation.","Inspect the PCollection type with beam.ValidateNonCompositeType in a debug path before the transform.","Restructure the pipeline so aggregations operate on single-value collections."],"exampleFix":"// before\navg := beam.Mean(s, kvCol) // kvCol is PCollection<beam.KV<K,V>>\n// after\nvals := beam.DropKey(s, kvCol)\navg := beam.Mean(s, vals)","handlingStrategy":"validation","validationCode":"if typex.IsComposite(col.Type().Type()) {\n    return fmt.Errorf(\"aggregation needs non-composite input\")\n}","typeGuard":"func isNonComposite(col beam.PCollection) bool {\n    return !typex.IsComposite(col.Type().Type())\n}","tryCatchPattern":"defer func() {\n    if r := recover(); r != nil {\n        if s, ok := r.(string); ok && strings.Contains(s, \"must be of non-composite type\") {\n            log.Fatalf(\"wrong transform input: %s\", s)\n        }\n        panic(r)\n    }\n}()","preventionTips":["Drop keys (beam.DropKey) before global aggregations on KV data","Use PerKey variants for keyed collections","Assert composite-ness in pipeline unit tests","Trace element types through each pipeline stage"],"tags":["go","panic","type-mismatch","beam-transform"],"backgroundTag":"type-mismatch","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"}