class GraphBuilder extends LogicalPlanVisitor<Map<Symbol, LogicalPlan>, JoinGraph> {

        private final UnaryOperator<LogicalPlan> resolvePlan;

        GraphBuilder(UnaryOperator<LogicalPlan> resolvePlan) {
            this.resolvePlan = resolvePlan;
        }

        @Override
        public JoinGraph visitPlan(LogicalPlan logicalPlan, Map<Symbol, LogicalPlan> context) {
            for (Symbol output : logicalPlan.outputs()) {
                context.put(output, logicalPlan);
            }
            return new JoinGraph(List.of(logicalPlan), Map.of(), List.of(), false);
        }

        @Override
        public JoinGraph visitGroupReference(GroupReference groupReference, Map<Symbol, LogicalPlan> context) {
            return resolvePlan.apply(groupReference).accept(this, context);
        }

        @Override
        public JoinGraph visitFilter(Filter filter, Map<Symbol, LogicalPlan> context) {
            var source = filter.source().accept(this, context);
            return source.withFilters(List.of(filter.query()));
        }

        @Override
        public JoinGraph visitJoinPlan(JoinPlan joinPlan, Map<Symbol, LogicalPlan> context) {<FILL_FUNCTION_BODY>}

        private static class EdgeCollector extends SymbolVisitor<Map<Symbol, LogicalPlan>, Void> {

            private final Map<LogicalPlan, Set<Edge>> edges = new HashMap<>();
            private final List<LogicalPlan> sources = new ArrayList<>();

            @Override
            public Void visitField(ScopedSymbol s, Map<Symbol, LogicalPlan> context) {
                sources.add(context.get(s));
                return null;
            }

            @Override
            public Void visitReference(Reference ref, Map<Symbol, LogicalPlan> context) {
                sources.add(context.get(ref));
                return null;
            }

            @Override
            public Void visitFunction(io.crate.expression.symbol.Function f, Map<Symbol, LogicalPlan> context) {
                var sizeSource = sources.size();
                f.arguments().forEach(x -> x.accept(this, context));
                if (f.name().equals(EqOperator.NAME)) {
                    assert sources.size() == sizeSource + 2 : "Source must be collected for each argument";
                    var fromSymbol = f.arguments().get(0);
                    var toSymbol = f.arguments().get(1);
                    var fromRelation = sources.get(sources.size() - 2);
                    var toRelation = sources.get(sources.size() - 1);
                    if (fromRelation != null && toRelation != null) {
                        // Edges are created and indexed for each equi-join condition
                        // from both directions e.g.:
                        // a.x = b.y
                        // becomes:
                        // a -> Edge[b, a.x, b.y]
                        // b -> Edge[a, a.x, b.y]
                        addEdge(fromRelation, new Edge(toRelation, fromSymbol, toSymbol));
                        addEdge(toRelation, new Edge(fromRelation, fromSymbol, toSymbol));
                    }
                }
                return null;
            }

            private void addEdge(LogicalPlan from, Edge edge) {
                var values = edges.get(from);
                if (values == null) {
                    values = Set.of(edge);
                } else {
                    values = new HashSet<>(values);
                    values.add(edge);
                }
                edges.put(from, values);
            }
        }
    }


            var left = joinPlan.lhs().accept(this, context);
            var right = joinPlan.rhs().accept(this, context);

            if (joinPlan.joinType() == JoinType.CROSS) {
                return left.joinWith(right).withCrossJoin(true);
            }

            if (joinPlan.joinType() != JoinType.INNER) {
                return left.joinWith(right);
            }

            var joinCondition = joinPlan.joinCondition();
            var edgeCollector = new EdgeCollector();
            var filters = new ArrayList<Symbol>();
            if (joinCondition != null) {
                var split = QuerySplitter.split(joinCondition);
                for (var entry : split.entrySet()) {
                    // we are only interested in equi-join conditions between
                    // two tables e.g.: a.x = b.y will result in
                    // (a,b) -> (a.x = b.y) and we can ignore any other
                    // filters. Therefore, we only want entries where we have
                    // two keys.
                    if (entry.getKey().size() == 2) {
                        entry.getValue().accept(edgeCollector, context);
                    } else {
                        filters.add(entry.getValue());
                    }
                }
            }
            return left.joinWith(right).withEdges(edgeCollector.edges).withFilters(filters);
        
<methods>public non-sealed void <init>() ,public JoinGraph visitCollect(io.crate.planner.operators.Collect, Map<io.crate.expression.symbol.Symbol,io.crate.planner.operators.LogicalPlan>) ,public JoinGraph visitCorrelatedJoin(io.crate.planner.operators.CorrelatedJoin, Map<io.crate.expression.symbol.Symbol,io.crate.planner.operators.LogicalPlan>) ,public JoinGraph visitCount(io.crate.planner.operators.Count, Map<io.crate.expression.symbol.Symbol,io.crate.planner.operators.LogicalPlan>) ,public JoinGraph visitEval(io.crate.planner.operators.Eval, Map<io.crate.expression.symbol.Symbol,io.crate.planner.operators.LogicalPlan>) ,public JoinGraph visitFetch(io.crate.planner.operators.Fetch, Map<io.crate.expression.symbol.Symbol,io.crate.planner.operators.LogicalPlan>) ,public JoinGraph visitFilter(io.crate.planner.operators.Filter, Map<io.crate.expression.symbol.Symbol,io.crate.planner.operators.LogicalPlan>) ,public JoinGraph visitForeignCollect(io.crate.planner.operators.ForeignCollect, Map<io.crate.expression.symbol.Symbol,io.crate.planner.operators.LogicalPlan>) ,public JoinGraph visitGet(io.crate.planner.operators.Get, Map<io.crate.expression.symbol.Symbol,io.crate.planner.operators.LogicalPlan>) ,public JoinGraph visitGroupHashAggregate(io.crate.planner.operators.GroupHashAggregate, Map<io.crate.expression.symbol.Symbol,io.crate.planner.operators.LogicalPlan>) ,public JoinGraph visitGroupReference(io.crate.planner.optimizer.iterative.GroupReference, Map<io.crate.expression.symbol.Symbol,io.crate.planner.operators.LogicalPlan>) ,public JoinGraph visitHashAggregate(io.crate.planner.operators.HashAggregate, Map<io.crate.expression.symbol.Symbol,io.crate.planner.operators.LogicalPlan>) ,public JoinGraph visitHashJoin(io.crate.planner.operators.HashJoin, Map<io.crate.expression.symbol.Symbol,io.crate.planner.operators.LogicalPlan>) ,public JoinGraph visitInsert(io.crate.planner.operators.Insert, Map<io.crate.expression.symbol.Symbol,io.crate.planner.operators.LogicalPlan>) ,public JoinGraph visitInsert(io.crate.planner.operators.InsertFromValues, Map<io.crate.expression.symbol.Symbol,io.crate.planner.operators.LogicalPlan>) ,public JoinGraph visitJoinPlan(io.crate.planner.operators.JoinPlan, Map<io.crate.expression.symbol.Symbol,io.crate.planner.operators.LogicalPlan>) ,public JoinGraph visitLimit(io.crate.planner.operators.Limit, Map<io.crate.expression.symbol.Symbol,io.crate.planner.operators.LogicalPlan>) ,public JoinGraph visitLimitDistinct(io.crate.planner.operators.LimitDistinct, Map<io.crate.expression.symbol.Symbol,io.crate.planner.operators.LogicalPlan>) ,public JoinGraph visitMultiPhase(io.crate.planner.operators.MultiPhase, Map<io.crate.expression.symbol.Symbol,io.crate.planner.operators.LogicalPlan>) ,public JoinGraph visitNestedLoopJoin(io.crate.planner.operators.NestedLoopJoin, Map<io.crate.expression.symbol.Symbol,io.crate.planner.operators.LogicalPlan>) ,public JoinGraph visitOrder(io.crate.planner.operators.Order, Map<io.crate.expression.symbol.Symbol,io.crate.planner.operators.LogicalPlan>) ,public JoinGraph visitPlan(io.crate.planner.operators.LogicalPlan, Map<io.crate.expression.symbol.Symbol,io.crate.planner.operators.LogicalPlan>) ,public JoinGraph visitProjectSet(io.crate.planner.operators.ProjectSet, Map<io.crate.expression.symbol.Symbol,io.crate.planner.operators.LogicalPlan>) ,public JoinGraph visitRename(io.crate.planner.operators.Rename, Map<io.crate.expression.symbol.Symbol,io.crate.planner.operators.LogicalPlan>) ,public JoinGraph visitRootRelationBoundary(io.crate.planner.operators.RootRelationBoundary, Map<io.crate.expression.symbol.Symbol,io.crate.planner.operators.LogicalPlan>) ,public JoinGraph visitTableFunction(io.crate.planner.operators.TableFunction, Map<io.crate.expression.symbol.Symbol,io.crate.planner.operators.LogicalPlan>) ,public JoinGraph visitUnion(io.crate.planner.operators.Union, Map<io.crate.expression.symbol.Symbol,io.crate.planner.operators.LogicalPlan>) ,public JoinGraph visitWindowAgg(io.crate.planner.operators.WindowAgg, Map<io.crate.expression.symbol.Symbol,io.crate.planner.operators.LogicalPlan>) <variables>