private List<PlanNode> processSeriesAggregationSource(
      SeriesAggregationSourceNode node, DistributionPlanContext context) {
    List<TRegionReplicaSet> dataDistribution =
        analysis.getPartitionInfo(node.getPartitionPath(), context.getPartitionTimeFilter());
    if (dataDistribution.size() == 1) {
      node.setRegionReplicaSet(dataDistribution.get(0));
      return Collections.singletonList(node);
    }
    List<AggregationDescriptor> leafAggDescriptorList = new ArrayList<>();
    node.getAggregationDescriptorList()
        .forEach(
            descriptor ->
                leafAggDescriptorList.add(
                    new AggregationDescriptor(
                        descriptor.getAggregationFuncName(),
                        AggregationStep.PARTIAL,
                        descriptor.getInputExpressions(),
                        descriptor.getInputAttributes())));
    leafAggDescriptorList.forEach(
        d -> updateTypeProviderByPartialAggregation(d, context.queryContext.getTypeProvider()));
    List<AggregationDescriptor> rootAggDescriptorList = new ArrayList<>();
    node.getAggregationDescriptorList()
        .forEach(
            descriptor ->
                rootAggDescriptorList.add(
                    new AggregationDescriptor(
                        descriptor.getAggregationFuncName(),
                        context.isRoot ? AggregationStep.FINAL : AggregationStep.INTERMEDIATE,
                        descriptor.getInputExpressions(),
                        descriptor.getInputAttributes())));

    AggregationNode aggregationNode =
        new AggregationNode(
            context.queryContext.getQueryId().genPlanNodeId(),
            rootAggDescriptorList,
            node.getGroupByTimeParameter(),
            node.getScanOrder());
    for (TRegionReplicaSet dataRegion : dataDistribution) {
      SeriesAggregationSourceNode split = (SeriesAggregationSourceNode) node.clone();
      split.setAggregationDescriptorList(leafAggDescriptorList);
      split.setPlanNodeId(context.queryContext.getQueryId().genPlanNodeId());
      split.setRegionReplicaSet(dataRegion);
      context
          .getQueryContext()
          .reserveMemoryForFrontEnd(
              MemoryEstimationHelper.getEstimatedSizeOfAccountableObject(split));
      aggregationNode.addChild(split);
    }
    return Collections.singletonList(aggregationNode);
  }
aggregation align by device, and aggregation is `count_if` or `diff`, or aggregation used with
group by parameter (session, variation, count), use the old aggregation logic
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/SourceRewriter.java