private Map<TRegionReplicaSet, List<SourceNode>> groupBySourceNodes(
      MultiChildProcessNode node, DistributionPlanContext context) {
    // Step 1: Get all source nodes. For the node which is not source, add it as the child of
    // current TimeJoinNode
    List<SourceNode> sources = new ArrayList<>();
    for (PlanNode child : node.getChildren()) {
      if (child instanceof SeriesSourceNode) {
        // If the child is SeriesScanNode, we need to check whether this node should be seperated
        // into several splits.
        SeriesSourceNode sourceNode = (SeriesSourceNode) child;
        List<TRegionReplicaSet> dataDistribution =
            analysis.getPartitionInfo(
                sourceNode.getPartitionPath(), context.getPartitionTimeFilter());
        if (dataDistribution.size() > 1) {
          // If there is some series which is distributed in multi DataRegions
          context.setOneSeriesInMultiRegion(true);
        }
        // If the size of dataDistribution is N, this SeriesScanNode should be seperated into N
        // SeriesScanNode.
        for (TRegionReplicaSet dataRegion : dataDistribution) {
          SeriesSourceNode split = (SeriesSourceNode) sourceNode.clone();
          split.setPlanNodeId(context.queryContext.getQueryId().genPlanNodeId());
          split.setRegionReplicaSet(dataRegion);
          sources.add(split);
        }
      }
    }

    // Step 2: For the source nodes, group them by the DataRegion.
    Map<TRegionReplicaSet, List<SourceNode>> sourceGroup =
        sources.stream().collect(Collectors.groupingBy(SourceNode::getRegionReplicaSet));
    if (sourceGroup.size() > 1) {
      context.setQueryMultiRegion(true);
    }

    return sourceGroup;
  }
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