@Test
  public void testConsumeAllChildrenPipelineBuilder5() throws IllegalPathException {
    TypeProvider typeProvider = new TypeProvider();
    FullOuterTimeJoinNode fullOuterTimeJoinNode = initFullOuterTimeJoinNode(typeProvider, 4);
    LocalExecutionPlanContext context = createLocalExecutionPlanContext(typeProvider);
    context.setDegreeOfParallelism(5);

    List<Operator> childrenOperator =
        operatorTreeGenerator.dealWithConsumeAllChildrenPipelineBreaker(
            fullOuterTimeJoinNode, context);
    // The number of pipeline is 4, since parent pipeline hasn't joined
    assertEquals(4, context.getPipelineNumber());

    // Validate the first pipeline
    assertEquals(4, childrenOperator.size());
    for (int i = 0; i < 4; i++) {
      assertEquals(ExchangeOperator.class, childrenOperator.get(i).getClass());
    }

    // Validate the changes of node structure
    assertEquals(4, fullOuterTimeJoinNode.getChildren().size());
    for (int i = 0; i < 4; i++) {
      assertEquals(SeriesScanNode.class, fullOuterTimeJoinNode.getChildren().get(i).getClass());
      assertEquals(
          String.format("root.sg.d%d.s1", i),
          fullOuterTimeJoinNode.getChildren().get(i).getOutputColumnNames().get(0));
    }

    // Validate the second pipeline
    ExchangeOperator exchangeOperator1 = (ExchangeOperator) childrenOperator.get(0);
    assertEquals("SeriesScanNode0", exchangeOperator1.getSourceId().getId());

    // Validate the third pipeline
    ExchangeOperator exchangeOperator2 = (ExchangeOperator) childrenOperator.get(1);
    assertEquals("SeriesScanNode1", exchangeOperator2.getSourceId().getId());

    // Validate the forth pipeline
    ExchangeOperator exchangeOperator3 = (ExchangeOperator) childrenOperator.get(2);
    assertEquals("SeriesScanNode2", exchangeOperator3.getSourceId().getId());

    // Validate the fifth pipeline
    ExchangeOperator exchangeOperator4 = (ExchangeOperator) childrenOperator.get(3);
    assertEquals("SeriesScanNode3", exchangeOperator4.getSourceId().getId());

    // Validate the number exchange operator
    assertEquals(4, context.getExchangeSumNum());

    // Validate PipelineMemoryEstimator
    try (Operator root =
        fullOuterTimeJoinNode.accept(
            operatorTreeGenerator, createLocalExecutionPlanContext(typeProvider))) {
      PipelineMemoryEstimator rootPipelineMemoryEstimator =
          context.constructPipelineMemoryEstimator(root, null, fullOuterTimeJoinNode, -1);
      assertEquals(
          ConsumeAllChildrenPipelineMemoryEstimator.class, rootPipelineMemoryEstimator.getClass());
      List<PipelineMemoryEstimator> childrenPipelineMemoryEstimators =
          rootPipelineMemoryEstimator.getChildren();
      assertEquals(4, childrenPipelineMemoryEstimators.size());
      for (int i = 0; i < 4; i++) {
        assertEquals(
            ConsumeAllChildrenPipelineMemoryEstimator.class,
            childrenPipelineMemoryEstimators.get(i).getClass());
        assertEquals(
            SeriesScanOperator.class, childrenPipelineMemoryEstimators.get(i).getRoot().getClass());
      }
    } catch (Exception e) {
      e.printStackTrace();
      fail();
    }
  }
This test will test dop = 5. Expected result is five pipelines:

<p>The first is: TimeJoin1 - [ExchangeOperator, ExchangeOperator, ExchangeOperator,
ExchangeOperator];

<p>The second is: ExchangeOperator - SeriesScan0.

<p>The third is: ExchangeOperator - SeriesScan1.

<p>The forth is: ExchangeOperator - SeriesScan2.

<p>The fifth is: ExchangeOperator - SeriesScan3.
iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/PipelineBuilderTest.java