public static FusedPipeline fuse(Pipeline p) {
    return new GreedyPipelineFuser(p).fusedPipeline;
  }
@Test
  public void flattenWithHeterogeneousInputsSingleEnvOutputPartiallyMaterialized() {
    Components components =
        Components.newBuilder()
            .putCoders("coder", Coder.newBuilder().build())
            .putCoders("windowCoder", Coder.newBuilder().build())
            .putWindowingStrategies(
                "ws", WindowingStrategy.newBuilder().setWindowCoderId("windowCoder").build())
            .putTransforms(
                "pyImpulse",
                PTransform.newBuilder()
                    .setUniqueName("PyImpulse")
                    .putOutputs("output", "pyImpulse.out")
                    .setSpec(
                        FunctionSpec.newBuilder()
                            .setUrn(PTransformTranslation.IMPULSE_TRANSFORM_URN))
                    .build())
            .putPcollections("pyImpulse.out", pc("pyImpulse.out"))
            .putTransforms(
                "pyRead",
                PTransform.newBuilder()
                    .setUniqueName("PyRead")
                    .putInputs("input", "pyImpulse.out")
                    .putOutputs("output", "pyRead.out")
                    .setSpec(
                        FunctionSpec.newBuilder()
                            .setUrn(PTransformTranslation.PAR_DO_TRANSFORM_URN)
                            .setPayload(
                                ParDoPayload.newBuilder()
                                    .setDoFn(FunctionSpec.newBuilder())
                                    .build()
                                    .toByteString()))
                    .setEnvironmentId("py")
                    .build())
            .putPcollections("pyRead.out", pc("pyRead.out"))
            .putTransforms(
                "goImpulse",
                PTransform.newBuilder()
                    .setUniqueName("GoImpulse")
                    .putOutputs("output", "goImpulse.out")
                    .setSpec(
                        FunctionSpec.newBuilder()
                            .setUrn(PTransformTranslation.IMPULSE_TRANSFORM_URN))
                    .build())
            .putPcollections("goImpulse.out", pc("goImpulse.out"))
            .putTransforms(
                "goRead",
                PTransform.newBuilder()
                    .setUniqueName("GoRead")
                    .putInputs("input", "goImpulse.out")
                    .putOutputs("output", "goRead.out")
                    .setSpec(
                        FunctionSpec.newBuilder()
                            .setUrn(PTransformTranslation.PAR_DO_TRANSFORM_URN)
                            .setPayload(
                                ParDoPayload.newBuilder()
                                    .setDoFn(FunctionSpec.newBuilder())
                                    .build()
                                    .toByteString()))
                    .setEnvironmentId("go")
                    .build())
            .putPcollections("goRead.out", pc("goRead.out"))
            .putTransforms(
                "flatten",
                PTransform.newBuilder()
                    .setUniqueName("Flatten")
                    .putInputs("goReadInput", "goRead.out")
                    .putInputs("pyReadInput", "pyRead.out")
                    .putOutputs("output", "flatten.out")
                    .setSpec(
                        FunctionSpec.newBuilder()
                            .setUrn(PTransformTranslation.FLATTEN_TRANSFORM_URN))
                    .build())
            .putPcollections("flatten.out", pc("flatten.out"))
            .putTransforms(
                "goParDo",
                PTransform.newBuilder()
                    .setUniqueName("GoParDo")
                    .putInputs("input", "flatten.out")
                    .putOutputs("output", "goParDo.out")
                    .setSpec(
                        FunctionSpec.newBuilder()
                            .setUrn(PTransformTranslation.PAR_DO_TRANSFORM_URN)
                            .setPayload(
                                ParDoPayload.newBuilder()
                                    .setDoFn(FunctionSpec.newBuilder())
                                    .build()
                                    .toByteString()))
                    .setEnvironmentId("go")
                    .build())
            .putPcollections("goParDo.out", pc("goParDo.out"))
            .putEnvironments("go", Environments.createDockerEnvironment("go"))
            .putEnvironments("py", Environments.createDockerEnvironment("py"))
            .build();
    FusedPipeline fused =
        GreedyPipelineFuser.fuse(Pipeline.newBuilder().setComponents(components).build());

    assertThat(
        fused.getRunnerExecutedTransforms(),
        containsInAnyOrder(
            PipelineNode.pTransform("pyImpulse", components.getTransformsOrThrow("pyImpulse")),
            PipelineNode.pTransform("goImpulse", components.getTransformsOrThrow("goImpulse"))));

    assertThat(
        fused.getFusedStages(),
        containsInAnyOrder(
            ExecutableStageMatcher.withInput("goImpulse.out")
                .withNoOutputs()
                .withTransforms("goRead", "flatten", "goParDo"),
            ExecutableStageMatcher.withInput("pyImpulse.out")
                .withOutputs("flatten.out")
                .withTransforms("pyRead", "flatten"),
            ExecutableStageMatcher.withInput("flatten.out")
                .withNoOutputs()
                .withTransforms("goParDo")));
  }