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 testCalculateInputOrderWithUnaffectedRelatedBoundaries() {
        // P = InputProperty.DamBehavior.PIPELINED, B = InputProperty.DamBehavior.BLOCKING
        // P1 = PIPELINED + priority 1
        //
        // 0 --(P0)-> 1 -------(B0)-----> 2 -(P0)-\
        //  \          \--(B0)-> 3 -(P1)-/         4
        //   \-(B0)-> 5 -------(P1)-----> 6 -(P0)-/
        //                     7 --(B0)--/
        TestingBatchExecNode[] nodes = new TestingBatchExecNode[8];
        for (int i = 0; i < nodes.length; i++) {
            nodes[i] = new TestingBatchExecNode("TestingBatchExecNode" + i);
        }
        nodes[1].addInput(nodes[0]);
        nodes[2].addInput(
                nodes[1],
                InputProperty.builder().damBehavior(InputProperty.DamBehavior.BLOCKING).build());
        nodes[2].addInput(nodes[3], InputProperty.builder().priority(1).build());
        nodes[3].addInput(
                nodes[1],
                InputProperty.builder().damBehavior(InputProperty.DamBehavior.BLOCKING).build());
        nodes[4].addInput(nodes[2]);
        nodes[4].addInput(nodes[6]);
        nodes[5].addInput(
                nodes[0],
                InputProperty.builder().damBehavior(InputProperty.DamBehavior.BLOCKING).build());
        nodes[6].addInput(nodes[5], InputProperty.builder().priority(1).build());
        nodes[6].addInput(
                nodes[7],
                InputProperty.builder().damBehavior(InputProperty.DamBehavior.BLOCKING).build());

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