apache/beam · error

pcollection must be of non-composite type

Error message

pcollection must be of non-composite type: %v

What it means

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.

Solutions

1. Extract the value component first: beam.DropKey or a ParDo that maps KV{k,v} -> v.
2. Use the keyed variants (MeanPerKey, LargestPerKey) if you actually want per-key aggregation.
3. Inspect the PCollection type with beam.ValidateNonCompositeType in a debug path before the transform.
4. Restructure the pipeline so aggregations operate on single-value collections.

Example fix

// before
avg := beam.Mean(s, kvCol) // kvCol is PCollection<beam.KV<K,V>>
// after
vals := beam.DropKey(s, kvCol)
avg := beam.Mean(s, vals)

if typex.IsComposite(col.Type().Type()) {
    return fmt.Errorf("aggregation needs non-composite input")
}

func isNonComposite(col beam.PCollection) bool {
    return !typex.IsComposite(col.Type().Type())
}

defer func() {
    if r := recover(); r != nil {
        if s, ok := r.(string); ok && strings.Contains(s, "must be of non-composite type") {
            log.Fatalf("wrong transform input: %s", s)
        }
        panic(r)
    }
}()

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Related errors

• pcollection must be of KV type
• err
• Nested FullValues must be nested as pointers.
• pubsubio.Write only accepts PCollections of
• received unknown value type: want a number:, got %T

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/84622060f207532b. Report an issue: GitHub.

Appendix: source

Thrown at sdks/go/pkg/beam/validate.go:41

	"github.com/apache/beam/sdks/v2/go/pkg/beam/internal/errors"
)

// ValidateKVType panics if the type of the PCollection is not KV<A,B>.
// It returns (A,B).
func ValidateKVType(col PCollection) (typex.FullType, typex.FullType) {
	t := col.Type()
	if !typex.IsKV(t) {
		panic(fmt.Sprintf("pcollection must be of KV type: %v", col))
	}
	return t.Components()[0], t.Components()[1]
}

// ValidateNonCompositeType panics if the type of the PCollection is not a
// composite type. It returns the type.
func ValidateNonCompositeType(col PCollection) typex.FullType {
	t := col.Type()
	if typex.IsComposite(t.Type()) {
		panic(fmt.Sprintf("pcollection must be of non-composite type: %v", col))
	}
	return t
}

// validate validates and processes the input collection and options. Private convenience
// function.
func validate(s Scope, col PCollection, opts []Option) ([]SideInput, map[string]reflect.Type, error) {
	if !s.IsValid() {
		return nil, nil, errors.New("invalid scope")
	}
	if !col.IsValid() {
		return nil, nil, errors.New("invalid main pcollection")
	}
	side, defs := parseOpts(opts)
	for i, in := range side {
		if !in.Input.IsValid() {
			return nil, nil, errors.Errorf("invalid side pcollection: index %v", i)
		}

View on GitHub (pinned to 12126d8942)