class MoveFilterBeneathUnion implements Rule<Filter> {

    private final Capture<Union> unionCapture;
    private final Pattern<Filter> pattern;

    public MoveFilterBeneathUnion() {
        this.unionCapture = new Capture<>();
        this.pattern = typeOf(Filter.class)
            .with(source(), typeOf(Union.class).capturedAs(unionCapture));
    }

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

    @Override
    public LogicalPlan apply(Filter filter,
                             Captures captures,
                             PlanStats planStats,
                             TransactionContext txnCtx,
                             NodeContext nodeCtx,
                             UnaryOperator<LogicalPlan> resolvePlan) {
        Union union = captures.get(unionCapture);
        LogicalPlan lhs = union.sources().get(0);
        LogicalPlan rhs = union.sources().get(1);
        return union.replaceSources(List.of(
            createNewFilter(filter, lhs),
            createNewFilter(filter, rhs)
        ));
    }

    private static Filter createNewFilter(Filter filter, LogicalPlan newSource) {<FILL_FUNCTION_BODY>}
}

        Symbol newQuery = FieldReplacer.replaceFields(filter.query(), f -> {
            int idx = filter.source().outputs().indexOf(f);
            if (idx < 0) {
                throw new IllegalArgumentException(
                    "Field used in filter must be present in its source outputs." +
                    f + " missing in " + filter.source().outputs());
            }
            return newSource.outputs().get(idx);
        });
        return new Filter(newSource, newQuery);
    
<no_super_class>