presto-main/src/main/java/io/prestosql/sql/planner/iterative/rule/TransformCorrelatedScalarSubquery.java/TransformCorrelatedScalarSubquery/apply
class TransformCorrelatedScalarSubquery:    @Override
    public Result apply(CorrelatedJoinNode correlatedJoinNode, Captures captures, Context context)
    {
        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();

        Range<Long> subqueryCardinality = extractCardinality(rewrittenSubquery, context.getLookup());
        boolean producesAtMostOneRow = Range.closed(0L, 1L).encloses(subqueryCardinality);
        if (producesAtMostOneRow) {
            boolean producesSingleRow = Range.singleton(1L).encloses(subqueryCardinality);
            return Result.ofPlanNode(new CorrelatedJoinNode(
                    context.getIdAllocator().getNextId(),
                    correlatedJoinNode.getInput(),
                    rewrittenSubquery,
                    correlatedJoinNode.getCorrelation(),
                    producesSingleRow ? correlatedJoinNode.getType() : LEFT,
                    correlatedJoinNode.getFilter(),
                    correlatedJoinNode.getOriginSubquery()));
        }

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

        CorrelatedJoinNode rewrittenCorrelatedJoinNode = new CorrelatedJoinNode(
                context.getIdAllocator().getNextId(),
                new AssignUniqueId(
                        context.getIdAllocator().getNextId(),
                        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().getNextId(),
                rewrittenCorrelatedJoinNode,
                isDistinct,
                rewrittenCorrelatedJoinNode.getInput().getOutputSymbols(),
                Optional.empty());

        FilterNode filterNode = new FilterNode(
                context.getIdAllocator().getNextId(),
                markDistinctNode,
                new SimpleCaseExpression(
                        isDistinct.toSymbolReference(),
                        ImmutableList.of(
                                new WhenClause(TRUE_LITERAL, TRUE_LITERAL)),
                        Optional.of(new Cast(
                                new FunctionCallBuilder(metadata)
                                        .setName(QualifiedName.of("fail"))
                                        .addArgument(INTEGER, new LongLiteral(Integer.toString(SUBQUERY_MULTIPLE_ROWS.toErrorCode().getCode())))
                                        .addArgument(VARCHAR, new StringLiteral("Scalar sub-query has returned multiple rows"))
                                        .build(),
                                toSqlType(BOOLEAN)))));

        return Result.ofPlanNode(new ProjectNode(
                context.getIdAllocator().getNextId(),
                filterNode,
                Assignments.identity(correlatedJoinNode.getOutputSymbols())));
    }