Enable dynamic filter pushdown for non-inner joins in physical plans

[adriangb]

Is your feature request related to a problem or challenge?

In #16954 we are only going to support inner joins because of complexities with other join types.

This would benefit cases such as:

COPY (SELECT 1 AS c1) TO 't1.parquet';
COPY (SELECT 1 AS c1, 2 AS c2) TO 't2.parquet';

CREATE EXTERNAL TABLE t1 (
    c1 INT NOT NULL
)
STORED AS PARQUET
LOCATION  't1.parquet';

CREATE EXTERNAL TABLE t2 (
    c1 INT NOT NULL,
    c2 INT NOT NULL
)
STORED AS PARQUET
LOCATION  't2.parquet';

select *
from t1
left join t2 using (c1)
order by t2.c2
limit 100;

Describe the solution you'd like

An implementation similar to that of the logical optimizer that accounts for the edge cases and allows some filters to be pushed down.

Describe alternatives you've considered

No response

Additional context

No response

[adriangb]

I'm particularly interested in left semi joins since that's what subqueries and where exists queries get turned into:

select trace_id, span_id
from records
where trace_id in (select trace_id from records where is_exception)

explain
select trace_id, span_id
from records r1
where exists (select from records r2 where is_exception and r2.trace_id = r1.trace_id)

[nuno-faria]

Are there also plans to support RightSemi? If the result inside the IN or EXISTS subquery is smaller I think the planner will chose it as the build side, meaning it will use a RightSemi.

For example, using IN:

explain analyze
select *
from orders
where o_custkey in (
    select c_custkey
    from customer
    where c_phone = '25-989-741-2988'
);

HashJoinExec: mode=Partitioned, join_type=RightSemi, on=[(c_custkey@0, o_custkey@1)], filter=[o_custkey@1 >= 1 AND o_custkey@1 <= 1]

~1.2s

Similarly with EXISTS:

explain analyze
select *
from orders
where exists (
    select 1
    from customer
    where c_phone = '25-989-741-2988'
        and c_custkey = o_custkey
);

HashJoinExec: mode=Partitioned, join_type=RightSemi, on=[(c_custkey@0, o_custkey@1)], filter=[o_custkey@1 >= 1 AND o_custkey@1 <= 1]

~1.2s

Meanwhile, an equivalent inner join now only takes around ~90ms, using dynamic filter pushdown.

[adriangb]

Are there also plans to support RightSemi? If the result inside the IN or EXISTS subquery is smaller I think the planner will chose it as the build side, meaning it will use a RightSemi.

Good point. Maybe we can do it in #17153? Would love it you take a look at that PR.

Meanwhile, an equivalent inner join now only takes around ~90ms, using dynamic filter pushdown.

🤯 🚀

[alamb]

Here is the relevant code for pushing filters through joins in the logical optimizer.

https://github.com/apache/datafusion/blob/0c7d8307a8c4f027d3e6b5c1b906c992a33a91b5/datafusion/optimizer/src/push_down_filter.rs#L431-L430

I think we coud implement similar rules in the physical plans

Specifically this code for pushing filters through HashJoinExec:

datafusion/datafusion/physical-plan/src/joins/hash_join/exec.rs

Lines 1145 to 1152 in 49d49fd

	if self.join_type != JoinType::Inner {
	// Other types of joins can support *some* filters, but restrictions are complex and error prone.
	// For now we don't support them.
	// See the logical optimizer rules for more details: datafusion/optimizer/src/push_down_filter.rs
	return Ok(FilterPushdownPropagation::all_unsupported(
	child_pushdown_result,
	));
	}

The pushdown flow is explained here

datafusion/datafusion/physical-plan/src/execution_plan.rs

Lines 594 to 611 in be3842b

	/// **Example Walkthrough:**
	/// Consider a query plan: `FilterExec (f1) -> HashJoinExec -> DataSourceExec`.
	/// 1. **Downward Phase (`gather_filters_for_pushdown`):** Starting at
	/// `FilterExec`, the filter `f1` is gathered and pushed down to
	/// `HashJoinExec`. `HashJoinExec` may allow `f1` to pass to one side of
	/// the join or add its own filters (e.g., a min-max filter from the build side),
	/// then pushes filters to `DataSourceExec`. `DataSourceExec`, being a leaf node,
	/// has no children to push to, so it prepares to handle filters in the
	/// upward phase.
	/// 2. **Upward Phase (`handle_child_pushdown_result`):** Starting at
	/// `DataSourceExec`, it absorbs applicable filters from `HashJoinExec`
	/// for late materialization during scanning, marking them as supported.
	/// `HashJoinExec` receives the result, decides whether to apply any
	/// remaining filters during the join, and passes unhandled filters back
	/// up to `FilterExec`. `FilterExec` absorbs any unhandled filters,
	/// updates its predicate if necessary, or removes itself if the predicate
	/// becomes trivial (e.g., `lit(true)`), and marks filters as supported
	/// for its parent.