public static FusedPipeline fuse(Pipeline p) {
    return new GreedyPipelineFuser(p).fusedPipeline;
  }
@Test
  public void sideInputRootsNewStage() {
    Components components =
        Components.newBuilder()
            .putCoders("coder", Coder.newBuilder().build())
            .putCoders("windowCoder", Coder.newBuilder().build())
            .putWindowingStrategies(
                "ws", WindowingStrategy.newBuilder().setWindowCoderId("windowCoder").build())
            .putTransforms(
                "mainImpulse",
                PTransform.newBuilder()
                    .setUniqueName("MainImpulse")
                    .putOutputs("output", "mainImpulse.out")
                    .setSpec(
                        FunctionSpec.newBuilder()
                            .setUrn(PTransformTranslation.IMPULSE_TRANSFORM_URN))
                    .build())
            .putPcollections("mainImpulse.out", pc("mainImpulse.out"))
            .putTransforms(
                "read",
                PTransform.newBuilder()
                    .setUniqueName("Read")
                    .putInputs("input", "mainImpulse.out")
                    .putOutputs("output", "read.out")
                    .setSpec(
                        FunctionSpec.newBuilder()
                            .setUrn(PTransformTranslation.PAR_DO_TRANSFORM_URN)
                            .setPayload(
                                ParDoPayload.newBuilder()
                                    .setDoFn(FunctionSpec.newBuilder())
                                    .build()
                                    .toByteString()))
                    .setEnvironmentId("py")
                    .build())
            .putPcollections("read.out", pc("read.out"))
            .putTransforms(
                "sideImpulse",
                PTransform.newBuilder()
                    .setUniqueName("SideImpulse")
                    .putOutputs("output", "sideImpulse.out")
                    .setSpec(
                        FunctionSpec.newBuilder()
                            .setUrn(PTransformTranslation.IMPULSE_TRANSFORM_URN))
                    .build())
            .putPcollections("sideImpulse.out", pc("sideImpulse.out"))
            .putTransforms(
                "sideRead",
                PTransform.newBuilder()
                    .setUniqueName("SideRead")
                    .putInputs("input", "sideImpulse.out")
                    .putOutputs("output", "sideRead.out")
                    .setSpec(
                        FunctionSpec.newBuilder()
                            .setUrn(PTransformTranslation.PAR_DO_TRANSFORM_URN)
                            .setPayload(
                                ParDoPayload.newBuilder()
                                    .setDoFn(FunctionSpec.newBuilder())
                                    .build()
                                    .toByteString()))
                    .setEnvironmentId("py")
                    .build())
            .putPcollections("sideRead.out", pc("sideRead.out"))
            .putTransforms(
                "leftParDo",
                PTransform.newBuilder()
                    .setUniqueName("LeftParDo")
                    .putInputs("main", "read.out")
                    .putOutputs("output", "leftParDo.out")
                    .setSpec(
                        FunctionSpec.newBuilder()
                            .setUrn(PTransformTranslation.PAR_DO_TRANSFORM_URN)
                            .setPayload(
                                ParDoPayload.newBuilder()
                                    .setDoFn(FunctionSpec.newBuilder())
                                    .build()
                                    .toByteString())
                            .build())
                    .setEnvironmentId("py")
                    .build())
            .putPcollections("leftParDo.out", pc("leftParDo.out"))
            .putTransforms(
                "rightParDo",
                PTransform.newBuilder()
                    .setUniqueName("RightParDo")
                    .putInputs("main", "read.out")
                    .putOutputs("output", "rightParDo.out")
                    .setSpec(
                        FunctionSpec.newBuilder()
                            .setUrn(PTransformTranslation.PAR_DO_TRANSFORM_URN)
                            .setPayload(
                                ParDoPayload.newBuilder()
                                    .setDoFn(FunctionSpec.newBuilder())
                                    .build()
                                    .toByteString())
                            .build())
                    .setEnvironmentId("py")
                    .build())
            .putPcollections("rightParDo.out", pc("rightParDo.out"))
            .putTransforms(
                "sideParDo",
                PTransform.newBuilder()
                    .setUniqueName("SideParDo")
                    .putInputs("main", "read.out")
                    .putInputs("side", "sideRead.out")
                    .putOutputs("output", "sideParDo.out")
                    .setSpec(
                        FunctionSpec.newBuilder()
                            .setUrn(PTransformTranslation.PAR_DO_TRANSFORM_URN)
                            .setPayload(
                                ParDoPayload.newBuilder()
                                    .setDoFn(FunctionSpec.newBuilder())
                                    .putSideInputs("side", SideInput.getDefaultInstance())
                                    .build()
                                    .toByteString())
                            .build())
                    .setEnvironmentId("py")
                    .build())
            .putPcollections("sideParDo.out", pc("sideParDo.out"))
            .putEnvironments("py", Environments.createDockerEnvironment("py"))
            .build();

    FusedPipeline fused =
        GreedyPipelineFuser.fuse(Pipeline.newBuilder().setComponents(components).build());

    assertThat(
        fused.getRunnerExecutedTransforms(),
        containsInAnyOrder(
            PipelineNode.pTransform("mainImpulse", components.getTransformsOrThrow("mainImpulse")),
            PipelineNode.pTransform(
                "sideImpulse", components.getTransformsOrThrow("sideImpulse"))));
    assertThat(
        fused.getFusedStages(),
        containsInAnyOrder(
            ExecutableStageMatcher.withInput("mainImpulse.out")
                .withOutputs("read.out")
                .withTransforms("read"),
            ExecutableStageMatcher.withInput("read.out")
                .withNoOutputs()
                .withTransforms("leftParDo", "rightParDo"),
            ExecutableStageMatcher.withInput("read.out")
                .withSideInputs(
                    RunnerApi.ExecutableStagePayload.SideInputId.newBuilder()
                        .setTransformId("sideParDo")
                        .setLocalName("side")
                        .build())
                .withNoOutputs()
                .withTransforms("sideParDo"),
            ExecutableStageMatcher.withInput("sideImpulse.out")
                .withOutputs("sideRead.out")
                .withTransforms("sideRead")));
  }