public static List<Transformation<?>> optimize(List<Transformation<?>> transformations) {
        final Map<Transformation<?>, Set<Transformation<?>>> outputMap =
                buildOutputMap(transformations);

        final LinkedHashSet<Transformation<?>> chainedTransformations = new LinkedHashSet<>();
        final Set<Transformation<?>> alreadyTransformed = Sets.newIdentityHashSet();
        final Queue<Transformation<?>> toTransformQueue = Queues.newArrayDeque(transformations);
        while (!toTransformQueue.isEmpty()) {
            final Transformation<?> transformation = toTransformQueue.poll();
            if (!alreadyTransformed.contains(transformation)) {
                alreadyTransformed.add(transformation);

                final ChainInfo chainInfo = chainWithInputIfPossible(transformation, outputMap);
                chainedTransformations.add(chainInfo.newTransformation);
                chainedTransformations.removeAll(chainInfo.oldTransformations);
                alreadyTransformed.addAll(chainInfo.oldTransformations);

                // Add the chained transformation and its inputs to the to-optimize list
                toTransformQueue.add(chainInfo.newTransformation);
                toTransformQueue.addAll(chainInfo.newTransformation.getInputs());
            }
        }
        return new ArrayList<>(chainedTransformations);
    }
@Test
    void testTransformationWithMultipleOutputs() {
        ExternalPythonProcessOperator<?, ?> processOperator1 =
                createProcessOperator("f1", Types.STRING(), Types.LONG());
        ExternalPythonProcessOperator<?, ?> processOperator2 =
                createProcessOperator("f2", Types.STRING(), Types.LONG());
        ExternalPythonProcessOperator<?, ?> processOperator3 =
                createProcessOperator("f3", Types.LONG(), Types.INT());

        Transformation<?> sourceTransformation = mock(SourceTransformation.class);
        Transformation<?> processTransformation1 =
                new OneInputTransformation(
                        sourceTransformation,
                        "process",
                        processOperator1,
                        processOperator1.getProducedType(),
                        2);
        Transformation<?> processTransformation2 =
                new OneInputTransformation(
                        processTransformation1,
                        "process",
                        processOperator2,
                        processOperator2.getProducedType(),
                        2);
        Transformation<?> processTransformation3 =
                new OneInputTransformation(
                        processTransformation1,
                        "process",
                        processOperator3,
                        processOperator3.getProducedType(),
                        2);

        List<Transformation<?>> transformations = new ArrayList<>();
        transformations.add(processTransformation2);
        transformations.add(processTransformation3);

        List<Transformation<?>> optimized =
                PythonOperatorChainingOptimizer.optimize(transformations);
        // no chaining optimization occurred
        assertThat(optimized).hasSize(4);
    }