@Override
  public List<PlanNode> visitInnerTimeJoin(
      InnerTimeJoinNode node, DistributionPlanContext context) {
    // All child nodes should be SourceNode
    List<SeriesSourceNode> seriesScanNodes = new ArrayList<>(node.getChildren().size());
    List<List<List<TTimePartitionSlot>>> sourceTimeRangeList = new ArrayList<>();
    for (PlanNode child : node.getChildren()) {
      if (!(child instanceof SeriesSourceNode)) {
        throw new IllegalStateException(
            "All child nodes of InnerTimeJoinNode should be SeriesSourceNode");
      }
      SeriesSourceNode sourceNode = (SeriesSourceNode) child;
      seriesScanNodes.add(sourceNode);
      sourceTimeRangeList.add(
          analysis.getTimePartitionRange(
              sourceNode.getPartitionPath(), context.getPartitionTimeFilter()));
    }

    List<List<TTimePartitionSlot>> res = splitTimePartition(sourceTimeRangeList);

    List<PlanNode> children = splitInnerTimeJoinNode(res, node, context, seriesScanNodes);

    if (children.size() == 1) {
      return children;
    } else {
      // add merge sort node for InnerTimeJoinNodes
      // TODO add new type of Node, just traverse all child nodes sequentially in the future
      List<String> outputColumnNames = node.getOutputColumnNames();
      MergeSortNode mergeSortNode =
          new MergeSortNode(
              context.queryContext.getQueryId().genPlanNodeId(),
              node.getMergeOrder() == Ordering.ASC ? TIME_ASC : TIME_DESC,
              outputColumnNames);
      // set unified outputColumnNames for each child InnerTimeJoinNode
      children.forEach(
          child -> ((InnerTimeJoinNode) child).setOutputColumnNames(outputColumnNames));

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