@Override
  public Operator visitSeriesAggregationScan(
      SeriesAggregationScanNode node, LocalExecutionPlanContext context) {
    NonAlignedFullPath seriesPath =
        (NonAlignedFullPath) IFullPath.convertToIFullPath(node.getSeriesPath());
    boolean ascending = node.getScanOrder() == ASC;
    List<AggregationDescriptor> aggregationDescriptors = node.getAggregationDescriptorList();
    List<TreeAggregator> aggregators = new ArrayList<>();
    aggregationDescriptors.forEach(
        o ->
            aggregators.add(
                new TreeAggregator(
                    AccumulatorFactory.createAccumulator(
                        o.getAggregationFuncName(),
                        o.getAggregationType(),
                        Collections.singletonList(node.getSeriesPath().getSeriesType()),
                        o.getInputExpressions(),
                        o.getInputAttributes(),
                        ascending,
                        o.getStep().isInputRaw()),
                    o.getStep())));

    GroupByTimeParameter groupByTimeParameter = node.getGroupByTimeParameter();
    ITimeRangeIterator timeRangeIterator =
        initTimeRangeIterator(groupByTimeParameter, ascending, true, context.getZoneId());
    long maxReturnSize =
        AggregationUtil.calculateMaxAggregationResultSize(
            node.getAggregationDescriptorList(), timeRangeIterator, context.getTypeProvider());

    SeriesScanOptions.Builder scanOptionsBuilder = getSeriesScanOptionsBuilder(context);
    scanOptionsBuilder.withAllSensors(
        context.getAllSensors(seriesPath.getDeviceId(), seriesPath.getMeasurement()));

    Expression pushDownPredicate = node.getPushDownPredicate();
    if (pushDownPredicate != null) {
      checkArgument(PredicateUtils.predicateCanPushIntoScan(pushDownPredicate));
      scanOptionsBuilder.withPushDownFilter(
          convertPredicateToFilter(
              pushDownPredicate,
              Collections.singletonList(node.getSeriesPath().getMeasurement()),
              context.getTypeProvider().getTemplatedInfo() != null,
              context.getTypeProvider(),
              context.getZoneId()));
    }

    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                SeriesAggregationScanOperator.class.getSimpleName());
    boolean canUseStatistics =
        !TSDataType.BLOB.equals(node.getSeriesPath().getSeriesType())
            || (aggregationDescriptors.stream()
                .allMatch(o -> judgeCanUseStatistics(o.getAggregationType(), TSDataType.BLOB)));
    SeriesAggregationScanOperator aggregateScanOperator =
        new SeriesAggregationScanOperator(
            node.getPlanNodeId(),
            seriesPath,
            node.getScanOrder(),
            node.isOutputEndTime(),
            scanOptionsBuilder.build(),
            operatorContext,
            aggregators,
            timeRangeIterator,
            node.getGroupByTimeParameter(),
            maxReturnSize,
            canUseStatistics);

    ((DataDriverContext) context.getDriverContext()).addSourceOperator(aggregateScanOperator);
    ((DataDriverContext) context.getDriverContext()).addPath(seriesPath);
    context.getDriverContext().setInputDriver(true);
    return aggregateScanOperator;
  }
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