public Map<ExecNode<?>, Integer> calculate() {
        createTopologyGraph();

        // some boundaries node may be connected from the outside of the sub-graph,
        // which we cannot deduce by the above process,
        // so we need to check each pair of boundaries and see if they're related
        dealWithPossiblyRelatedBoundaries();
        Map<ExecNode<?>, Integer> distances = graph.calculateMaximumDistance();

        // extract only the distances of the boundaries and renumbering the distances
        // so that the smallest value starts from 0
        // the smaller the distance, the higher the priority
        Set<Integer> boundaryDistanceSet = new HashSet<>();
        for (ExecNode<?> boundary : boundaries) {
            boundaryDistanceSet.add(distances.getOrDefault(boundary, 0));
        }
        List<Integer> boundaryDistanceList = new ArrayList<>(boundaryDistanceSet);
        Collections.sort(boundaryDistanceList);

        Map<ExecNode<?>, Integer> results = new HashMap<>();
        for (ExecNode<?> boundary : boundaries) {
            results.put(boundary, boundaryDistanceList.indexOf(distances.get(boundary)));
        }
        return results;
    }
@Test
    void testCalculateInputOrder() {
        // P = InputProperty.DamBehavior.PIPELINED, B = InputProperty.DamBehavior.BLOCKING
        // P1 = PIPELINED + priority 1
        //
        // 0 -(P1)-> 3 -(B0)-\
        //                    6 -(B0)-\
        //            /-(P1)-/         \
        // 1 -(P1)-> 4                  8
        //            \-(B0)-\         /
        //                    7 -(P1)-/
        // 2 -(P1)-> 5 -(P1)-/
        TestingBatchExecNode[] nodes = new TestingBatchExecNode[9];
        for (int i = 0; i < nodes.length; i++) {
            nodes[i] = new TestingBatchExecNode("TestingBatchExecNode" + i);
        }
        nodes[3].addInput(nodes[0], InputProperty.builder().priority(1).build());
        nodes[4].addInput(nodes[1], InputProperty.builder().priority(1).build());
        nodes[5].addInput(nodes[2], InputProperty.builder().priority(1).build());
        nodes[6].addInput(
                nodes[3],
                InputProperty.builder()
                        .damBehavior(InputProperty.DamBehavior.BLOCKING)
                        .priority(0)
                        .build());
        nodes[6].addInput(nodes[4], InputProperty.builder().priority(1).build());
        nodes[7].addInput(
                nodes[4],
                InputProperty.builder()
                        .damBehavior(InputProperty.DamBehavior.BLOCKING)
                        .priority(0)
                        .build());
        nodes[7].addInput(nodes[5], InputProperty.builder().priority(1).build());
        nodes[8].addInput(
                nodes[6],
                InputProperty.builder()
                        .damBehavior(InputProperty.DamBehavior.BLOCKING)
                        .priority(0)
                        .build());
        nodes[8].addInput(nodes[7], InputProperty.builder().priority(1).build());

        InputOrderCalculator calculator =
                new InputOrderCalculator(
                        nodes[8],
                        new HashSet<>(Arrays.asList(nodes[1], nodes[3], nodes[5])),
                        InputProperty.DamBehavior.BLOCKING);
        Map<ExecNode<?>, Integer> result = calculator.calculate();
        assertThat(result).hasSize(3);
        assertThat(result.get(nodes[3]).intValue()).isEqualTo(0);
        assertThat(result.get(nodes[1]).intValue()).isEqualTo(1);
        assertThat(result.get(nodes[5]).intValue()).isEqualTo(2);
    }