flink-ml-core/src/main/java/org/apache/flink/ml/builder/Graph.java [92:114]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
        GraphNode node;
        while ((node = executionHelper.pollNextReadyNode()) != null) {
            Stage<?> stage = node.stage;
            // Invokes fit(...) if stageType == ESTIMATOR.
            if (node.stageType == StageType.ESTIMATOR) {
                stage =
                        ((Estimator<?, ?>) stage)
                                .fit(executionHelper.getTables(node.estimatorInputIds));
            }
            // Invokes setModelData(...).
            if (node.inputModelDataIds != null) {
                Table[] nodeInputModelData = executionHelper.getTables(node.inputModelDataIds);
                ((Model<?>) stage).setModelData(nodeInputModelData);
            }
            // Invokes transform(...).
            Table[] nodeOutputs =
                    ((AlgoOperator<?>) stage)
                            .transform(executionHelper.getTables(node.algoOpInputIds));
            executionHelper.setTables(node.outputIds, nodeOutputs);
            // Invokes getModelData().
            if (node.outputModelDataIds != null) {
                Table[] nodeOutputModelData = ((Model<?>) stage).getModelData();
                executionHelper.setTables(node.outputModelDataIds, nodeOutputModelData);
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -



flink-ml-core/src/main/java/org/apache/flink/ml/builder/GraphModel.java [85:107]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
        GraphNode node;
        while ((node = executionHelper.pollNextReadyNode()) != null) {
            Stage<?> stage = node.stage;
            // Invokes fit(...) if stageType == ESTIMATOR.
            if (node.stageType == StageType.ESTIMATOR) {
                stage =
                        ((Estimator<?, ?>) stage)
                                .fit(executionHelper.getTables(node.estimatorInputIds));
            }
            // Invokes setModelData(...).
            if (node.inputModelDataIds != null) {
                Table[] nodeInputModelData = executionHelper.getTables(node.inputModelDataIds);
                ((Model<?>) stage).setModelData(nodeInputModelData);
            }
            // Invokes transform(...).
            Table[] nodeOutputs =
                    ((AlgoOperator<?>) stage)
                            .transform(executionHelper.getTables(node.algoOpInputIds));
            executionHelper.setTables(node.outputIds, nodeOutputs);
            // Invokes getModelData().
            if (node.outputModelDataIds != null) {
                Table[] nodeOutputModelData = ((Model<?>) stage).getModelData();
                executionHelper.setTables(node.outputModelDataIds, nodeOutputModelData);
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -