static DeduplicationResult ensureSingleProducer(
      QueryablePipeline pipeline,
      Collection<ExecutableStage> stages,
      Collection<PipelineNode.PTransformNode> unfusedTransforms) {
    RunnerApi.Components.Builder unzippedComponents = pipeline.getComponents().toBuilder();

    Multimap<PipelineNode.PCollectionNode, StageOrTransform> pcollectionProducers =
        getProducers(pipeline, stages, unfusedTransforms);
    Multimap<StageOrTransform, PipelineNode.PCollectionNode> requiresNewOutput =
        HashMultimap.create();
    // Create a synthetic PCollection for each of these nodes. The transforms in the runner
    // portion of the graph that creates them should be replaced in the result components. The
    // ExecutableStage must also be rewritten to have updated outputs and transforms.
    for (Map.Entry<PipelineNode.PCollectionNode, Collection<StageOrTransform>> collectionProducer :
        pcollectionProducers.asMap().entrySet()) {
      if (collectionProducer.getValue().size() > 1) {
        for (StageOrTransform producer : collectionProducer.getValue()) {
          requiresNewOutput.put(producer, collectionProducer.getKey());
        }
      }
    }

    Map<ExecutableStage, ExecutableStage> updatedStages = new LinkedHashMap<>();
    Map<String, PipelineNode.PTransformNode> updatedTransforms = new LinkedHashMap<>();
    Multimap<String, PipelineNode.PCollectionNode> originalToPartial = HashMultimap.create();
    for (Map.Entry<StageOrTransform, Collection<PipelineNode.PCollectionNode>>
        deduplicationTargets : requiresNewOutput.asMap().entrySet()) {
      if (deduplicationTargets.getKey().getStage() != null) {
        StageDeduplication deduplication =
            deduplicatePCollections(
                deduplicationTargets.getKey().getStage(),
                deduplicationTargets.getValue(),
                unzippedComponents::containsPcollections);
        for (Entry<String, PipelineNode.PCollectionNode> originalToPartialReplacement :
            deduplication.getOriginalToPartialPCollections().entrySet()) {
          originalToPartial.put(
              originalToPartialReplacement.getKey(), originalToPartialReplacement.getValue());
          unzippedComponents.putPcollections(
              originalToPartialReplacement.getValue().getId(),
              originalToPartialReplacement.getValue().getPCollection());
        }
        updatedStages.put(
            deduplicationTargets.getKey().getStage(), deduplication.getUpdatedStage());
      } else if (deduplicationTargets.getKey().getTransform() != null) {
        PTransformDeduplication deduplication =
            deduplicatePCollections(
                deduplicationTargets.getKey().getTransform(),
                deduplicationTargets.getValue(),
                unzippedComponents::containsPcollections);
        for (Entry<String, PipelineNode.PCollectionNode> originalToPartialReplacement :
            deduplication.getOriginalToPartialPCollections().entrySet()) {
          originalToPartial.put(
              originalToPartialReplacement.getKey(), originalToPartialReplacement.getValue());
          unzippedComponents.putPcollections(
              originalToPartialReplacement.getValue().getId(),
              originalToPartialReplacement.getValue().getPCollection());
        }
        updatedTransforms.put(
            deduplicationTargets.getKey().getTransform().getId(),
            deduplication.getUpdatedTransform());
      } else {
        throw new IllegalStateException(
            String.format(
                "%s with no %s or %s",
                StageOrTransform.class.getSimpleName(),
                ExecutableStage.class.getSimpleName(),
                PipelineNode.PTransformNode.class.getSimpleName()));
      }
    }

    Set<PipelineNode.PTransformNode> introducedFlattens = new LinkedHashSet<>();
    for (Map.Entry<String, Collection<PipelineNode.PCollectionNode>> partialFlattenTargets :
        originalToPartial.asMap().entrySet()) {
      String flattenId =
          SyntheticComponents.uniqueId("unzipped_flatten", unzippedComponents::containsTransforms);
      PTransform flattenPartialPCollections =
          createFlattenOfPartials(
              flattenId, partialFlattenTargets.getKey(), partialFlattenTargets.getValue());
      unzippedComponents.putTransforms(flattenId, flattenPartialPCollections);
      introducedFlattens.add(PipelineNode.pTransform(flattenId, flattenPartialPCollections));
    }

    Components components = unzippedComponents.build();
    return DeduplicationResult.of(components, introducedFlattens, updatedStages, updatedTransforms);
  }
@Test
  public void unchangedWithNoDuplicates() {
    /* When all the PCollections are produced by only one transform or stage, the result should be
     * empty/identical to the input.
     *
     * Pipeline:
     *              /-> one -> .out \
     * red -> .out ->                -> blue -> .out
     *              \-> two -> .out /
     */
    PCollection redOut = PCollection.newBuilder().setUniqueName("red.out").build();
    PTransform red =
        PTransform.newBuilder()
            .setSpec(
                FunctionSpec.newBuilder()
                    .setUrn(PTransformTranslation.PAR_DO_TRANSFORM_URN)
                    .build())
            .putOutputs("out", redOut.getUniqueName())
            .build();
    PCollection oneOut = PCollection.newBuilder().setUniqueName("one.out").build();
    PTransform one =
        PTransform.newBuilder()
            .setSpec(
                FunctionSpec.newBuilder()
                    .setUrn(PTransformTranslation.PAR_DO_TRANSFORM_URN)
                    .build())
            .putInputs("in", redOut.getUniqueName())
            .putOutputs("out", oneOut.getUniqueName())
            .build();
    PCollection twoOut = PCollection.newBuilder().setUniqueName("two.out").build();
    PTransform two =
        PTransform.newBuilder()
            .setSpec(
                FunctionSpec.newBuilder()
                    .setUrn(PTransformTranslation.PAR_DO_TRANSFORM_URN)
                    .build())
            .putInputs("in", redOut.getUniqueName())
            .putOutputs("out", twoOut.getUniqueName())
            .build();
    PCollection blueOut = PCollection.newBuilder().setUniqueName("blue.out").build();
    PTransform blue =
        PTransform.newBuilder()
            .setSpec(
                FunctionSpec.newBuilder()
                    .setUrn(PTransformTranslation.PAR_DO_TRANSFORM_URN)
                    .build())
            .putInputs("one", oneOut.getUniqueName())
            .putInputs("two", twoOut.getUniqueName())
            .putOutputs("out", blueOut.getUniqueName())
            .build();
    RunnerApi.Components components =
        Components.newBuilder()
            .putTransforms("one", one)
            .putPcollections(oneOut.getUniqueName(), oneOut)
            .putTransforms("two", two)
            .putPcollections(twoOut.getUniqueName(), twoOut)
            .putTransforms("red", red)
            .putPcollections(redOut.getUniqueName(), redOut)
            .putTransforms("blue", blue)
            .putPcollections(blueOut.getUniqueName(), blueOut)
            .build();
    ExecutableStage oneStage =
        ImmutableExecutableStage.of(
            components,
            Environment.getDefaultInstance(),
            PipelineNode.pCollection(redOut.getUniqueName(), redOut),
            ImmutableList.of(),
            ImmutableList.of(),
            ImmutableList.of(),
            ImmutableList.of(PipelineNode.pTransform("one", one)),
            ImmutableList.of(PipelineNode.pCollection(oneOut.getUniqueName(), oneOut)),
            DEFAULT_WIRE_CODER_SETTINGS);
    ExecutableStage twoStage =
        ImmutableExecutableStage.of(
            components,
            Environment.getDefaultInstance(),
            PipelineNode.pCollection(redOut.getUniqueName(), redOut),
            ImmutableList.of(),
            ImmutableList.of(),
            ImmutableList.of(),
            ImmutableList.of(PipelineNode.pTransform("two", two)),
            ImmutableList.of(PipelineNode.pCollection(twoOut.getUniqueName(), twoOut)),
            DEFAULT_WIRE_CODER_SETTINGS);
    PTransformNode redTransform = PipelineNode.pTransform("red", red);
    PTransformNode blueTransform = PipelineNode.pTransform("blue", blue);
    QueryablePipeline pipeline = QueryablePipeline.forPrimitivesIn(components);
    OutputDeduplicator.DeduplicationResult result =
        OutputDeduplicator.ensureSingleProducer(
            pipeline,
            ImmutableList.of(oneStage, twoStage),
            ImmutableList.of(redTransform, blueTransform));

    assertThat(result.getDeduplicatedComponents(), equalTo(components));
    assertThat(result.getDeduplicatedStages().keySet(), empty());
    assertThat(result.getDeduplicatedTransforms().keySet(), empty());
    assertThat(result.getIntroducedTransforms(), empty());
  }