presto-main/src/main/java/io/prestosql/sql/planner/optimizations/PruneUnreferencedOutputs.java/Rewriter/visitCorrelatedJoin
class Rewriter:        @Override
        public PlanNode visitCorrelatedJoin(CorrelatedJoinNode node, RewriteContext<Set<Symbol>> context)
        {
            Set<Symbol> expectedFilterSymbols = SymbolsExtractor.extractUnique(node.getFilter());

            Set<Symbol> expectedFilterAndContextSymbols = ImmutableSet.<Symbol>builder()
                    .addAll(expectedFilterSymbols)
                    .addAll(context.get())
                    .build();

            PlanNode subquery = context.rewrite(node.getSubquery(), expectedFilterAndContextSymbols);

            // remove unused correlated join nodes
            if (intersection(ImmutableSet.copyOf(subquery.getOutputSymbols()), context.get()).isEmpty()) {
                // remove unused subquery of inner join
                if (node.getType() == INNER && isScalar(subquery) && node.getFilter().equals(TRUE_LITERAL)) {
                    return context.rewrite(node.getInput(), context.get());
                }
                // remove unused subquery of left join
                if (node.getType() == LEFT && isAtMostScalar(subquery)) {
                    return context.rewrite(node.getInput(), context.get());
                }
            }

            // prune not used correlation symbols
            Set<Symbol> subquerySymbols = SymbolsExtractor.extractUnique(subquery);
            List<Symbol> newCorrelation = node.getCorrelation().stream()
                    .filter(subquerySymbols::contains)
                    .collect(toImmutableList());

            Set<Symbol> expectedCorrelationAndContextSymbols = ImmutableSet.<Symbol>builder()
                    .addAll(newCorrelation)
                    .addAll(context.get())
                    .build();
            Set<Symbol> inputContext = ImmutableSet.<Symbol>builder()
                    .addAll(expectedCorrelationAndContextSymbols)
                    .addAll(expectedFilterSymbols)
                    .build();
            PlanNode input = context.rewrite(node.getInput(), inputContext);

            // remove unused input nodes
            if (intersection(ImmutableSet.copyOf(input.getOutputSymbols()), expectedCorrelationAndContextSymbols).isEmpty()) {
                // remove unused input of inner join
                if (node.getType() == INNER && isScalar(input) && node.getFilter().equals(TRUE_LITERAL)) {
                    return subquery;
                }
                // remove unused input of right join
                if (node.getType() == RIGHT && isAtMostScalar(input)) {
                    return subquery;
                }
            }

            return new CorrelatedJoinNode(node.getId(), input, subquery, newCorrelation, node.getType(), node.getFilter(), node.getOriginSubquery());
        }