What causes "invalid pcollection to flatten: index %v" in apache/beam?

Inspect the PCollection at the reported index and assign it the output of a real transform.

invalid pcollection to flatten: index

Error message

invalid pcollection to flatten: index %v

What it means

TryFlatten (sdks/go/pkg/beam/flatten.go) merges several PCollections; before building the flatten node it verifies each input is a valid PCollection. An invalid (zero-value or wrongly-scoped) input at position i triggers "invalid pcollection to flatten: index i". It prevents silently flattening uninitialized collections.

Solutions

Inspect the PCollection at the reported index and assign it the output of a real transform.
Guard each input with in.IsValid() before calling Flatten.
Use beam.TryFlatten to get an error instead of a panic from Flatten.
Collect PCollections into a slice only from successful transform results.

Example fix

// before
var cols []beam.PCollection
var maybe beam.PCollection // never assigned on some path
cols = append(cols, maybe)
beam.Flatten(s, cols...)
// after
if !maybe.IsValid() { return errors.New("missing branch output") }
cols = append(cols, maybe)
out, err := beam.TryFlatten(s, cols...)

Defensive patterns

Strategy: validation

Validate before calling

for i, c := range cols {
    if !c.IsValid() {
        return fmt.Errorf("flatten input %d is invalid", i)
    }
}

Try / catch

out, err := beam.TryFlatten(s, cols...)
if err != nil {
    return fmt.Errorf("flatten failed: %w", err)
}

Prevention

Check every conditional branch assigns the PCollection before flattening
Avoid declaring PCollections with var and appending unconditionally
Build input slices only from transform outputs

When it happens

Trigger: Calling beam.Flatten (or TryFlatten) with a zero-value beam.PCollection{} among the cols arguments, e.g. an uninitialized variable in a variadic call.

Common situations: Building inputs in a loop with a declared-outside var that never gets assigned; conditionally assigning a PCollection but passing it regardless of the branch taken.

Understand the failure class

Background: "Must be a positive integer", "Invalid value", "Unsupported": the invalid-argument-value error family, when a library rejects the value you pass — this error's family across 35 libraries.

Related errors

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

Appendix: source

Thrown at sdks/go/pkg/beam/flatten.go:43

// all the input PCollections. The name "Flatten" suggests taking a list of lists
// and flattening them into a single list.
//
// By default, the Coder of the output PCollection is the same as the Coder
// of the first PCollection.
func Flatten(s Scope, cols ...PCollection) PCollection {
	return Must(TryFlatten(s, cols...))
}

// TryFlatten merges incoming PCollections of type 'A' to a single PCollection
// of type 'A'. Returns an error indicating the set of PCollections that could
// not be flattened.
func TryFlatten(s Scope, cols ...PCollection) (PCollection, error) {
	if !s.IsValid() {
		return PCollection{}, errors.New("invalid scope")
	}
	for i, in := range cols {
		if !in.IsValid() {
			return PCollection{}, errors.Errorf("invalid pcollection to flatten: index %v", i)
		}
	}
	if len(cols) == 0 {
		return PCollection{}, errors.New("no input pcollections")
	}
	if len(cols) == 1 {
		return cols[0], nil // no-op
	}

	var in []*graph.Node
	for _, s := range cols {
		in = append(in, s.n)
	}
	edge, err := graph.NewFlatten(s.real, s.scope, in)
	if err != nil {
		return PCollection{}, err
	}
	ret := PCollection{edge.Output[0].To}

View on GitHub (pinned to 12126d8942)