@Override
  public Result apply(CorrelatedJoinNode correlatedJoinNode, Captures captures, Context context) {
    // decorrelate nested plan
    PlanNodeDecorrelator decorrelator =
        new PlanNodeDecorrelator(plannerContext, context.getSymbolAllocator(), context.getLookup());
    Optional<PlanNodeDecorrelator.DecorrelatedNode> decorrelatedSource =
        decorrelator.decorrelateFilters(
            captures.get(AGGREGATION).getChild(), correlatedJoinNode.getCorrelation());
    if (!decorrelatedSource.isPresent()) {
      return Result.empty();
    }

    PlanNode 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", LongType.getInstance()));

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

    // restore aggregation
    AggregationNode aggregation = captures.get(AGGREGATION);
    aggregation =
        new AggregationNode(
            aggregation.getPlanNodeId(),
            join,
            aggregation.getAggregations(),
            singleGroupingSet(
                ImmutableList.<Symbol>builder()
                    .addAll(join.getLeftOutputSymbols())
                    .addAll(aggregation.getGroupingKeys())
                    .build()),
            ImmutableList.of(),
            aggregation.getStep(),
            Optional.empty(),
            Optional.empty());

    // restrict outputs and apply projection
    Set<Symbol> outputSymbols = new HashSet<>(correlatedJoinNode.getOutputSymbols());
    List<Symbol> expectedAggregationOutputs =
        aggregation.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(), aggregation, assignments));
  }
This rule decorrelates a correlated subquery of LEFT correlated join with distinct operator
(grouped aggregation with no aggregation assignments)

<p>Transforms:

<pre>
- CorrelatedJoin LEFT (correlation: [c], filter: true, output: a, x)
     - Input (a, c)
     - Project (x <- b + 100)
          - Aggregation "distinct operator" group by [b]
               - Source (b) with correlated filter (b > c)
</pre>

Into:

<pre>
- Project (a <- a, x <- b + 100)
     - Aggregation "distinct operator" group by [a, c, unique, b]
          - LEFT 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/TransformCorrelatedDistinctAggregationWithProjection.java