@Override
  public Result apply(CorrelatedJoinNode correlatedJoinNode, Captures captures, Context context) {
    // lateral references are only allowed for INNER or LEFT correlated join
    checkArgument(
        correlatedJoinNode.getJoinType() == INNER || correlatedJoinNode.getJoinType() == LEFT,
        "unexpected correlated join type: %s",
        correlatedJoinNode.getJoinType());
    PlanNode subquery = context.getLookup().resolve(correlatedJoinNode.getSubquery());

    if (!searchFrom(subquery, context.getLookup())
        .where(EnforceSingleRowNode.class::isInstance)
        .recurseOnlyWhen(ProjectNode.class::isInstance)
        .matches()) {
      return Result.empty();
    }

    PlanNode rewrittenSubquery =
        searchFrom(subquery, context.getLookup())
            .where(EnforceSingleRowNode.class::isInstance)
            .recurseOnlyWhen(ProjectNode.class::isInstance)
            .removeFirst();

    Cardinality subqueryCardinality = extractCardinality(rewrittenSubquery, context.getLookup());
    boolean producesAtMostOneRow = subqueryCardinality.isAtMostScalar();
    if (producesAtMostOneRow) {
      boolean producesSingleRow = subqueryCardinality.isScalar();
      return Result.ofPlanNode(
          new CorrelatedJoinNode(
              context.getIdAllocator().genPlanNodeId(),
              correlatedJoinNode.getInput(),
              rewrittenSubquery,
              correlatedJoinNode.getCorrelation(),
              // EnforceSingleRowNode guarantees that exactly single matching row is produced
              // for every input row (independently of correlated join type). Decorrelated plan
              // must preserve this semantics.
              producesSingleRow ? INNER : LEFT,
              correlatedJoinNode.getFilter(),
              correlatedJoinNode.getOriginSubquery()));
    }

    Symbol unique = context.getSymbolAllocator().newSymbol("unique", INT64);

    CorrelatedJoinNode rewrittenCorrelatedJoinNode =
        new CorrelatedJoinNode(
            context.getIdAllocator().genPlanNodeId(),
            new AssignUniqueId(
                context.getIdAllocator().genPlanNodeId(), correlatedJoinNode.getInput(), unique),
            rewrittenSubquery,
            correlatedJoinNode.getCorrelation(),
            LEFT,
            correlatedJoinNode.getFilter(),
            correlatedJoinNode.getOriginSubquery());

    Symbol isDistinct = context.getSymbolAllocator().newSymbol("is_distinct", BOOLEAN);
    MarkDistinctNode markDistinctNode =
        new MarkDistinctNode(
            context.getIdAllocator().genPlanNodeId(),
            rewrittenCorrelatedJoinNode,
            isDistinct,
            rewrittenCorrelatedJoinNode.getInput().getOutputSymbols(),
            Optional.empty());

    FilterNode filterNode =
        new FilterNode(
            context.getIdAllocator().genPlanNodeId(),
            markDistinctNode,
            new SimpleCaseExpression(
                isDistinct.toSymbolReference(),
                ImmutableList.of(new WhenClause(TRUE_LITERAL, TRUE_LITERAL)),
                new Cast(
                    new FunctionCall(
                        QualifiedName.of(FAIL_FUNCTION_NAME),
                        ImmutableList.of(
                            new StringLiteral("Scalar sub-query has returned multiple rows."))),
                    toSqlType(BOOLEAN))));

    return Result.ofPlanNode(
        new ProjectNode(
            context.getIdAllocator().genPlanNodeId(),
            filterNode,
            Assignments.identity(correlatedJoinNode.getOutputSymbols())));
  }
Scalar filter scan query is something like:

<pre>
    SELECT a,b,c FROM rel WHERE a = correlated1 AND b = correlated2
</pre>

<p>This optimizer can rewrite to mark distinct and filter over a left outer join:

<p>From:

<pre>
- CorrelatedJoin (with correlation list: [C])
  - (input) plan which produces symbols: [A, B, C]
  - (scalar subquery) Project F
    - Filter(D = C AND E > 5)
      - plan which produces symbols: [D, E, F]
</pre>

to:

<pre>
- Filter(CASE isDistinct WHEN true THEN true ELSE fail('Scalar sub-query has returned multiple rows'))
  - MarkDistinct(isDistinct)
    - CorrelatedJoin (with correlation list: [C])
      - AssignUniqueId(adds symbol U)
        - (input) plan which produces symbols: [A, B, C]
      - non scalar subquery
</pre>

<p>This must be run after aggregation decorrelation rules.

<p>This rule is used to support non-aggregation scalar subquery.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/TransformCorrelatedScalarSubquery.java