@Override
  public Operator visitGroupByTag(GroupByTagNode node, LocalExecutionPlanContext context) {
    checkArgument(!node.getTagKeys().isEmpty(), "GroupByTag tag keys cannot be empty");
    checkArgument(
        node.getTagValuesToAggregationDescriptors().size() >= 1,
        "GroupByTag aggregation descriptors cannot be empty");

    List<Operator> children = dealWithConsumeAllChildrenPipelineBreaker(node, context);

    boolean ascending = node.getScanOrder() == ASC;
    Map<String, List<InputLocation>> layout = makeLayout(node);
    List<List<String>> groups = new ArrayList<>();
    List<List<TreeAggregator>> groupedAggregators = new ArrayList<>();
    int aggregatorCount = 0;
    for (Map.Entry<List<String>, List<CrossSeriesAggregationDescriptor>> entry :
        node.getTagValuesToAggregationDescriptors().entrySet()) {
      groups.add(entry.getKey());
      List<TreeAggregator> aggregators = new ArrayList<>();
      for (CrossSeriesAggregationDescriptor aggregationDescriptor : entry.getValue()) {
        if (aggregationDescriptor == null) {
          aggregators.add(null);
          continue;
        }
        List<InputLocation[]> inputLocations = calcInputLocationList(aggregationDescriptor, layout);
        List<TSDataType> inputDataTypes =
            aggregationDescriptor.getInputExpressions().stream()
                .map(x -> context.getTypeProvider().getTreeModelType(x.getExpressionString()))
                .collect(Collectors.toList());
        aggregators.add(
            new TreeAggregator(
                AccumulatorFactory.createAccumulator(
                    aggregationDescriptor.getAggregationFuncName(),
                    aggregationDescriptor.getAggregationType(),
                    inputDataTypes,
                    aggregationDescriptor.getInputExpressions(),
                    aggregationDescriptor.getInputAttributes(),
                    ascending,
                    aggregationDescriptor.getStep().isInputRaw()),
                aggregationDescriptor.getStep(),
                inputLocations));
      }
      groupedAggregators.add(aggregators);
      aggregatorCount += aggregators.size();
    }
    GroupByTimeParameter groupByTimeParameter = node.getGroupByTimeParameter();
    ITimeRangeIterator timeRangeIterator =
        initTimeRangeIterator(groupByTimeParameter, ascending, false, context.getZoneId());
    List<AggregationDescriptor> aggregationDescriptors =
        node.getTagValuesToAggregationDescriptors().values().stream()
            .flatMap(Collection::stream)
            .filter(Objects::nonNull)
            .collect(Collectors.toList());

    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                TagAggregationOperator.class.getSimpleName());
    long maxReturnSize =
        calculateMaxAggregationResultSize(
            aggregationDescriptors, timeRangeIterator, context.getTypeProvider());

    return new TagAggregationOperator(
        operatorContext, groups, groupedAggregators, children, maxReturnSize);
  }
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