{"record":{"id":"8089ebae037af09a","repo":"apache/beam","slug":"pcollections-come-from-different-pipelines","errorCode":null,"errorMessage":"PCollections come from different Pipelines","messagePattern":"PCollections come from different Pipelines","errorType":"exception","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/join/KeyedPCollectionTuple.java","lineNumber":71,"sourceCode":"  }\n\n  /**\n   * A version of {@link #of(TupleTag, PCollection)} that takes in a string instead of a TupleTag.\n   *\n   * <p>This method is simpler for cases when a typed tuple-tag is not needed to extract a\n   * PCollection, for example when using schema transforms.\n   */\n  public static <K, InputT> KeyedPCollectionTuple<K> of(String tag, PCollection<KV<K, InputT>> pc) {\n    return of(new TupleTag<>(tag), pc);\n  }\n\n  /**\n   * Returns a new {@code KeyedPCollectionTuple<K>} that is the same as this, appended with the\n   * given PCollection.\n   */\n  public <V> KeyedPCollectionTuple<K> and(TupleTag<V> tag, PCollection<KV<K, V>> pc) {\n    if (pc.getPipeline() != getPipeline()) {\n      throw new IllegalArgumentException(\"PCollections come from different Pipelines\");\n    }\n    TaggedKeyedPCollection<K, ?> wrapper = new TaggedKeyedPCollection<>(tag, pc);\n    Coder<K> myKeyCoder = keyCoder == null ? getKeyCoder(pc) : keyCoder;\n    List<TaggedKeyedPCollection<K, ?>> newKeyedCollections = copyAddLast(keyedCollections, wrapper);\n    return new KeyedPCollectionTuple<>(\n        getPipeline(), newKeyedCollections, schema.getTupleTagList().and(tag), myKeyCoder);\n  }\n\n  /**\n   * A version of {@link #and(String, PCollection)} that takes in a string instead of a TupleTag.\n   *\n   * <p>This method is simpler for cases when a typed tuple-tag is not needed to extract a\n   * PCollection, for example when using schema transforms.\n   */\n  public <V> KeyedPCollectionTuple<K> and(String tag, PCollection<KV<K, V>> pc) {\n    return and(new TupleTag<>(tag), pc);\n  }\n","sourceCodeStart":53,"sourceCodeEnd":89,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/join/KeyedPCollectionTuple.java#L53-L89","documentation":"All PCollections in a KeyedPCollectionTuple must belong to the same Pipeline object, since they will be combined into one CoGroupByKey transform. Appending a PCollection from a different Pipeline is rejected.","triggerScenarios":"Calling keyedTuple.and(tag, pc) where pc was created by a different Pipeline than the tuple's pipeline, e.g. merging results from p1.apply(...) into a tuple built on p2.","commonSituations":"Two pipelines created in the same test/program; results of an inner pipeline passed to an outer one; reusing builder objects across Pipeline instances.","solutions":["Create all inputs and the KeyedPCollectionTuple from the same Pipeline instance","If combining data from two pipelines, write one side to a data source and re-read it in the other pipeline before joining","Pass the same Pipeline (e.g. p) when constructing both the tuple and the PCollections"],"exampleFix":"// before\nPipeline p1 = Pipeline.create(); Pipeline p2 = Pipeline.create();\nKeyedPCollectionTuple.of(tag, p1.apply(...)).and(tag2, p2.apply(...)); // throws\n// after\nKeyedPCollectionTuple.of(tag, p1.apply(...)).and(tag2, p1.apply(...));","handlingStrategy":"validation","validationCode":"if (pc.getPipeline() != tuple.getPipeline()) { throw new IllegalArgumentException(\"PCollection pipeline mismatch\"); }","typeGuard":null,"tryCatchPattern":"try { return tuple.and(tag, pc); } catch (IllegalArgumentException e) { throw new IllegalStateException(\"Cannot join across pipelines\", e); }","preventionTips":["Use a single Pipeline instance for all join inputs","Never mix PCollections from p1 and p2 in one tuple","Refactor multi-pipeline flows to serialize/re-read data between pipelines"],"tags":["beam","pipeline","keyedpcollectiontuple","cross-pipeline"],"backgroundTag":"invalid-argument-value","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}