PCollections of different lengths, got %v expected %v

Error message

PCollections of different lengths, got %v expected %v

What it means

passert.Floats compares observed vs expected PCollections of floats. The assertion first checks that both sides contain the same number of elements; this error fires when the lengths differ, before any value-level comparison.

Source

Thrown at sdks/go/pkg/beam/testing/passert/floats.go:85

}

type thresholdFn struct {
	Threshold float64
}

func (f *thresholdFn) ProcessElement(_ []byte, observed, expected func(*beam.T) bool) error {
	var observedValues, expectedValues []float64
	var observedInput, expectedInput beam.T
	for observed(&observedInput) {
		val := toFloat(observedInput)
		observedValues = append(observedValues, val)
	}
	for expected(&expectedInput) {
		val := toFloat(expectedInput)
		expectedValues = append(expectedValues, val)
	}
	if len(observedValues) != len(expectedValues) {
		return errors.Errorf("PCollections of different lengths, got %v expected %v", len(observedValues), len(expectedValues))
	}
	sort.Float64s(observedValues)
	sort.Float64s(expectedValues)
	var tooLow, tooHigh []string
	for i := 0; i < len(observedValues); i++ {
		delta := observedValues[i] - expectedValues[i]
		if delta > f.Threshold {
			tooHigh = append(tooHigh, fmt.Sprintf("%v > %v,", observedValues[i], expectedValues[i]))
		} else if delta < f.Threshold*-1 {
			tooLow = append(tooLow, fmt.Sprintf("%v < %v,", observedValues[i], expectedValues[i]))
		}
	}
	if len(tooLow)+len(tooHigh) == 0 {
		return nil
	}
	errorStrings := []string{}
	if len(tooLow) != 0 {
		errorStrings = append(errorStrings, fmt.Sprintf("values below expected: %v", tooLow))

View on GitHub (pinned to 12126d8942)

Solutions

Count elements on both sides and find the transform adding/removing elements; fix it.
Align expectations with intended counts if behavior legitimately changed.
Check windowing/triggering so elements are not lost or emitted twice.
Derive both PCollections from the same deterministic input.

Example fix

// before
filtered := beam.Filter(s, func(x float64) bool { return x > 0 })
passert.FloatsEquals(s, expected, filtered) // filter drops negatives expected contains
// after
out := beam.ParDo(s, normalizeFn, in) // 1:1 element counts
passert.FloatsEquals(s, expected, out)

Defensive patterns

Strategy: validation

Validate before calling

if len(observed) != len(expected) {
	t.Fatalf("element counts differ before assertion: %d vs %d", len(observed), len(expected))
}

Try / catch

if err := beamx.Execute(ctx, p); err != nil {
	if strings.Contains(err.Error(), "PCollections of different lengths") {
		// find the transform dropping/duplicating elements and fix it
	}
}

Prevention

Avoid filters or expansions in assertion input paths unless mirrored on both sides.
Use deterministic windowing in tests.
Derive expected values from the same input data.

When it happens

Trigger: passert.FloatsEquals / AllWithinBounds runs where the observed PCollection has more or fewer elements than the expected one — an element was dropped or duplicated upstream.

Common situations: A filter or expansion step changes element counts; windowing drops or duplicates elements; test expectations updated on one side only.

Related errors

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