GroupIntoBatchesWithShardedKey: input PCollection must be KV

Error message

GroupIntoBatchesWithShardedKey: input PCollection must be KV-typed; got %v

What it means

GroupIntoBatchesWithShardedKey is generic over key type K and requires the input PCollection's KV key component to match K exactly. A mismatch cannot be reconciled, so the transform panics with both types in the message. Re-shape the input or supply the correct type parameter.

Source

Thrown at sdks/go/pkg/beam/transforms/batch/batch.go:663

// GroupIntoBatchesWithShardedKey wraps each user key with a
// ShardedKey{Key: K, ShardID: [24]byte} and then applies
// GroupIntoBatches. Output is PCollection<KV<ShardedKey[K], []V>>.
//
// The key type K must have been registered via
// RegisterShardedKeyType[K] at init time. Common types (string,
// []byte, int, int64) are registered automatically.
//
// Sharding spreads the processing of a single hot logical key across
// multiple workers: each shard is independent state, so distributed
// runners can parallelize without the user's key type changing.
func GroupIntoBatchesWithShardedKey[K any](s beam.Scope, params Params, col beam.PCollection) beam.PCollection {
	s = s.Scope("batch.GroupIntoBatchesWithShardedKey")

	if err := params.validate(); err != nil {
		panic(fmt.Errorf("GroupIntoBatchesWithShardedKey: %w", err))
	}
	if !typex.IsKV(col.Type()) {
		panic(fmt.Errorf(
			"GroupIntoBatchesWithShardedKey: input PCollection must be KV-typed; got %v",
			col.Type()))
	}
	keyFT := col.Type().Components()[0]
	var zero K
	if keyFT.Type() != reflect.TypeOf(zero) {
		panic(fmt.Errorf(
			"GroupIntoBatchesWithShardedKey: type parameter K (%v) does not match input key type (%v)",
			reflect.TypeOf(zero), keyFT.Type()))
	}

	wrapped := beam.ParDo(s, &wrapShardedKeyFn[K]{}, col)
	return GroupIntoBatches(s, params, wrapped)
}

View on GitHub (pinned to 12126d8942)

Solutions

Make the input a KV with key type equal to K, or change K to match keyFT
Check col.Type().Components()[0] before calling
Insert a ParDo to convert keys to the expected type
Pin K explicitly at the call site instead of relying on inference

Example fix

// before
batch.GroupIntoBatchesWithShardedKey[string](s, params, intKeyed) // keys are int
// after
batch.GroupIntoBatchesWithShardedKey[int](s, params, intKeyed)

Defensive patterns

Strategy: type-guard

Validate before calling

var zero K
if col.Type().Components()[0].Type() != reflect.TypeOf(zero) {
    return fmt.Errorf("key type mismatch")
}

Type guard

func keyMatches[K any](col beam.PCollection) bool {
    var zero K
    return col.Type().Components()[0].Type() == reflect.TypeOf(zero)
}

Prevention

Specify K explicitly at call sites
Check component types before generic calls
Test generic helpers with representative key types

When it happens

Trigger: Calling GroupIntoBatchesWithShardedKey[string](s, params, kvCol) where kvCol's key component is int (or another type), failing keyFT.Type() != reflect.TypeOf(zero K).

Common situations: Letting Go infer K incorrectly, or feeding a KV collection whose key type changed upstream after a refactor.

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

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