@Override
  public List<PlanNode> visitAggregation(AggregationNode node, PlanContext context) {
    List<Symbol> preGroupedSymbols = node.getPreGroupedSymbols();
    OrderingScheme expectedOrderingSchema;
    if (node.isStreamable()) {
      expectedOrderingSchema = constructOrderingSchema(preGroupedSymbols);
      context.setExpectedOrderingScheme(expectedOrderingSchema);
    } else {
      expectedOrderingSchema = null;
      context.clearExpectedOrderingScheme();
    }

    List<PlanNode> childrenNodes = node.getChild().accept(this, context);
    OrderingScheme childOrdering = nodeOrderingMap.get(childrenNodes.get(0).getPlanNodeId());
    if (node.isStreamable()) {
      // Child has Ordering, we need to check if it is the Ordering we expected
      if (childOrdering != null) {
        if (prefixMatched(childOrdering, node.getPreGroupedSymbols())) {
          nodeOrderingMap.put(node.getPlanNodeId(), expectedOrderingSchema);
        } else {
          throw new IllegalStateException(
              String.format(
                  "Should never reach here. Child ordering: %s. PreGroupedSymbols: %s",
                  childOrdering.getOrderBy(), node.getPreGroupedSymbols()));
        }
      }
      // Child has no Ordering, do nothing here because the logical optimizer
      // 'TransformAggregationToStreamable' will ensure the grouped property of child
    }

    if (childrenNodes.size() == 1) {
      node.setChild(childrenNodes.get(0));
      return Collections.singletonList(node);
    }

    // We cannot do multi-stage Aggregate if any aggregation-function is distinct.
    // For Aggregation with mask, there is no need to do multi-stage Aggregate because the
    // MarkDistinctNode will merge all data from different child.
    if (node.getAggregations().values().stream()
        .anyMatch(aggregation -> aggregation.isDistinct() || aggregation.hasMask())) {
      node.setChild(
          mergeChildrenViaCollectOrMergeSort(
              nodeOrderingMap.get(childrenNodes.get(0).getPlanNodeId()), childrenNodes));
      return Collections.singletonList(node);
    }

    Pair<AggregationNode, AggregationNode> splitResult = split(node, symbolAllocator, queryId);
    AggregationNode intermediate = splitResult.right;

    childrenNodes =
        childrenNodes.stream()
            .map(
                child -> {
                  PlanNodeId planNodeId = queryId.genPlanNodeId();
                  AggregationNode aggregationNode =
                      new AggregationNode(
                          planNodeId,
                          child,
                          intermediate.getAggregations(),
                          intermediate.getGroupingSets(),
                          intermediate.getPreGroupedSymbols(),
                          intermediate.getStep(),
                          intermediate.getHashSymbol(),
                          intermediate.getGroupIdSymbol());
                  if (node.isStreamable() && childOrdering != null) {
                    nodeOrderingMap.put(planNodeId, expectedOrderingSchema);
                  }
                  return aggregationNode;
                })
            .collect(Collectors.toList());
    splitResult.left.setChild(
        mergeChildrenViaCollectOrMergeSort(
            nodeOrderingMap.get(childrenNodes.get(0).getPlanNodeId()), childrenNodes));
    return Collections.singletonList(splitResult.left);
  }
This class is used to generate distributed plan for table model.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java