public List<Operator> dealWithConsumeAllChildrenPipelineBreaker(
      PlanNode node, LocalExecutionPlanContext context) {
    // children after pipelining
    List<Operator> parentPipelineChildren = new ArrayList<>();
    if (context.getDegreeOfParallelism() == 1 || node.getChildren().size() == 1) {
      // If dop = 1, we don't create extra pipeline
      for (PlanNode localChild : node.getChildren()) {
        Operator childOperation = localChild.accept(this, context);
        parentPipelineChildren.add(childOperation);
        // if we don't create extra pipeline, the root of the child pipeline should be current root
        List<PipelineMemoryEstimator> childPipelineMemoryEstimators =
            context.getParentPlanNodeIdToMemoryEstimator().get(localChild.getPlanNodeId());
        if (childPipelineMemoryEstimators != null) {
          context.getParentPlanNodeIdToMemoryEstimator().remove(localChild.getPlanNodeId());
          context
              .getParentPlanNodeIdToMemoryEstimator()
              .put(node.getPlanNodeId(), childPipelineMemoryEstimators);
        }
      }
    } else {
      int finalExchangeNum = context.getExchangeSumNum();
      // Keep it since we may change the structure of origin children nodes
      List<PlanNode> afterwardsNodes = new ArrayList<>();
      // 1. Calculate localChildren size
      int localChildrenSize = 0;
      int firstChildIndex = -1;
      for (int i = 0; i < node.getChildren().size(); i++) {
        if (!(node.getChildren().get(i) instanceof ExchangeNode)) {
          localChildrenSize++;
          firstChildIndex = firstChildIndex == -1 ? i : firstChildIndex;
          // deal with exchangeNode at head
        } else if (firstChildIndex == -1) {
          Operator childOperation = node.getChildren().get(i).accept(this, context);
          finalExchangeNum += 1;
          parentPipelineChildren.add(childOperation);
          afterwardsNodes.add(node.getChildren().get(i));
        }
      }
      if (firstChildIndex == -1) {
        context.setExchangeSumNum(finalExchangeNum);
        return parentPipelineChildren;
      }
      // If dop > localChildrenSize + 1, we can allocate extra dop to child node
      // Extra dop = dop - localChildrenSize, since dop = 1 means serial but not 0
      int dopForChild = Math.max(1, context.getDegreeOfParallelism() - localChildrenSize);
      // If dop > localChildrenSize, we create one new pipeline for each child
      if (context.getDegreeOfParallelism() > localChildrenSize) {
        for (int i = firstChildIndex; i < node.getChildren().size(); i++) {
          PlanNode childNode = node.getChildren().get(i);
          if (childNode instanceof ExchangeNode) {
            Operator childOperation = childNode.accept(this, context);
            finalExchangeNum += 1;
            parentPipelineChildren.add(childOperation);
          } else {
            LocalExecutionPlanContext subContext = context.createSubContext();
            subContext.setDegreeOfParallelism(dopForChild);

            int originPipeNum = context.getPipelineNumber();
            Operator sourceOperator =
                createNewPipelineForChildNode(context, subContext, childNode, node.getPlanNodeId());
            parentPipelineChildren.add(sourceOperator);
            dopForChild =
                Math.max(1, dopForChild - (subContext.getPipelineNumber() - 1 - originPipeNum));
            finalExchangeNum += subContext.getExchangeSumNum() - context.getExchangeSumNum() + 1;
          }
        }
      } else {
        // If dop <= localChildrenSize, we have to divide every childNumInEachPipeline localChildren
        // to different pipeline
        int[] childNumInEachPipeline =
            getChildNumInEachPipeline(
                node.getChildren(), localChildrenSize, context.getDegreeOfParallelism());
        int childGroupNum = Math.min(context.getDegreeOfParallelism(), localChildrenSize);
        int startIndex;
        int endIndex = firstChildIndex;
        for (int i = 0; i < childGroupNum; i++) {
          startIndex = endIndex;
          endIndex += childNumInEachPipeline[i];
          // Only if dop >= size(children) + 1, split all children to new pipeline
          // Otherwise, the first group will belong to the parent pipeline
          if (i == 0) {
            for (int j = startIndex; j < endIndex; j++) {
              context.setDegreeOfParallelism(1);
              Operator childOperation = node.getChildren().get(j).accept(this, context);
              parentPipelineChildren.add(childOperation);
              afterwardsNodes.add(node.getChildren().get(j));
            }
            continue;
          }
          LocalExecutionPlanContext subContext = context.createSubContext();
          subContext.setDegreeOfParallelism(1);
          // Create partial parent operator for children
          PlanNode partialParentNode;
          if (endIndex - startIndex == 1) {
            partialParentNode = node.getChildren().get(startIndex);
          } else {
            // PartialParentNode is equals to parentNode except children
            partialParentNode = node.createSubNode(i, startIndex, endIndex);
          }

          Operator sourceOperator =
              createNewPipelineForChildNode(
                  context, subContext, partialParentNode, node.getPlanNodeId());
          parentPipelineChildren.add(sourceOperator);
          afterwardsNodes.add(partialParentNode);
          finalExchangeNum += subContext.getExchangeSumNum() - context.getExchangeSumNum() + 1;
        }

        if (node instanceof MultiChildProcessNode) {
          ((MultiChildProcessNode) node).setChildren(afterwardsNodes);
        } else if (node instanceof TwoChildProcessNode) {
          checkState(afterwardsNodes.size() == 2);
          ((TwoChildProcessNode) node).setLeftChild(afterwardsNodes.get(0));
          ((TwoChildProcessNode) node).setRightChild(afterwardsNodes.get(1));
        } else {
          throw new IllegalArgumentException("Unknown node type: " + node.getClass().getName());
        }
      }
      context.setExchangeSumNum(finalExchangeNum);
    }
    return parentPipelineChildren;
  }
This Visitor is responsible for transferring PlanNode Tree to Operator Tree.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/OperatorTreeGenerator.java