@Test
  public void testConsumeOneByOneChildrenPipelineBuilderDependency() throws IllegalPathException {
    TypeProvider typeProvider = new TypeProvider();
    typeProvider.setTreeModelType("root.sg.d0.s1", TSDataType.INT64);
    typeProvider.setTreeModelType("root.sg.d1.s1", TSDataType.INT64);
    typeProvider.setTreeModelType("count(root.sg.d0.s1)", TSDataType.INT64);
    typeProvider.setTreeModelType("count(root.sg.d1.s1)", TSDataType.INT64);
    DeviceViewNode deviceViewNode =
        new DeviceViewNode(new PlanNodeId("DeviceViewNode"), null, null, null);
    for (int i = 0; i < 2; i++) {
      PartialPath path = new MeasurementPath(String.format("root.sg.d%d.s1", i), TSDataType.INT64);
      AggregationNode aggregationNode =
          new AggregationNode(
              new PlanNodeId(String.format("AggregationOperator%d", i)),
              Collections.singletonList(
                  new AggregationDescriptor(
                      TAggregationType.COUNT.name().toLowerCase(),
                      AggregationStep.FINAL,
                      Collections.singletonList(new TimeSeriesOperand(path)))),
              null,
              Ordering.ASC);
      SeriesAggregationScanNode seriesAggregationScanNode =
          new SeriesAggregationScanNode(
              new PlanNodeId(String.format("seriesAggregationScanNode%d", i)),
              (MeasurementPath) path,
              Collections.singletonList(
                  new AggregationDescriptor(
                      TAggregationType.COUNT.name().toLowerCase(),
                      AggregationStep.PARTIAL,
                      Collections.singletonList(new TimeSeriesOperand(path)))));
      ExchangeNode exchangeNode =
          new ExchangeNode(new PlanNodeId(String.format("ExchangeNode%d", i)));
      exchangeNode.setUpstream(
          new TEndPoint("127.0.0.1", 6667),
          new FragmentInstanceId(new PlanFragmentId("q", 1), "ds"),
          new PlanNodeId("test"));
      aggregationNode.addChild(seriesAggregationScanNode);
      aggregationNode.addChild(exchangeNode);
      deviceViewNode.addChild(aggregationNode);
    }
    LocalExecutionPlanContext context = createLocalExecutionPlanContext(typeProvider);
    context.setDegreeOfParallelism(3);

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

    assertEquals(2, childrenOperator.size());
    for (int i = 0; i < 2; i++) {
      assertEquals(ExchangeOperator.class, childrenOperator.get(i).getClass());
    }

    // Validate the first pipeline
    assertEquals(-1, context.getPipelineDriverFactories().get(0).getDependencyPipelineIndex());

    // Validate the second pipeline
    assertEquals(-1, context.getPipelineDriverFactories().get(1).getDependencyPipelineIndex());

    // Validate the third pipeline
    assertEquals(1, context.getPipelineDriverFactories().get(2).getDependencyPipelineIndex());

    // Validate the forth pipeline
    assertEquals(1, context.getPipelineDriverFactories().get(2).getDependencyPipelineIndex());
  }
The operator structure is:

<p>DeviceViewOperator - [AggregationOperator1 - [SeriesAggregationScanOperator1,
ExchangeOperator1]], [AggregationOperator2 - [SeriesAggregationScanOperator2,
ExchangeOperator2]].

<p>This test will test dop = 3. Expected result is five pipelines with dependency:

<p>The pipeline0 is: ExchangeOperator - SeriesAggregationScanOperator1.

<p>The pipeline1 is: ExchangeOperator - AggregationOperator1.

<p>The pipeline2 is: ExchangeOperator - SeriesAggregationScanOperator2, which has dependency 1.

<p>The pipeline3 is: ExchangeOperator - AggregationOperator2, which has dependency1.

<p>The pipeline4 is: DeviceView - [ExchangeOperator, ExchangeOperator]
iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/PipelineBuilderTest.java