apache/beam · error · IllegalArgumentException

must have at least one input to a KeyedPCollections

Error message

must have at least one input to a KeyedPCollections

What it means

CoGroupByKey.expand was given an empty KeyedPCollectionTuple: a cogroup by definition joins two or more tagged keyed collections, and with zero inputs there is nothing to group and no output schema could be derived. The empty KeyedPCollectionTuple input is at fault.

Source

Thrown at sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/join/CoGroupByKey.java:87

 * @param <K> the type of the keys in the input and output {@code PCollection}s
 */
public class CoGroupByKey<K>
    extends PTransform<KeyedPCollectionTuple<K>, PCollection<KV<K, CoGbkResult>>> {
  /**
   * Returns a {@code CoGroupByKey<K>} {@code PTransform}.
   *
   * @param <K> the type of the keys in the input and output {@code PCollection}s
   */
  public static <K> CoGroupByKey<K> create() {
    return new CoGroupByKey<>();
  }

  private CoGroupByKey() {}

  @Override
  public PCollection<KV<K, CoGbkResult>> expand(KeyedPCollectionTuple<K> input) {
    if (input.isEmpty()) {
      throw new IllegalArgumentException("must have at least one input to a KeyedPCollections");
    }

    // First build the union coder.
    // TODO: Look at better integration of union types with the
    // schema specified in the input.
    List<Coder<?>> codersList = new ArrayList<>();
    for (TaggedKeyedPCollection<K, ?> entry : input.getKeyedCollections()) {
      codersList.add(getValueCoder(entry.pCollection));
    }
    UnionCoder unionCoder = UnionCoder.of(codersList);
    Coder<K> keyCoder = input.getKeyCoder();
    KvCoder<K, RawUnionValue> kVCoder = KvCoder.of(keyCoder, unionCoder);

    PCollectionList<KV<K, RawUnionValue>> unionTables = PCollectionList.empty(input.getPipeline());

    // TODO: Use the schema to order the indices rather than depending
    // on the fact that the schema ordering is identical to the ordering from
    // input.getJoinCollections().

View on GitHub (pinned to 12126d8942)

Solutions

1. Check input.isEmpty() before applying CoGroupByKey and skip the transform
2. Ensure at least one KeyedPCollectionTuple.and(tag, pc) call always executes
3. Restructure the pipeline to a default branch when no inputs are available

Example fix

// before
PCollection<KV<K, CoGbkResult>> r = input.apply(CoGroupByKey.create());
// after
if (!input.isEmpty()) {
  PCollection<KV<K, CoGbkResult>> r = input.apply(CoGroupByKey.create());
}

if (input == null || input.isEmpty()) { throw new IllegalArgumentException("CoGroupByKey requires at least one input"); }

try { return input.apply(CoGroupByKey.create()); } catch (IllegalArgumentException e) { /* handle empty tuple */ }

When it happens

Trigger: Applying CoGroupByKey.create() to an empty KeyedPCollectionTuple, e.g. building the tuple in a loop that added zero PCollections (all inputs filtered out or list empty).

Common situations: Dynamic pipeline construction where collections are conditionally added; an upstream branch yielded no PCollections to join.

Understand the failure class

Background: "missing required argument" and "the following required arguments were not provided": what required-argument errors mean and how to fix them — this error's family across 20 libraries.

Related errors

• Cannot merge schemas with different numbers of fields. schem
• Attempting to call and() on a CoGbkResult apparently not cre
• PCollection does not use a KvCoder
• Unsupported type of %s: %s
• Can't convert 'null' to non-nullable field

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