class MoveFilterBeneathJoin implements Rule<Filter> {

    private final Capture<AbstractJoinPlan> joinCapture;
    private final Pattern<Filter> pattern;
    private static final Set<JoinType> SUPPORTED_JOIN_TYPES = EnumSet.of(INNER, LEFT, RIGHT, CROSS);

    public MoveFilterBeneathJoin() {
        this.joinCapture = new Capture<>();
        this.pattern = typeOf(Filter.class)
            .with(source(),
                typeOf(AbstractJoinPlan.class)
                    .capturedAs(joinCapture)
                    .with(join -> SUPPORTED_JOIN_TYPES.contains(join.joinType()))
            );
    }

    @Override
    public Pattern<Filter> pattern() {
        return pattern;
    }

    @Override
    public LogicalPlan apply(Filter filter,
                             Captures captures,
                             PlanStats planStats,
                             TransactionContext txnCtx,
                             NodeContext nodeCtx,
                             UnaryOperator<LogicalPlan> resolvePlan) {<FILL_FUNCTION_BODY>}

    static LogicalPlan getNewSource(@Nullable Symbol splitQuery, LogicalPlan source) {
        return splitQuery == null ? source : new Filter(source, splitQuery);
    }
}

        var join = captures.get(joinCapture);
        var query = filter.query();
        if (!WhereClause.canMatch(query)) {
            return join.replaceSources(List.of(
                getNewSource(query, join.lhs()),
                getNewSource(query, join.rhs())
            ));
        }
        var splitQueries = QuerySplitter.split(query);
        final int initialParts = splitQueries.size();
        if (splitQueries.size() == 1 && splitQueries.keySet().iterator().next().size() > 1) {
            return null;
        }
        var lhs = join.lhs();
        var rhs = join.rhs();

        var lhsRelations = new HashSet<>(lhs.relationNames());
        var rhsRelations = new HashSet<>(rhs.relationNames());

        var leftQuery = splitQueries.remove(lhsRelations);
        var rightQuery = splitQueries.remove(rhsRelations);

        if (leftQuery == null && rightQuery == null) {
            // we don't have a match for the filter on rhs/lhs yet
            // let's see if we have partial match with a subsection of the relations
            var it = splitQueries.entrySet().iterator();
            while (it.hasNext()) {
                var entry = it.next();
                var relationNames = entry.getKey();
                var splitQuery = entry.getValue();

                var matchesLhs = Sets.intersection(lhsRelations, relationNames);
                var matchesRhs = Sets.intersection(rhsRelations, relationNames);

                if (matchesRhs.isEmpty() == false && matchesLhs.isEmpty()) {
                    rightQuery = rightQuery == null ? splitQuery : AndOperator.of(rightQuery, splitQuery);
                    it.remove();
                } else if (matchesRhs.isEmpty() && matchesLhs.isEmpty() == false) {
                    leftQuery = leftQuery == null ? splitQuery : AndOperator.of(leftQuery, splitQuery);
                    it.remove();
                }
            }
        }

        var newLhs = lhs;
        var newRhs = rhs;
        var joinType = join.joinType();
        if (joinType == JoinType.INNER || joinType == JoinType.CROSS) {
            newLhs = getNewSource(leftQuery, lhs);
            newRhs = getNewSource(rightQuery, rhs);
        } else {
            if (joinType == JoinType.LEFT) {
                newLhs = getNewSource(leftQuery, lhs);
                if (rightQuery != null) {
                    splitQueries.put(rhsRelations, rightQuery);
                }
            } else if (joinType == JoinType.RIGHT) {
                newRhs = getNewSource(rightQuery, rhs);
                if (leftQuery != null) {
                    splitQueries.put(lhsRelations, leftQuery);
                }
            }
        }

        if (newLhs == lhs && newRhs == rhs) {
            return null;
        }

        var newJoin = join.replaceSources(List.of(newLhs, newRhs));

        if (splitQueries.isEmpty()) {
            return newJoin;
        } else if (initialParts == splitQueries.size()) {
            return null;
        } else {
            return new Filter(newJoin, AndOperator.join(splitQueries.values()));
        }
    
<no_super_class>