public List<Operator> dealWithConsumeChildrenOneByOneNode(
      PlanNode node, LocalExecutionPlanContext context) {
    checkArgument(PipelineMemoryEstimatorFactory.isConsumeChildrenOneByOneNode(node));
    List<Operator> parentPipelineChildren = new ArrayList<>();
    int originExchangeNum = context.getExchangeSumNum();
    int finalExchangeNum = context.getExchangeSumNum();

    // 1. divide every child to pipeline using the max dop
    if (context.getDegreeOfParallelism() == 1 || node.getChildren().size() == 1) {
      // If dop = 1, we don't create extra pipeline
      for (PlanNode childSource : node.getChildren()) {
        Operator childOperation = childSource.accept(this, context);
        finalExchangeNum = Math.max(finalExchangeNum, context.getExchangeSumNum());
        context.setExchangeSumNum(originExchangeNum);
        parentPipelineChildren.add(childOperation);
        // if node.getChildren().size() == 1 and we don't create extra pipeline, the root of the
        // child pipeline should be current root
        // for example, we have IdentitySinkNode -> DeviceViewNode -> [ScanNode, ScanNode, ScanNode]
        // the parent of the pipeline of ScanNode should be IdentitySinkNode in the map, otherwise
        // we will lose the information of these pipelines
        List<PipelineMemoryEstimator> childPipelineMemoryEstimators =
            context.getParentPlanNodeIdToMemoryEstimator().get(childSource.getPlanNodeId());
        if (childPipelineMemoryEstimators != null) {
          context.getParentPlanNodeIdToMemoryEstimator().remove(childSource.getPlanNodeId());
          context
              .getParentPlanNodeIdToMemoryEstimator()
              .put(node.getPlanNodeId(), childPipelineMemoryEstimators);
        }
      }
    } else {
      List<Integer> childPipelineNums = new ArrayList<>();
      List<Integer> childExchangeNums = new ArrayList<>();
      int sumOfChildPipelines = 0;
      int sumOfChildExchangeNums = 0;
      int dependencyChildNode = 0;
      int dependencyPipeId = 0;
      for (PlanNode childNode : node.getChildren()) {
        if (childNode instanceof ExchangeNode) {
          Operator childOperation = childNode.accept(this, context);
          finalExchangeNum = Math.max(finalExchangeNum, context.getExchangeSumNum());
          context.setExchangeSumNum(originExchangeNum);
          parentPipelineChildren.add(childOperation);
        } else {
          LocalExecutionPlanContext subContext = context.createSubContext();
          // Only context.getDegreeOfParallelism() - 1 can be allocated to child
          int dopForChild = context.getDegreeOfParallelism() - 1;
          subContext.setDegreeOfParallelism(dopForChild);
          int originPipeNum = context.getPipelineNumber();
          Operator childOperation = childNode.accept(this, subContext);
          ISinkChannel localSinkChannel =
              MPP_DATA_EXCHANGE_MANAGER.createLocalSinkChannelForPipeline(
                  // Attention, there is no parent node, use first child node instead
                  context.getDriverContext(), childNode.getPlanNodeId().getId());
          subContext.setISink(localSinkChannel);
          subContext.addPipelineDriverFactory(childOperation, subContext.getDriverContext(), 0);

          // OneByOneChild may be divided into more than dop pipelines, but the number of running
          // actually is dop
          int curChildPipelineNum =
              Math.min(dopForChild, subContext.getPipelineNumber() - originPipeNum);

          childPipelineNums.add(curChildPipelineNum);
          sumOfChildPipelines += curChildPipelineNum;
          // If sumOfChildPipelines > dopForChild, we have to wait until some pipelines finish
          if (sumOfChildPipelines > dopForChild) {
            // Update dependencyPipeId, after which finishes we can submit curChildPipeline
            while (sumOfChildPipelines > dopForChild) {
              sumOfChildPipelines -= childPipelineNums.get(dependencyChildNode);
              // The dependency pipeline must be a parent pipeline rather than a child pipeline
              dependencyPipeId = context.getPipelineNumber() - sumOfChildPipelines - 1;
              sumOfChildExchangeNums -= childExchangeNums.get(dependencyChildNode);
              dependencyChildNode++;
            }
          }
          // Add dependency for all pipelines under current node
          if (dependencyChildNode != 0) {
            for (int i = originPipeNum; i < subContext.getPipelineNumber(); i++) {
              context.getPipelineDriverFactories().get(i).setDependencyPipeline(dependencyPipeId);
            }
          }

          subContext.constructPipelineMemoryEstimator(
              childOperation,
              node.getPlanNodeId(),
              childNode,
              dependencyChildNode == 0 ? -1 : dependencyPipeId);

          ExchangeOperator sourceOperator =
              new ExchangeOperator(
                  context
                      .getDriverContext()
                      .addOperatorContext(
                          context.getNextOperatorId(),
                          null,
                          ExchangeOperator.class.getSimpleName()),
                  MPP_DATA_EXCHANGE_MANAGER.createLocalSourceHandleForPipeline(
                      ((LocalSinkChannel) localSinkChannel).getSharedTsBlockQueue(),
                      context.getDriverContext()),
                  childNode.getPlanNodeId(),
                  childOperation.calculateMaxReturnSize());
          context.getCurrentPipelineDriverFactory().setDownstreamOperator(sourceOperator);
          parentPipelineChildren.add(sourceOperator);
          context.addExchangeOperator(sourceOperator);
          int childExchangeNum = subContext.getExchangeSumNum() - context.getExchangeSumNum() + 1;
          sumOfChildExchangeNums += childExchangeNum;
          childExchangeNums.add(childExchangeNum);
          finalExchangeNum =
              Math.max(finalExchangeNum, context.getExchangeSumNum() + sumOfChildExchangeNums);
        }
      }
    }
    context.setExchangeSumNum(finalExchangeNum);
    return parentPipelineChildren;
  }
Now, we allocate children to each pipeline as average as possible. For example, 5 children with
3 dop, the children group will be [1, 2, 2]. After we can estimate the workload of each
operator, maybe we can allocate based on workload rather than child number.

<p>If child is ExchangeNode, it won't affect the children number of current group.

<p>This method can only be invoked when dop <= localChildrenSize.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/OperatorTreeGenerator.java