apache/beam · error · IllegalArgumentException

PCollections come from different Pipelines

Error message

PCollections come from different Pipelines

What it means

PCollectionTuple.and verified that the appended PCollection belongs to the same Pipeline instance as the existing tuple entries; it did not, so the tuple would be internally inconsistent (a pipeline proto cannot reference PCollections from another Pipeline). The incoming PCollection from the different Pipeline is the input at fault.

Source

Thrown at sdks/java/core/src/main/java/org/apache/beam/sdk/values/PCollectionTuple.java:175

    return of(tag1, pc1, tag2, pc2, tag3, pc3, tag4, pc4).and(tag5, pc5);
  }

  // To create a PCollectionTuple with more than five inputs, use the and() builder method.

  /**
   * Returns a new {@link PCollectionTuple} that has each {@link PCollection} and {@link TupleTag}
   * of this {@link PCollectionTuple} plus the given {@link PCollection} associated with the given
   * {@link TupleTag}.
   *
   * <p>The given {@link TupleTag} should not already be mapped to a {@link PCollection} in this
   * {@link PCollectionTuple}.
   *
   * <p>Each {@link PCollection} in the resulting {@link PCollectionTuple} must be part of the same
   * {@link Pipeline}.
   */
  public <T> PCollectionTuple and(TupleTag<T> tag, PCollection<T> pc) {
    if (pc.getPipeline() != pipeline) {
      throw new IllegalArgumentException("PCollections come from different Pipelines");
    }

    return new PCollectionTuple(
        pipeline,
        new ImmutableMap.Builder<TupleTag<?>, PCollection<?>>()
            .putAll(pcollectionMap)
            .put(tag, pc)
            .build());
  }

  /**
   * A version of {@link #and(TupleTag, PCollection)} that takes in a String instead of a TupleTag.
   *
   * <p>This method is simpler for cases when a typed tuple-tag is not needed to extract a
   * PCollection, for example when using schema transforms.
   */
  public <T> PCollectionTuple and(String tag, PCollection<T> pc) {
    return and(new TupleTag<>(tag), pc);

View on GitHub (pinned to 12126d8942)

Solutions

1. Create all PCollections from the same Pipeline instance as the tuple
2. Derive the PCollection from tuple.getPipeline() when adding new outputs
3. Refactor helpers to accept and use a single Pipeline argument

Example fix

// before
tuple.and(tag, pcBuiltOnOtherPipeline); // throws
// after
PCollection<T> pc = tuple.getPipeline().apply(...);
tuple.and(tag, pc);

if (pc.getPipeline() != tuple.getPipeline()) { throw new IllegalArgumentException("wrong pipeline"); }

boolean samePipeline(PCollection<?> pc, PCollectionTuple t) { return pc.getPipeline() == t.getPipeline(); }

try { return tuple.and(tag, pc); } catch (IllegalArgumentException e) { /* re-create pc on tuple's pipeline */ }

When it happens

Trigger: Adding a PCollection from a different Pipeline instance to a PCollectionTuple via and(tag, pc), including recursive calls from mapToTuple-style helpers.

Common situations: Assembling multi-output transforms where some outputs were computed on a different Pipeline; unit tests mixing a TestPipeline with a programmatically created Pipeline.

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

• PCollections come from different Pipelines
• PCollections come from different Pipelines
• The pipeline has not been run.
• The pipeline contains abandoned PTransform(s).
• ApproximateUnique.PerKey needs an estimation error between 1

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