private Operator constructAlignedSeriesAggregationScanOperator(
      PlanNodeId planNodeId,
      AlignedPath alignedPath,
      List<AggregationDescriptor> aggregationDescriptorList,
      Expression pushDownPredicate,
      Ordering scanOrder,
      GroupByTimeParameter groupByTimeParameter,
      boolean outputEndTime,
      LocalExecutionPlanContext context) {
    AlignedFullPath seriesPath = (AlignedFullPath) IFullPath.convertToIFullPath(alignedPath);
    boolean ascending = scanOrder == ASC;
    List<TreeAggregator> aggregators = new ArrayList<>();
    boolean canUseStatistics = true;
    for (AggregationDescriptor descriptor : aggregationDescriptorList) {
      checkArgument(
          descriptor.getInputExpressions().size() == 1,
          "descriptor's input expression size is not 1");

      Expression expression = descriptor.getInputExpressions().get(0);
      if (expression instanceof TimeSeriesOperand) {
        String inputSeries =
            ((TimeSeriesOperand) (descriptor.getInputExpressions().get(0)))
                .getPath()
                .getMeasurement();
        int seriesIndex = alignedPath.getMeasurementList().indexOf(inputSeries);
        TSDataType seriesDataType = alignedPath.getSchemaList().get(seriesIndex).getType();
        if (!judgeCanUseStatistics(descriptor.getAggregationType(), seriesDataType)) {
          canUseStatistics = false;
        }
        aggregators.add(
            new TreeAggregator(
                AccumulatorFactory.createAccumulator(
                    descriptor.getAggregationFuncName(),
                    descriptor.getAggregationType(),
                    Collections.singletonList(seriesDataType),
                    descriptor.getInputExpressions(),
                    descriptor.getInputAttributes(),
                    ascending,
                    descriptor.getStep().isInputRaw()),
                descriptor.getStep(),
                Collections.singletonList(
                    new InputLocation[] {new InputLocation(0, seriesIndex)})));
      } else if (expression instanceof TimestampOperand) {
        aggregators.add(
            new TreeAggregator(
                AccumulatorFactory.createAccumulator(
                    descriptor.getAggregationFuncName(),
                    descriptor.getAggregationType(),
                    Collections.singletonList(TSDataType.INT64),
                    descriptor.getInputExpressions(),
                    descriptor.getInputAttributes(),
                    ascending,
                    descriptor.getStep().isInputRaw()),
                descriptor.getStep(),
                Collections.singletonList(new InputLocation[] {new InputLocation(0, -1)})));
      } else {
        throw new IllegalArgumentException(
            "descriptor's input expression must be TimeSeriesOperand/TimestampOperand, current is "
                + expression);
      }
    }

    ITimeRangeIterator timeRangeIterator =
        initTimeRangeIterator(groupByTimeParameter, ascending, true, context.getZoneId());
    long maxReturnSize =
        AggregationUtil.calculateMaxAggregationResultSize(
            aggregationDescriptorList, timeRangeIterator, context.getTypeProvider());

    SeriesScanOptions.Builder scanOptionsBuilder = getSeriesScanOptionsBuilder(context);
    scanOptionsBuilder.withAllSensors(new HashSet<>(alignedPath.getMeasurementList()));

    if (pushDownPredicate != null) {
      checkArgument(PredicateUtils.predicateCanPushIntoScan(pushDownPredicate));
      scanOptionsBuilder.withPushDownFilter(
          convertPredicateToFilter(
              pushDownPredicate,
              alignedPath.getMeasurementList(),
              context.getTypeProvider().getTemplatedInfo() != null,
              context.getTypeProvider(),
              context.getZoneId()));
    }

    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                planNodeId,
                AlignedSeriesAggregationScanOperator.class.getSimpleName());
    AlignedSeriesAggregationScanOperator seriesAggregationScanOperator =
        new AlignedSeriesAggregationScanOperator(
            planNodeId,
            seriesPath,
            scanOrder,
            outputEndTime,
            scanOptionsBuilder.build(),
            operatorContext,
            aggregators,
            timeRangeIterator,
            groupByTimeParameter,
            maxReturnSize,
            canUseStatistics);

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