Join DynamicFilter enhancements

[adriangb]

Tracking issue for improvements to dynamic filters w/ joins

• Generate filters for more join types
• Allow parent filters to pass through in more cases (Support parent dynamic filters for more join types #20192)
• Support nested joins (Enable dynamic filters for nested joins #20214)

Generate filters for more join types

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.

[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.

[adriangb]

@nuno-faria @jackkleeman I think you are both interested in fixing the bugs in dynamic filters for joins + supporting more use cases (nested joins, more join types). I'd love for you to work together on this project. I'll be happy to help out with reviews, etc.

[jackkleeman]

👋 hi @nuno-faria !