@Override
  public List<PlanNode> visitGroupByLevel(GroupByLevelNode root, DistributionPlanContext context) {
    if (shouldUseNaiveAggregation(root)) {
      return defaultRewrite(root, context);
    }
    // Firstly, we build the tree structure for GroupByLevelNode
    Map<TRegionReplicaSet, List<SeriesAggregationSourceNode>> sourceGroup =
        splitAggregationSourceByPartition(root, context);

    boolean containsSlidingWindow =
        root.getChildren().size() == 1
            && root.getChildren().get(0) instanceof SlidingWindowAggregationNode;

    GroupByLevelNode newRoot =
        containsSlidingWindow
            ? groupSourcesForGroupByLevelWithSlidingWindow(
                root,
                (SlidingWindowAggregationNode) root.getChildren().get(0),
                sourceGroup,
                context)
            : groupSourcesForGroupByLevel(root, sourceGroup, context);

    // Then, we calculate the attributes for GroupByLevelNode in each level
    Map<String, List<Expression>> columnNameToExpression = new HashMap<>();
    for (CrossSeriesAggregationDescriptor originalDescriptor :
        newRoot.getGroupByLevelDescriptors()) {
      columnNameToExpression.putAll(originalDescriptor.getGroupedInputStringToExpressionsMap());
      // for example, max_by(root.*.x1, root.*.y1), we have "root.*.x1, root.*.y1" =>
      // [root.*.x1(TimeSeriesOperand), root.*.y1(TimeSeriesOperand)]
      columnNameToExpression.put(
          originalDescriptor.getParametersString(), originalDescriptor.getOutputExpressions());
    }

    context.setColumnNameToExpression(columnNameToExpression);
    calculateGroupByLevelNodeAttributes(newRoot, 0, context);
    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