@Override
  public Result apply(CorrelatedJoinNode correlatedJoinNode, Captures captures, Context context) {
    PlanNode source = captures.get(SOURCE);

    // if we fail to decorrelate the nested plan, and it contains a distinct operator, we can
    // extract and special-handle the distinct operator
    AggregationNode distinct = null;

    // decorrelate nested plan
    PlanNodeDecorrelator decorrelator =
        new PlanNodeDecorrelator(plannerContext, context.getSymbolAllocator(), context.getLookup());
    Optional<PlanNodeDecorrelator.DecorrelatedNode> decorrelatedSource =
        decorrelator.decorrelateFilters(source, correlatedJoinNode.getCorrelation());
    if (!decorrelatedSource.isPresent()) {
      // we failed to decorrelate the nested plan, so check if we can extract a distinct operator
      // from the nested plan
      if (isDistinctOperator(source)) {
        distinct = (AggregationNode) source;
        source = distinct.getChild();
        decorrelatedSource =
            decorrelator.decorrelateFilters(source, correlatedJoinNode.getCorrelation());
      }
      if (!decorrelatedSource.isPresent()) {
        return Result.empty();
      }
    }

    source = decorrelatedSource.get().getNode();

    // assign unique id on correlated join's input. It will be used to distinguish between original
    // input rows after join
    PlanNode inputWithUniqueId =
        new AssignUniqueId(
            context.getIdAllocator().genPlanNodeId(),
            correlatedJoinNode.getInput(),
            context.getSymbolAllocator().newSymbol("unique", BooleanType.getInstance()));

    JoinNode join =
        new JoinNode(
            context.getIdAllocator().genPlanNodeId(),
            JoinNode.JoinType.INNER,
            inputWithUniqueId,
            source,
            ImmutableList.of(),
            Optional.empty(),
            inputWithUniqueId.getOutputSymbols(),
            source.getOutputSymbols(),
            decorrelatedSource.get().getCorrelatedPredicates(),
            Optional.empty());

    // restore distinct aggregation
    if (distinct != null) {
      distinct =
          restoreDistinctAggregation(
              distinct,
              join,
              ImmutableList.<Symbol>builder()
                  .addAll(join.getLeftOutputSymbols())
                  .addAll(distinct.getGroupingKeys())
                  .build());
    }

    // restore grouped aggregation
    AggregationNode groupedAggregation = captures.get(AGGREGATION);
    groupedAggregation =
        AggregationNode.builderFrom(groupedAggregation)
            .setSource(distinct != null ? distinct : join)
            .setGroupingSets(
                singleGroupingSet(
                    ImmutableList.<Symbol>builder()
                        .addAll(join.getLeftOutputSymbols())
                        .addAll(groupedAggregation.getGroupingKeys())
                        .build()))
            .setPreGroupedSymbols(ImmutableList.of())
            .setHashSymbol(Optional.empty())
            .setGroupIdSymbol(Optional.empty())
            .build();

    // restrict outputs and apply projection
    Set<Symbol> outputSymbols = new HashSet<>(correlatedJoinNode.getOutputSymbols());
    List<Symbol> expectedAggregationOutputs =
        groupedAggregation.getOutputSymbols().stream()
            .filter(outputSymbols::contains)
            .collect(toImmutableList());

    Assignments assignments =
        Assignments.builder()
            .putIdentities(expectedAggregationOutputs)
            .putAll(captures.get(PROJECTION).getAssignments())
            .build();

    return Result.ofPlanNode(
        new ProjectNode(context.getIdAllocator().genPlanNodeId(), groupedAggregation, assignments));
  }
This rule decorrelates a correlated subquery of INNER correlated join with: - single grouped
aggregation, or - grouped aggregation over distinct operator (grouped aggregation with no
aggregation assignments), in case when the distinct operator cannot be de-correlated by
PlanNodeDecorrelator

<p>In the case of single aggregation, it transforms:

<pre>
- CorrelatedJoin INNER (correlation: [c], filter: true, output: a, count, agg)
     - Input (a, c)
     - Project (x <- f(count), y <- f'(agg))
          - Aggregation (group by b)
            count <- count(*)
            agg <- agg(d)
               - Source (b, d) with correlated filter (b > c)
</pre>

Into:

<pre>
- Project (a <- a, x <- f(count), y <- f'(agg))
     - Aggregation (group by [a, c, unique, b])
       count <- count(*)
       agg <- agg(d)
          - INNER join (filter: b > c)
               - UniqueId (unique)
                    - Input (a, c)
               - Source (b, d) decorrelated
</pre>

<p>In the case of grouped aggregation over distinct operator, it transforms:

<pre>
- CorrelatedJoin INNER (correlation: [c], filter: true, output: a, count, agg)
     - Input (a, c)
     - Project (x <- f(count), y <- f'(agg))
          - Aggregation (group by b)
            count <- count(*)
            agg <- agg(b)
               - Aggregation "distinct operator" group by [b]
                    - Source (b) with correlated filter (b > c)
</pre>

Into:

<pre>
- Project (a <- a, x <- f(count), y <- f'(agg))
     - Aggregation (group by [a, c, unique, b])
       count <- count(*)
       agg <- agg(b)
          - Aggregation "distinct operator" group by [a, c, unique, b]
               - INNER join (filter: b > c)
                    - UniqueId (unique)
                         - Input (a, c)
                    - Source (b) decorrelated
</pre>
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/TransformCorrelatedGroupedAggregationWithProjection.java