private List<PlanNode> planAggregationWithTimeJoin(
      FullOuterTimeJoinNode root, DistributionPlanContext context) {
    Map<TRegionReplicaSet, List<SeriesAggregationSourceNode>> sourceGroup;

    // construct newRoot
    MultiChildProcessNode newRoot;
    if (context.isRoot) {
      // This node is the root of PlanTree,
      // if the aggregated series have only one region,
      // upstream will give the final aggregate result,
      // the step of this series' aggregator will be `STATIC`
      List<SeriesAggregationSourceNode> sources = new ArrayList<>();
      Map<PartialPath, Integer> regionCountPerSeries = new HashMap<>();
      boolean[] eachSeriesOneRegion = {true};
      sourceGroup =
          splitAggregationSourceByPartition(
              root, context, sources, eachSeriesOneRegion, regionCountPerSeries);

      if (eachSeriesOneRegion[0]) {
        newRoot = new HorizontallyConcatNode(context.queryContext.getQueryId().genPlanNodeId());
      } else {
        List<AggregationDescriptor> rootAggDescriptorList = new ArrayList<>();
        for (PlanNode child : root.getChildren()) {
          SeriesAggregationSourceNode handle = (SeriesAggregationSourceNode) child;
          handle
              .getAggregationDescriptorList()
              .forEach(
                  descriptor ->
                      rootAggDescriptorList.add(
                          new AggregationDescriptor(
                              descriptor.getAggregationFuncName(),
                              regionCountPerSeries.get(handle.getPartitionPath()) == 1
                                  ? AggregationStep.STATIC
                                  : AggregationStep.FINAL,
                              descriptor.getInputExpressions(),
                              descriptor.getInputAttributes())));
        }
        SeriesAggregationSourceNode seed = (SeriesAggregationSourceNode) root.getChildren().get(0);
        newRoot =
            new AggregationNode(
                context.queryContext.getQueryId().genPlanNodeId(),
                rootAggDescriptorList,
                seed.getGroupByTimeParameter(),
                seed.getScanOrder());
      }
    } else {
      // If this node is not the root node of PlanTree,
      // it declares that there have something to do at downstream,
      // the step of this AggregationNode should be `INTERMEDIATE`
      sourceGroup = splitAggregationSourceByPartition(root, context);
      List<AggregationDescriptor> rootAggDescriptorList = new ArrayList<>();
      for (PlanNode child : root.getChildren()) {
        SeriesAggregationSourceNode handle = (SeriesAggregationSourceNode) child;
        handle
            .getAggregationDescriptorList()
            .forEach(
                descriptor ->
                    rootAggDescriptorList.add(
                        new AggregationDescriptor(
                            descriptor.getAggregationFuncName(),
                            AggregationStep.INTERMEDIATE,
                            descriptor.getInputExpressions(),
                            descriptor.getInputAttributes())));
      }
      SeriesAggregationSourceNode seed = (SeriesAggregationSourceNode) root.getChildren().get(0);
      newRoot =
          new AggregationNode(
              context.queryContext.getQueryId().genPlanNodeId(),
              rootAggDescriptorList,
              seed.getGroupByTimeParameter(),
              seed.getScanOrder());
    }

    boolean[] addParent = {false};
    sourceGroup.forEach(
        (dataRegion, sourceNodes) -> {
          if (sourceNodes.size() == 1) {
            newRoot.addChild(sourceNodes.get(0));
          } else {
            if (!addParent[0]) {
              sourceNodes.forEach(newRoot::addChild);
              addParent[0] = true;
            } else {
              HorizontallyConcatNode parentOfGroup =
                  new HorizontallyConcatNode(context.queryContext.getQueryId().genPlanNodeId());
              sourceNodes.forEach(parentOfGroup::addChild);
              newRoot.addChild(parentOfGroup);
            }
          }
        });

    return Collections.singletonList(newRoot);
  }
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