private void calculateGroupByLevelNodeAttributes(
      PlanNode node, int level, DistributionPlanContext context) {
    if (node == null) {
      return;
    }
    node.getChildren()
        .forEach(child -> calculateGroupByLevelNodeAttributes(child, level + 1, context));

    // Construct all outputColumns from children. Using Set here to avoid duplication
    Set<String> childrenOutputColumns = new HashSet<>();
    node.getChildren().forEach(child -> childrenOutputColumns.addAll(child.getOutputColumnNames()));

    if (node instanceof SlidingWindowAggregationNode) {
      SlidingWindowAggregationNode handle = (SlidingWindowAggregationNode) node;
      List<AggregationDescriptor> descriptorList = new ArrayList<>();
      for (AggregationDescriptor originalDescriptor : handle.getAggregationDescriptorList()) {
        boolean keep = false;
        for (String childColumn : childrenOutputColumns) {
          for (String groupedInputExpressionsString :
              originalDescriptor.getInputExpressionsAsStringList()) {
            if (isAggColumnMatchExpression(childColumn, groupedInputExpressionsString)) {
              keep = true;
              break;
            }
          }
        }
        if (keep) {
          descriptorList.add(originalDescriptor);
          updateTypeProviderByPartialAggregation(
              originalDescriptor, context.queryContext.getTypeProvider());
        }
      }
      handle.setAggregationDescriptorList(descriptorList);
    }

    if (node instanceof GroupByLevelNode) {
      GroupByLevelNode handle = (GroupByLevelNode) node;
      // Check every OutputColumn of GroupByLevelNode and set the Expression of corresponding
      // AggregationDescriptor
      List<CrossSeriesAggregationDescriptor> descriptorList = new ArrayList<>();
      Map<String, List<Expression>> columnNameToExpression = context.getColumnNameToExpression();
      Map<String, List<Expression>> childrenExpressionMap = new HashMap<>();
      for (String childColumn : childrenOutputColumns) {
        String childInput =
            childColumn.substring(childColumn.indexOf("(") + 1, childColumn.lastIndexOf(")"));
        childrenExpressionMap.put(childInput, columnNameToExpression.get(childInput));
      }

      for (CrossSeriesAggregationDescriptor originalDescriptor :
          handle.getGroupByLevelDescriptors()) {
        Set<Expression> descriptorExpressions = new HashSet<>();

        if (childrenExpressionMap.containsKey(originalDescriptor.getParametersString())) {
          descriptorExpressions.addAll(originalDescriptor.getOutputExpressions());
        }

        if (analysis.useLogicalView()) {
          for (List<Expression> groupedInputExpressions :
              originalDescriptor.getGroupedInputExpressions()) {
            String groupedInputExpressionsString =
                originalDescriptor.getInputString(groupedInputExpressions);
            List<Expression> inputExpressions =
                childrenExpressionMap.get(groupedInputExpressionsString);
            if (inputExpressions != null && !inputExpressions.isEmpty()) {
              descriptorExpressions.addAll(groupedInputExpressions);
            }
          }
        } else {
          for (String groupedInputExpressionString :
              originalDescriptor.getGroupedInputExpressionStrings()) {
            List<Expression> inputExpressions =
                childrenExpressionMap.get(groupedInputExpressionString);
            if (inputExpressions != null && !inputExpressions.isEmpty()) {
              descriptorExpressions.addAll(inputExpressions);
            }
          }
        }

        if (descriptorExpressions.isEmpty()) {
          continue;
        }
        CrossSeriesAggregationDescriptor descriptor = originalDescriptor.deepClone();
        descriptor.setStep(level == 0 ? AggregationStep.FINAL : AggregationStep.INTERMEDIATE);
        descriptor.setInputExpressions(new ArrayList<>(descriptorExpressions));
        descriptorList.add(descriptor);
        updateTypeProviderByPartialAggregation(descriptor, context.queryContext.getTypeProvider());
      }
      handle.setGroupByLevelDescriptors(descriptorList);
    }
  }
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