private List<PlanNode> splitInnerTimeJoinNode(
      List<List<TTimePartitionSlot>> continuousTimeRange,
      InnerTimeJoinNode node,
      DistributionPlanContext context,
      List<SeriesSourceNode> seriesScanNodes) {
    List<PlanNode> subInnerJoinNode = new ArrayList<>(continuousTimeRange.size());
    for (List<TTimePartitionSlot> oneRegion : continuousTimeRange) {
      if (!oneRegion.isEmpty()) {
        // TODO InnerTimeJoinNode's header
        InnerTimeJoinNode innerTimeJoinNode = (InnerTimeJoinNode) node.clone();
        innerTimeJoinNode.setPlanNodeId(context.queryContext.getQueryId().genPlanNodeId());

        List<Long> timePartitionIds = convertToTimePartitionIds(oneRegion);
        innerTimeJoinNode.setTimePartitions(timePartitionIds);

        // region group id -> parent InnerTimeJoinNode
        Map<Integer, PlanNode> map = new HashMap<>();
        for (SeriesSourceNode sourceNode : seriesScanNodes) {
          TRegionReplicaSet dataRegion =
              analysis.getPartitionInfo(sourceNode.getPartitionPath(), oneRegion.get(0));
          InnerTimeJoinNode parent =
              (InnerTimeJoinNode)
                  map.computeIfAbsent(
                      dataRegion.regionId.id,
                      k -> {
                        InnerTimeJoinNode value = (InnerTimeJoinNode) node.clone();
                        value.setPlanNodeId(context.queryContext.getQueryId().genPlanNodeId());
                        value.setTimePartitions(timePartitionIds);
                        return value;
                      });
          SeriesSourceNode split = (SeriesSourceNode) sourceNode.clone();
          split.setPlanNodeId(context.queryContext.getQueryId().genPlanNodeId());
          split.setRegionReplicaSet(dataRegion);
          parent.addChild(split);
        }

        for (Map.Entry<Integer, PlanNode> entry : map.entrySet()) {
          InnerTimeJoinNode joinNode = (InnerTimeJoinNode) entry.getValue();
          if (joinNode.getChildren().size() == 1) {
            map.put(entry.getKey(), joinNode.getChildren().get(0));
          }
        }

        if (map.size() > 1) {
          map.values().forEach(innerTimeJoinNode::addChild);
          subInnerJoinNode.add(innerTimeJoinNode);
        } else {
          subInnerJoinNode.add(map.values().iterator().next());
        }
      }
    }

    if (subInnerJoinNode.isEmpty()) {
      for (SeriesSourceNode sourceNode : seriesScanNodes) {
        sourceNode.setRegionReplicaSet(NOT_ASSIGNED);
      }
      return Collections.singletonList(node);
    }
    return subInnerJoinNode;
  }
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