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 testChainingUnorderedTransformations() {
        ExternalPythonKeyedProcessOperator<?> keyedProcessOperator =
                createKeyedProcessOperator(
                        "f1", new RowTypeInfo(Types.INT(), Types.INT()), Types.STRING());
        ExternalPythonProcessOperator<?, ?> processOperator1 =
                createProcessOperator("f2", Types.STRING(), Types.LONG());
        ExternalPythonProcessOperator<?, ?> processOperator2 =
                createProcessOperator("f3", Types.LONG(), Types.INT());

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

        List<Transformation<?>> transformations = new ArrayList<>();
        transformations.add(sourceTransformation);
        transformations.add(processTransformation2);
        transformations.add(processTransformation1);
        transformations.add(keyedProcessTransformation);

        List<Transformation<?>> optimized =
                PythonOperatorChainingOptimizer.optimize(transformations);
        assertThat(optimized).hasSize(2);

        OneInputTransformation<?, ?> chainedTransformation =
                (OneInputTransformation<?, ?>) optimized.get(1);
        assertThat(sourceTransformation.getOutputType())
                .isEqualTo(chainedTransformation.getInputType());
        assertThat(processOperator2.getProducedType())
                .isEqualTo(chainedTransformation.getOutputType());

        OneInputStreamOperator<?, ?> chainedOperator = chainedTransformation.getOperator();
        assertThat(chainedOperator).isInstanceOf(ExternalPythonKeyedProcessOperator.class);
        validateChainedPythonFunctions(
                ((ExternalPythonKeyedProcessOperator<?>) chainedOperator).getPythonFunctionInfo(),
                "f3",
                "f2",
                "f1");
    }