apache / apache/datafusion

Support filter pushdown through `SortMergeJoinExec`

Open
#20,443 5 comments 2 reactions 2 assignees Claimed by @stuhood View on GitHub
enhancement
Dominant language
Rust
Stars
9.3k
Forks
2.4k
Avg merge
3d 7h
Merged PRs (30d)
344

Description

### Is your feature request related to a problem or challenge?

`SortMergeJoinExec` currently relies on the default `gather_filters_for_pushdown` implementation, which marks all parent filters as unsupported. This prevents `DynamicFilterPhysicalExpr` (e.g. from `SortExec`'s TopK) from being pushed through sort-merge joins to reach scan nodes below them.

`HashJoinExec` already implements proper filter routing via `FilterDescription::from_children` + `FilterPushdownPropagation::if_any`, which routes each parent filter to whichever child's schema contains the referenced columns. There's no reason `SortMergeJoinExec` can't do the same — both join types have clearly delineated left/right children with independent schemas.

As a workaround, we currently wrap `SortMergeJoinExec` in a thin passthrough node that overrides the two filter-pushdown methods. This works but adds an unnecessary layer of indirection.

### Describe the solution you'd like

Implement `gather_filters_for_pushdown` and `handle_child_pushdown_result` on `SortMergeJoinExec`, following the same pattern as `HashJoinExec`:

```rust
// gather_filters_for_pushdown
fn gather_filters_for_pushdown(
&self,
_phase: FilterPushdownPhase,
parent_filters: Vec>,
_config: &ConfigOptions,
) -> Result {
if self.join_type != JoinType::Inner {
return Ok(FilterDescription::all_unsupported(
&parent_filters,
&self.children(),
));
}
FilterDescription::from_children(parent_filters, &self.children())
}

// handle_child_pushdown_result
fn handle_child_pushdown_result(
&self,
_phase: FilterPushdownPhase,
child_pushdown_result: ChildPushdownResult,
_config: &ConfigOptions,
) -> Result>> {
if self.join_type != JoinType::Inner {
return Ok(FilterPushdownPropagation::all_unsupported(child_pushdown_result));
}
Ok(FilterPushdownPropagation::if_any(child_pushdown_result))
}
```

This uses `from_children` for column-based routing (a filter on column `x` is routed only to the child whose schema contains `x`) and `if_any` for propagation (matching `HashJoinExec`'s behavior, since a filter can only apply to one child).

### Describe alternatives you've considered

Wrapping `SortMergeJoinExec` in a custom `ExecutionPlan` that overrides `gather_filters_for_pushdown` and `handle_child_pushdown_result`. This is what we're currently doing — it works, but adds unnecessary plan complexity. Ideally this should be built into `SortMergeJoinExec` itself.

### Additional context

The same gap may apply to other join types (`NestedLoopJoinExec`, `SymmetricHashJoinExec`, `CrossJoinExec`) that also use the default no-op filter pushdown. `SortMergeJoinExec` is the most impactful since it's used for sorted-input joins where TopK dynamic filters are most beneficial.

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.