You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/05/21 10:42:01 UTC
(datafusion) branch main updated: PushdownFilter optimizations (#21668)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new d318324481 PushdownFilter optimizations (#21668)
d318324481 is described below
commit d318324481352af859fa0dd58d7e26ef597e6af1
Author: Georgi Krastev <ge...@coralogix.com>
AuthorDate: Thu May 21 13:18:55 2026 +0300
PushdownFilter optimizations (#21668)
## Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax. For example
`Closes #123` indicates that this PR will close issue #123.
-->
- Addresses #20002
- Includes #21643
- Looks like we had a similar idea with #21667, cc @kumarUjjawal
- [x] I like the name `try_unchecked` more so I would adopt that
- I'm not sure if we need to keep the unaliasing, the optimizer rule
already does that
- I have additional changes for more aggressive optimizations, so I'm
happy to combine the PRs or rebase after merge
## Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly
in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand
your changes and offer better suggestions for fixes.
-->
## What changes are included in this PR?
- Add a hidden `Filter::new` constructor that skips type-checking
- Less allocations, more modification of mutable plan nodes
- Less cloning, use references when possible
## Are these changes tested?
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example,
are they covered by existing tests)?
-->
Relying on existing tests mostly, added a few more tests.
## Are there any user-facing changes?
<!--
If there are user-facing changes then we may require documentation to be
updated before approving the PR.
-->
<!--
If there are any breaking changes to public APIs, please add the `api
change` label.
-->
`make_filter` is deprecated, probably wasn't meant to be a public
function.
---
datafusion-testing | 2 +-
datafusion/expr/src/logical_plan/plan.rs | 13 +
datafusion/optimizer/src/push_down_filter.rs | 612 ++++++++++-----------
.../simplify_expressions/simplify_predicates.rs | 14 +-
4 files changed, 306 insertions(+), 335 deletions(-)
diff --git a/datafusion-testing b/datafusion-testing
index 7833a65d5b..13bbae3877 160000
--- a/datafusion-testing
+++ b/datafusion-testing
@@ -1 +1 @@
-Subproject commit 7833a65d5b08be2ca484ea938f471cf01df54e18
+Subproject commit 13bbae38776c2bfbc1fab1be7e7220222d4284bf
diff --git a/datafusion/expr/src/logical_plan/plan.rs b/datafusion/expr/src/logical_plan/plan.rs
index c572b202f0..2f1061c438 100644
--- a/datafusion/expr/src/logical_plan/plan.rs
+++ b/datafusion/expr/src/logical_plan/plan.rs
@@ -2495,6 +2495,19 @@ pub struct Filter {
}
impl Filter {
+ /// Create a new filter operator.
+ ///
+ /// Skips the type-checking and dealiasing done in [Self::try_new].
+ /// For internal use in DataFusion only.
+ ///
+ /// **Preconditions:**
+ /// - the `predicate` expression returns a boolean value
+ /// - the `predicate` expression is not aliased
+ #[doc(hidden)]
+ pub fn new(predicate: Expr, input: Arc<LogicalPlan>) -> Self {
+ Self { predicate, input }
+ }
+
/// Create a new filter operator.
///
/// Notes: as Aliases have no effect on the output of a filter operator,
diff --git a/datafusion/optimizer/src/push_down_filter.rs b/datafusion/optimizer/src/push_down_filter.rs
index 46e129ad4b..9c2ac07ff0 100644
--- a/datafusion/optimizer/src/push_down_filter.rs
+++ b/datafusion/optimizer/src/push_down_filter.rs
@@ -30,17 +30,18 @@ use datafusion_common::tree_node::{
Transformed, TransformedResult, TreeNode, TreeNodeRecursion,
};
use datafusion_common::{
- Column, DFSchema, Result, assert_eq_or_internal_err, assert_or_internal_err,
- internal_err, plan_err, qualified_name,
+ Column, DFSchema, Result, assert_eq_or_internal_err, internal_err, plan_err,
+ qualified_name,
};
use datafusion_expr::expr::WindowFunction;
use datafusion_expr::expr_rewriter::replace_col;
-use datafusion_expr::logical_plan::{Join, JoinType, LogicalPlan, TableScan, Union};
+use datafusion_expr::logical_plan::{Join, JoinType, LogicalPlan};
use datafusion_expr::utils::{
conjunction, expr_to_columns, split_conjunction, split_conjunction_owned,
};
use datafusion_expr::{
- BinaryExpr, Expr, Filter, Operator, Projection, TableProviderFilterPushDown, and, or,
+ BinaryExpr, Distinct, Expr, Filter, Operator, Projection,
+ TableProviderFilterPushDown, and, or,
};
use crate::optimizer::ApplyOrder;
@@ -447,16 +448,13 @@ fn push_down_all_join(
let mut on_filter_join_conditions = vec![];
let (on_left_preserved, on_right_preserved) = on_lr_is_preserved(join.join_type);
-
- if !on_filter.is_empty() {
- for on in on_filter {
- if on_left_preserved && checker.is_left_only(&on) {
- left_push.push(on)
- } else if on_right_preserved && checker.is_right_only(&on) {
- right_push.push(on)
- } else {
- on_filter_join_conditions.push(on)
- }
+ for on in on_filter {
+ if on_left_preserved && checker.is_left_only(&on) {
+ left_push.push(on)
+ } else if on_right_preserved && checker.is_right_only(&on) {
+ right_push.push(on)
+ } else {
+ on_filter_join_conditions.push(on)
}
}
@@ -498,41 +496,46 @@ fn push_down_all_join(
));
}
+ // Add any new join conditions as the non join predicates
+ let join_conditions_empty = join_conditions.is_empty();
+ join_conditions.extend(on_filter_join_conditions);
+ join.filter = conjunction(join_conditions);
+
+ if join_conditions_empty && left_push.is_empty() && right_push.is_empty() {
+ // wrap the join on the filter whose predicates must be kept, if any
+ return Ok(Transformed::no(with_filters(
+ keep_predicates,
+ LogicalPlan::Join(join),
+ )));
+ }
+
if let Some(predicate) = conjunction(left_push) {
- join.left = Arc::new(LogicalPlan::Filter(Filter::try_new(predicate, join.left)?));
+ join.left = Arc::new(LogicalPlan::Filter(Filter::new(predicate, join.left)));
}
+
if let Some(predicate) = conjunction(right_push) {
- join.right =
- Arc::new(LogicalPlan::Filter(Filter::try_new(predicate, join.right)?));
+ join.right = Arc::new(LogicalPlan::Filter(Filter::new(predicate, join.right)));
}
- // Add any new join conditions as the non join predicates
- join_conditions.extend(on_filter_join_conditions);
- join.filter = conjunction(join_conditions);
-
// wrap the join on the filter whose predicates must be kept, if any
- let plan = LogicalPlan::Join(join);
- let plan = if let Some(predicate) = conjunction(keep_predicates) {
- LogicalPlan::Filter(Filter::try_new(predicate, Arc::new(plan))?)
- } else {
- plan
- };
- Ok(Transformed::yes(plan))
+ Ok(Transformed::yes(with_filters(
+ keep_predicates,
+ LogicalPlan::Join(join),
+ )))
}
fn push_down_join(
- join: Join,
- parent_predicate: Option<&Expr>,
+ mut join: Join,
+ parent_predicate: Option<Expr>,
) -> Result<Transformed<LogicalPlan>> {
// Split the parent predicate into individual conjunctive parts.
- let predicates = parent_predicate
- .map_or_else(Vec::new, |pred| split_conjunction_owned(pred.clone()));
+ let predicates = parent_predicate.map_or_else(Vec::new, split_conjunction_owned);
// Extract conjunctions from the JOIN's ON filter, if present.
let on_filters = join
.filter
- .as_ref()
- .map_or_else(Vec::new, |filter| split_conjunction_owned(filter.clone()));
+ .take()
+ .map_or_else(Vec::new, split_conjunction_owned);
// Are there any new join predicates that can be inferred from the filter expressions?
let inferred_join_predicates = with_debug_timing("infer_join_predicates", || {
@@ -773,13 +776,11 @@ impl OptimizerRule for PushDownFilter {
plan: LogicalPlan,
config: &dyn OptimizerConfig,
) -> Result<Transformed<LogicalPlan>> {
- let _ = config.options();
+ let _ = config;
if let LogicalPlan::Join(join) = plan {
return push_down_join(join, None);
};
- let plan_schema = Arc::clone(plan.schema());
-
let LogicalPlan::Filter(mut filter) = plan else {
return Ok(Transformed::no(plan));
};
@@ -812,44 +813,48 @@ impl OptimizerRule for PushDownFilter {
}
match Arc::unwrap_or_clone(filter.input) {
- LogicalPlan::Filter(child_filter) => {
- // child filters first to preserve execution order
- let new_predicates = split_conjunction_owned(child_filter.predicate)
- .into_iter()
- .chain(split_conjunction_owned(filter.predicate))
- // use IndexSet to remove duplicates while preserving predicate order
- .collect::<IndexSet<_>>();
+ LogicalPlan::Filter(mut child_filter) => {
+ // Child filters first to preserve execution order.
+ // Use IndexSet to remove duplicates while preserving predicate order.
+ let new_predicates: IndexSet<Expr> =
+ split_conjunction_owned(child_filter.predicate)
+ .into_iter()
+ .chain(split_conjunction_owned(filter.predicate))
+ .collect();
let Some(new_predicate) = conjunction(new_predicates) else {
return plan_err!("at least one expression exists");
};
- let new_filter = LogicalPlan::Filter(Filter::try_new(
- new_predicate,
- child_filter.input,
- )?);
-
- self.rewrite(new_filter, config)
+ child_filter.predicate = new_predicate;
+ self.rewrite(LogicalPlan::Filter(child_filter), config)
}
- LogicalPlan::Repartition(repartition) => {
- let new_filter =
- Filter::try_new(filter.predicate, Arc::clone(&repartition.input))
- .map(LogicalPlan::Filter)?;
- insert_below(LogicalPlan::Repartition(repartition), new_filter)
+ LogicalPlan::Repartition(mut repartition) => {
+ filter.input = repartition.input;
+ repartition.input = Arc::new(LogicalPlan::Filter(filter));
+ Ok(Transformed::yes(LogicalPlan::Repartition(repartition)))
}
LogicalPlan::Distinct(distinct) => {
- let new_filter =
- Filter::try_new(filter.predicate, Arc::clone(distinct.input()))
- .map(LogicalPlan::Filter)?;
- insert_below(LogicalPlan::Distinct(distinct), new_filter)
+ let distinct = match distinct {
+ Distinct::All(input) => {
+ filter.input = input;
+ Distinct::All(Arc::new(LogicalPlan::Filter(filter)))
+ }
+ Distinct::On(mut distinct) => {
+ filter.input = distinct.input;
+ distinct.input = Arc::new(LogicalPlan::Filter(filter));
+ Distinct::On(distinct)
+ }
+ };
+
+ Ok(Transformed::yes(LogicalPlan::Distinct(distinct)))
}
- LogicalPlan::Sort(sort) => {
- let new_filter =
- Filter::try_new(filter.predicate, Arc::clone(&sort.input))
- .map(LogicalPlan::Filter)?;
- insert_below(LogicalPlan::Sort(sort), new_filter)
+ LogicalPlan::Sort(mut sort) => {
+ filter.input = sort.input;
+ sort.input = Arc::new(LogicalPlan::Filter(filter));
+ Ok(Transformed::yes(LogicalPlan::Sort(sort)))
}
- LogicalPlan::SubqueryAlias(subquery_alias) => {
+ LogicalPlan::SubqueryAlias(mut subquery_alias) => {
let mut replace_map = HashMap::new();
for (i, (qualifier, field)) in
subquery_alias.input.schema().iter().enumerate()
@@ -861,30 +866,24 @@ impl OptimizerRule for PushDownFilter {
Expr::Column(Column::new(qualifier.cloned(), field.name())),
);
}
- let new_predicate = replace_cols_by_name(filter.predicate, &replace_map)?;
- let new_filter = LogicalPlan::Filter(Filter::try_new(
- new_predicate,
- Arc::clone(&subquery_alias.input),
- )?);
- insert_below(LogicalPlan::SubqueryAlias(subquery_alias), new_filter)
+ filter.predicate = replace_cols_by_name(filter.predicate, &replace_map)?;
+ filter.input = subquery_alias.input;
+ subquery_alias.input = Arc::new(LogicalPlan::Filter(filter));
+ Ok(Transformed::yes(LogicalPlan::SubqueryAlias(subquery_alias)))
}
LogicalPlan::Projection(projection) => {
let predicates = split_conjunction_owned(filter.predicate.clone());
- let (new_projection, keep_predicate) =
+ let (mut result, keep_predicates) =
rewrite_projection(predicates, projection)?;
- if new_projection.transformed {
- match keep_predicate {
- None => Ok(new_projection),
- Some(keep_predicate) => new_projection.map_data(|child_plan| {
- Filter::try_new(keep_predicate, Arc::new(child_plan))
- .map(LogicalPlan::Filter)
- }),
- }
+ if result.transformed {
+ result.data = with_filters(keep_predicates, result.data)
} else {
- filter.input = Arc::new(new_projection.data);
- Ok(Transformed::no(LogicalPlan::Filter(filter)))
+ filter.input = Arc::new(result.data);
+ result.data = LogicalPlan::Filter(filter)
}
+
+ Ok(result)
}
LogicalPlan::Unnest(mut unnest) => {
let predicates = split_conjunction_owned(filter.predicate.clone());
@@ -895,11 +894,10 @@ impl OptimizerRule for PushDownFilter {
for idx in &unnest.struct_type_columns {
let (sub_qualifier, field) =
unnest.input.schema().qualified_field(*idx);
- let field_name = field.name().clone();
-
if let DataType::Struct(children) = field.data_type() {
+ let field_name = field.name();
for child in children {
- let child_name = child.name().clone();
+ let child_name = child.name();
unnest_struct_columns.push(Column::new(
sub_qualifier.cloned(),
format!("{field_name}.{child_name}"),
@@ -942,29 +940,21 @@ impl OptimizerRule for PushDownFilter {
// Filter
// Unnest Input (Projection)
- let unnest_input = std::mem::take(&mut unnest.input);
-
- let filter_with_unnest_input = LogicalPlan::Filter(Filter::try_new(
- conjunction(non_unnest_predicates).unwrap(), // Safe to unwrap since non_unnest_predicates is not empty.
- unnest_input,
- )?);
-
+ // Safe to unwrap since non_unnest_predicates is not empty.
+ filter.predicate = conjunction(non_unnest_predicates).unwrap();
+ filter.input = unnest.input;
// Directly assign new filter plan as the new unnest's input.
// The new filter plan will go through another rewrite pass since the rule itself
// is applied recursively to all the child from top to down
- let unnest_plan =
- insert_below(LogicalPlan::Unnest(unnest), filter_with_unnest_input)?;
-
- match conjunction(unnest_predicates) {
- None => Ok(unnest_plan),
- Some(predicate) => Ok(Transformed::yes(LogicalPlan::Filter(
- Filter::try_new(predicate, Arc::new(unnest_plan.data))?,
- ))),
- }
+ unnest.input = Arc::new(LogicalPlan::Filter(filter));
+ Ok(Transformed::yes(with_filters(
+ unnest_predicates,
+ LogicalPlan::Unnest(unnest),
+ )))
}
- LogicalPlan::Union(ref union) => {
+ LogicalPlan::Union(mut union) => {
let mut inputs = Vec::with_capacity(union.inputs.len());
- for input in &union.inputs {
+ for input in union.inputs {
let mut replace_map = HashMap::new();
for (i, (qualifier, field)) in input.schema().iter().enumerate() {
let (union_qualifier, union_field) =
@@ -977,72 +967,51 @@ impl OptimizerRule for PushDownFilter {
let push_predicate =
replace_cols_by_name(filter.predicate.clone(), &replace_map)?;
- inputs.push(Arc::new(LogicalPlan::Filter(Filter::try_new(
+ inputs.push(Arc::new(LogicalPlan::Filter(Filter::new(
push_predicate,
- Arc::clone(input),
- )?)))
+ input,
+ ))))
}
- Ok(Transformed::yes(LogicalPlan::Union(Union {
- inputs,
- schema: Arc::clone(&plan_schema),
- })))
+
+ union.inputs = inputs;
+ Ok(Transformed::yes(LogicalPlan::Union(union)))
}
- LogicalPlan::Aggregate(agg) => {
+ LogicalPlan::Aggregate(mut agg) => {
// We can push down Predicate which in groupby_expr.
- let group_expr_columns = agg
- .group_expr
- .iter()
- .map(|e| {
- let (relation, name) = e.qualified_name();
- Column::new(relation, name)
- })
- .collect::<HashSet<_>>();
+ let group_expr_columns = expr_columns(&agg.group_expr);
- let predicates = split_conjunction_owned(filter.predicate);
+ // As for plan Filter: Column(a+b) > 0 -- Agg: groupby:[Column(a)+Column(b)]
+ // After push, we need to replace `a+b` with Column(a)+Column(b)
+ // So we need create a replace_map, add {`a+b` --> Expr(Column(a)+Column(b))}
+ let mut replace_map = HashMap::new();
+ for expr in &agg.group_expr {
+ replace_map.insert(expr.schema_name().to_string(), unalias(expr));
+ }
+ let predicates = split_conjunction_owned(filter.predicate);
let mut keep_predicates = vec![];
let mut push_predicates = vec![];
for expr in predicates {
let cols = expr.column_refs();
if cols.iter().all(|c| group_expr_columns.contains(c)) {
- push_predicates.push(expr);
+ push_predicates.push(replace_cols_by_name(expr, &replace_map)?);
} else {
keep_predicates.push(expr);
}
}
- // As for plan Filter: Column(a+b) > 0 -- Agg: groupby:[Column(a)+Column(b)]
- // After push, we need to replace `a+b` with Column(a)+Column(b)
- // So we need create a replace_map, add {`a+b` --> Expr(Column(a)+Column(b))}
- let mut replace_map = HashMap::new();
- for expr in &agg.group_expr {
- replace_map.insert(expr.schema_name().to_string(), expr.clone());
- }
- let replaced_push_predicates = push_predicates
- .into_iter()
- .map(|expr| replace_cols_by_name(expr, &replace_map))
- .collect::<Result<Vec<_>>>()?;
-
- let agg_input = Arc::clone(&agg.input);
- Transformed::yes(LogicalPlan::Aggregate(agg))
- .transform_data(|new_plan| {
- // If we have a filter to push, we push it down to the input of the aggregate
- if let Some(predicate) = conjunction(replaced_push_predicates) {
- let new_filter = make_filter(predicate, agg_input)?;
- insert_below(new_plan, new_filter)
- } else {
- Ok(Transformed::no(new_plan))
- }
- })?
- .map_data(|child_plan| {
- // if there are any remaining predicates we can't push, add them
- // back as a filter
- if let Some(predicate) = conjunction(keep_predicates) {
- make_filter(predicate, Arc::new(child_plan))
- } else {
- Ok(child_plan)
- }
- })
+ // If we have a filter to push, we push it down to the input of the aggregate
+ let result = if let Some(predicate) = conjunction(push_predicates) {
+ filter.predicate = predicate;
+ filter.input = agg.input;
+ agg.input = Arc::new(LogicalPlan::Filter(filter));
+ Transformed::yes(LogicalPlan::Aggregate(agg))
+ } else {
+ Transformed::no(LogicalPlan::Aggregate(agg))
+ };
+
+ // If there are any remaining predicates we can't push, add them back as a filter
+ result.map_data(|plan| Ok(with_filters(keep_predicates, plan)))
}
// Tries to push filters based on the partition key(s) of the window function(s) used.
// Example:
@@ -1054,22 +1023,16 @@ impl OptimizerRule for PushDownFilter {
// Filter: (b > 1) and (c > 1)
// Window: func() PARTITION BY [a] ...
// Filter: (a > 1)
- LogicalPlan::Window(window) => {
+ LogicalPlan::Window(mut window) => {
// Retrieve the set of potential partition keys where we can push filters by.
// Unlike aggregations, where there is only one statement per SELECT, there can be
// multiple window functions, each with potentially different partition keys.
// Therefore, we need to ensure that any potential partition key returned is used in
// ALL window functions. Otherwise, filters cannot be pushed by through that column.
- let extract_partition_keys = |func: &WindowFunction| {
- func.params
- .partition_by
- .iter()
- .map(|c| {
- let (relation, name) = c.qualified_name();
- Column::new(relation, name)
- })
- .collect::<HashSet<_>>()
- };
+ fn extract_partition_keys(func: &WindowFunction) -> HashSet<Column> {
+ expr_columns(&func.params.partition_by)
+ }
+
let potential_partition_keys = window
.window_expr
.iter()
@@ -1119,31 +1082,22 @@ impl OptimizerRule for PushDownFilter {
// place, so we can use `push_predicates` directly. This is consistent with other
// optimizers, such as the one used by Postgres.
- let window_input = Arc::clone(&window.input);
- Transformed::yes(LogicalPlan::Window(window))
- .transform_data(|new_plan| {
- // If we have a filter to push, we push it down to the input of the window
- if let Some(predicate) = conjunction(push_predicates) {
- let new_filter = make_filter(predicate, window_input)?;
- insert_below(new_plan, new_filter)
- } else {
- Ok(Transformed::no(new_plan))
- }
- })?
- .map_data(|child_plan| {
- // if there are any remaining predicates we can't push, add them
- // back as a filter
- if let Some(predicate) = conjunction(keep_predicates) {
- make_filter(predicate, Arc::new(child_plan))
- } else {
- Ok(child_plan)
- }
- })
+ // If we have a filter to push, we push it down to the input of the aggregate
+ let result = if let Some(predicate) = conjunction(push_predicates) {
+ filter.predicate = predicate;
+ filter.input = window.input;
+ window.input = Arc::new(LogicalPlan::Filter(filter));
+ Transformed::yes(LogicalPlan::Window(window))
+ } else {
+ Transformed::no(LogicalPlan::Window(window))
+ };
+
+ // If there are any remaining predicates we can't push, add them back as a filter
+ result.map_data(|plan| Ok(with_filters(keep_predicates, plan)))
}
- LogicalPlan::Join(join) => push_down_join(join, Some(&filter.predicate)),
- LogicalPlan::TableScan(scan) => {
+ LogicalPlan::Join(join) => push_down_join(join, Some(filter.predicate)),
+ LogicalPlan::TableScan(mut scan) => {
let filter_predicates = split_conjunction(&filter.predicate);
-
// Filters containing scalar subqueries cannot be pushed to
// providers because the subquery result is not available
// until execution time.
@@ -1169,13 +1123,21 @@ impl OptimizerRule for PushDownFilter {
non_volatile_filters.len()
);
+ if supported_filters
+ .iter()
+ .all(|res| res == &TableProviderFilterPushDown::Unsupported)
+ {
+ filter.input = Arc::new(LogicalPlan::TableScan(scan));
+ return Ok(Transformed::no(LogicalPlan::Filter(filter)));
+ }
+
// Compose scan filters from non-volatile filters of `Exact` or `Inexact` pushdown type
- let zip = non_volatile_filters.into_iter().zip(supported_filters);
+ let zip = non_volatile_filters.iter().zip(supported_filters.iter());
let new_scan_filters = zip
.clone()
- .filter(|(_, res)| res != &TableProviderFilterPushDown::Unsupported)
- .map(|(pred, _)| pred);
+ .filter(|(_, res)| *res != &TableProviderFilterPushDown::Unsupported)
+ .map(|(&pred, _)| pred);
// Add new scan filters
let new_scan_filters: Vec<Expr> = scan
@@ -1186,28 +1148,31 @@ impl OptimizerRule for PushDownFilter {
.cloned()
.collect();
+ if supported_filters
+ .iter()
+ .all(|res| res == &TableProviderFilterPushDown::Inexact)
+ && scan.filters == new_scan_filters
+ {
+ filter.input = Arc::new(LogicalPlan::TableScan(scan));
+ return Ok(Transformed::no(LogicalPlan::Filter(filter)));
+ } else {
+ scan.filters = new_scan_filters;
+ }
+
// Compose predicates to be of `Unsupported` or `Inexact` pushdown type,
// and also include volatile and subquery-containing filters
let new_predicate: Vec<Expr> = zip
- .filter(|(_, res)| res != &TableProviderFilterPushDown::Exact)
- .map(|(pred, _)| pred)
+ .filter(|(_, res)| *res != &TableProviderFilterPushDown::Exact)
+ .map(|(&pred, _)| pred)
.chain(volatile_filters)
.chain(subquery_filters)
.cloned()
.collect();
- let new_scan = LogicalPlan::TableScan(TableScan {
- filters: new_scan_filters,
- ..scan
- });
-
- Transformed::yes(new_scan).transform_data(|new_scan| {
- if let Some(predicate) = conjunction(new_predicate) {
- make_filter(predicate, Arc::new(new_scan)).map(Transformed::yes)
- } else {
- Ok(Transformed::no(new_scan))
- }
- })
+ Ok(Transformed::yes(with_filters(
+ new_predicate,
+ LogicalPlan::TableScan(scan),
+ )))
}
LogicalPlan::Extension(extension_plan) => {
// This check prevents the Filter from being removed when the extension node has no children,
@@ -1222,17 +1187,16 @@ impl OptimizerRule for PushDownFilter {
// determine if we can push any predicates down past the extension node
// each element is true for push, false to keep
- let predicate_push_or_keep = split_conjunction(&filter.predicate)
- .iter()
- .map(|expr| {
- let cols = expr.column_refs();
- if cols.iter().any(|c| prevent_cols.contains(&c.name)) {
- Ok(false) // No push (keep)
- } else {
- Ok(true) // push
- }
- })
- .collect::<Result<Vec<_>>>()?;
+ let predicate_push_or_keep: Vec<bool> =
+ split_conjunction(&filter.predicate)
+ .iter()
+ .map(|expr| {
+ !expr
+ .column_refs()
+ .iter()
+ .any(|c| prevent_cols.contains(&c.name))
+ })
+ .collect();
// all predicates are kept, no changes needed
if predicate_push_or_keep.iter().all(|&x| !x) {
@@ -1254,33 +1218,25 @@ impl OptimizerRule for PushDownFilter {
}
}
- let new_children = match conjunction(push_predicates) {
- Some(predicate) => extension_plan
- .node
- .inputs()
- .into_iter()
- .map(|child| {
- Ok(LogicalPlan::Filter(Filter::try_new(
- predicate.clone(),
- Arc::new(child.clone()),
- )?))
- })
- .collect::<Result<Vec<_>>>()?,
- None => extension_plan.node.inputs().into_iter().cloned().collect(),
- };
+ // Unwrap - push_predicates is not empty, predicate_push_or_keep checked.
+ let predicate = conjunction(push_predicates).unwrap();
+ let new_children = extension_plan
+ .node
+ .inputs()
+ .into_iter()
+ .map(|child| {
+ LogicalPlan::Filter(Filter::new(
+ predicate.clone(),
+ Arc::new(child.clone()),
+ ))
+ })
+ .collect();
+
// extension with new inputs.
- let child_plan = LogicalPlan::Extension(extension_plan);
- let new_extension =
- child_plan.with_new_exprs(child_plan.expressions(), new_children)?;
-
- let new_plan = match conjunction(keep_predicates) {
- Some(predicate) => LogicalPlan::Filter(Filter::try_new(
- predicate,
- Arc::new(new_extension),
- )?),
- None => new_extension,
- };
- Ok(Transformed::yes(new_plan))
+ let extension = LogicalPlan::Extension(extension_plan);
+ let new_plan =
+ extension.with_new_exprs(extension.expressions(), new_children)?;
+ Ok(Transformed::yes(with_filters(keep_predicates, new_plan)))
}
child => {
filter.input = Arc::new(child);
@@ -1320,22 +1276,19 @@ impl OptimizerRule for PushDownFilter {
fn rewrite_projection(
predicates: Vec<Expr>,
mut projection: Projection,
-) -> Result<(Transformed<LogicalPlan>, Option<Expr>)> {
+) -> Result<(Transformed<LogicalPlan>, Vec<Expr>)> {
// Partition projection expressions into non-pushable vs pushable.
// Non-pushable expressions are volatile (must not be duplicated) or
// MoveTowardsLeafNodes (cheap expressions like get_field where re-inlining
// into a filter causes optimizer instability — ExtractLeafExpressions will
// undo the push-down, creating an infinite loop that runs until the
// iteration limit is hit).
- let (non_pushable_map, pushable_map): (HashMap<_, _>, HashMap<_, _>) = projection
+ let (non_pushable_map, pushable_map) = projection
.schema
.iter()
.zip(projection.expr.iter())
.map(|((qualifier, field), expr)| {
- // strip alias, as they should not be part of filters
- let expr = expr.clone().unalias();
-
- (qualified_name(qualifier, field.name()), expr)
+ (qualified_name(qualifier, field.name()), unalias(expr))
})
.partition(|(_, value)| {
value.is_volatile()
@@ -1352,67 +1305,30 @@ fn rewrite_projection(
}
}
- match conjunction(push_predicates) {
- Some(expr) => {
- // re-write all filters based on this projection
- // E.g. in `Filter: b\n Projection: a > 1 as b`, we can swap them, but the filter must be "a > 1"
- let new_filter = LogicalPlan::Filter(Filter::try_new(
- replace_cols_by_name(expr, &pushable_map)?,
- std::mem::take(&mut projection.input),
- )?);
-
- projection.input = Arc::new(new_filter);
-
- Ok((
- Transformed::yes(LogicalPlan::Projection(projection)),
- conjunction(keep_predicates),
- ))
- }
- None => Ok((
- Transformed::no(LogicalPlan::Projection(projection)),
- conjunction(keep_predicates),
- )),
- }
+ let projection = if let Some(expr) = conjunction(push_predicates) {
+ // re-write all filters based on this projection
+ // E.g. in `Filter: b\n Projection: a > 1 as b`, we can swap them, but the filter must be "a > 1"
+ projection.input = Arc::new(LogicalPlan::Filter(Filter::new(
+ replace_cols_by_name(expr, &pushable_map)?,
+ projection.input,
+ )));
+
+ Transformed::yes(LogicalPlan::Projection(projection))
+ } else {
+ Transformed::no(LogicalPlan::Projection(projection))
+ };
+
+ Ok((projection, keep_predicates))
}
/// Creates a new LogicalPlan::Filter node.
+///
+/// Deprecated: use [`Filter::try_new`] directly.
+#[deprecated]
pub fn make_filter(predicate: Expr, input: Arc<LogicalPlan>) -> Result<LogicalPlan> {
Filter::try_new(predicate, input).map(LogicalPlan::Filter)
}
-/// Replace the existing child of the single input node with `new_child`.
-///
-/// Starting:
-/// ```text
-/// plan
-/// child
-/// ```
-///
-/// Ending:
-/// ```text
-/// plan
-/// new_child
-/// ```
-fn insert_below(
- plan: LogicalPlan,
- new_child: LogicalPlan,
-) -> Result<Transformed<LogicalPlan>> {
- let mut new_child = Some(new_child);
- let transformed_plan = plan.map_children(|_child| {
- if let Some(new_child) = new_child.take() {
- Ok(Transformed::yes(new_child))
- } else {
- // already took the new child
- internal_err!("node had more than one input")
- }
- })?;
-
- // make sure we did the actual replacement
- assert_or_internal_err!(new_child.is_none(), "node had no inputs");
-
- Ok(transformed_plan)
-}
-
impl PushDownFilter {
#[expect(missing_docs)]
pub fn new() -> Self {
@@ -1439,41 +1355,64 @@ where
/// replaces columns by its name on the projection.
pub fn replace_cols_by_name(
e: Expr,
- replace_map: &HashMap<String, Expr>,
+ replace_map: &HashMap<String, impl AsRef<Expr>>,
) -> Result<Expr> {
e.transform_up(|expr| {
- Ok(if let Expr::Column(c) = &expr {
- match replace_map.get(&c.flat_name()) {
- Some(new_c) => Transformed::yes(new_c.clone()),
- None => Transformed::no(expr),
- }
+ if let Expr::Column(c) = &expr
+ && let Some(new_expr) = replace_map.get(&c.flat_name())
+ {
+ Ok(Transformed::yes(new_expr.as_ref().clone()))
} else {
- Transformed::no(expr)
- })
+ Ok(Transformed::no(expr))
+ }
})
.data()
}
+/// Unalias expression reference.
+fn unalias(expr: &Expr) -> &Expr {
+ if let Expr::Alias(alias) = expr {
+ unalias(&alias.expr)
+ } else {
+ expr
+ }
+}
+
/// check whether the expression uses the columns in `check_map`.
-fn contain(e: &Expr, check_map: &HashMap<String, Expr>) -> bool {
+fn contain<T>(e: &Expr, check_map: &HashMap<String, T>) -> bool {
let mut is_contain = false;
e.apply(|expr| {
- Ok(if let Expr::Column(c) = &expr {
- match check_map.get(&c.flat_name()) {
- Some(_) => {
- is_contain = true;
- TreeNodeRecursion::Stop
- }
- None => TreeNodeRecursion::Continue,
- }
+ if let Expr::Column(c) = &expr
+ && check_map.contains_key(&c.flat_name())
+ {
+ is_contain = true;
+ Ok(TreeNodeRecursion::Stop)
} else {
- TreeNodeRecursion::Continue
- })
+ Ok(TreeNodeRecursion::Continue)
+ }
})
.unwrap();
is_contain
}
+fn with_filters(predicates: Vec<Expr>, plan: LogicalPlan) -> LogicalPlan {
+ if let Some(predicate) = conjunction(predicates) {
+ LogicalPlan::Filter(Filter::new(predicate, Arc::new(plan)))
+ } else {
+ plan
+ }
+}
+
+fn expr_columns(exprs: &[Expr]) -> HashSet<Column> {
+ exprs
+ .iter()
+ .map(|expr| {
+ let (relation, name) = expr.qualified_name();
+ Column::new(relation, name)
+ })
+ .collect()
+}
+
#[cfg(test)]
mod tests {
use std::cmp::Ordering;
@@ -1487,9 +1426,9 @@ mod tests {
use datafusion_expr::logical_plan::table_scan;
use datafusion_expr::{
ColumnarValue, ExprFunctionExt, Extension, LogicalPlanBuilder,
- ScalarFunctionArgs, ScalarUDF, ScalarUDFImpl, Signature, TableSource, TableType,
- UserDefinedLogicalNodeCore, Volatility, WindowFunctionDefinition, col, in_list,
- in_subquery, lit,
+ ScalarFunctionArgs, ScalarUDF, ScalarUDFImpl, Signature, TableScan, TableSource,
+ TableType, UserDefinedLogicalNodeCore, Volatility, WindowFunctionDefinition, col,
+ in_list, in_subquery, lit,
};
use crate::OptimizerContext;
@@ -1536,6 +1475,17 @@ mod tests {
}};
}
+ /// For testing that we don't return [Transformed::yes] when not necessary,
+ /// as it triggers rebuilding parent plan nodes.
+ macro_rules! assert_plan_not_transformed {
+ ($plan:expr) => {{
+ let transformed = PushDownFilter::new()
+ .rewrite($plan, &OptimizerContext::new())
+ .expect("failed to optimize plan");
+ assert!(!transformed.transformed);
+ }};
+ }
+
#[test]
fn filter_before_projection() -> Result<()> {
let table_scan = test_table_scan()?;
@@ -1684,6 +1634,8 @@ mod tests {
.aggregate(vec![col("a")], vec![sum(col("b")).alias("b")])?
.filter(col("b").gt(lit(10i64)))?
.build()?;
+ assert_plan_not_transformed!(plan.clone());
+
// filter of aggregate is after aggregation since they are non-commutative
assert_optimized_plan_equal!(
plan,
@@ -1876,6 +1828,7 @@ mod tests {
.window(vec![window])?
.filter(col("c").gt(lit(10i64)))?
.build()?;
+ assert_plan_not_transformed!(plan.clone());
assert_optimized_plan_equal!(
plan,
@@ -3101,6 +3054,7 @@ mod tests {
Some(filter),
)?
.build()?;
+ assert_plan_not_transformed!(plan.clone());
// not part of the test, just good to know:
assert_snapshot!(plan,
@@ -3207,15 +3161,16 @@ mod tests {
let plan =
table_scan_with_pushdown_provider(TableProviderFilterPushDown::Inexact)?;
- let optimized_plan = PushDownFilter::new()
+ let optimized = PushDownFilter::new()
.rewrite(plan, &OptimizerContext::new())
- .expect("failed to optimize plan")
- .data;
+ .expect("failed to optimize plan");
+ assert!(optimized.transformed);
+ assert_plan_not_transformed!(optimized.data.clone());
// Optimizing the same plan multiple times should produce the same plan
// each time.
assert_optimized_plan_equal!(
- optimized_plan,
+ optimized.data,
@r"
Filter: a = Int64(1)
TableScan: test, partial_filters=[a = Int64(1)]
@@ -3227,6 +3182,7 @@ mod tests {
fn filter_with_table_provider_unsupported() -> Result<()> {
let plan =
table_scan_with_pushdown_provider(TableProviderFilterPushDown::Unsupported)?;
+ assert_plan_not_transformed!(plan.clone());
assert_optimized_plan_equal!(
plan,
@@ -4217,7 +4173,7 @@ mod tests {
plan,
@r"
Projection: a, b
- Filter: t.a > Int32(5) AND t.b > Int32(10) AND TestScalarUDF() > Float64(0.1)
+ Filter: TestScalarUDF() > Float64(0.1) AND t.a > Int32(5) AND t.b > Int32(10)
TableScan: test
"
)
diff --git a/datafusion/optimizer/src/simplify_expressions/simplify_predicates.rs b/datafusion/optimizer/src/simplify_expressions/simplify_predicates.rs
index e811ce7313..356f2711b7 100644
--- a/datafusion/optimizer/src/simplify_expressions/simplify_predicates.rs
+++ b/datafusion/optimizer/src/simplify_expressions/simplify_predicates.rs
@@ -63,12 +63,14 @@ pub fn simplify_predicates(predicates: Vec<Expr>) -> Result<Vec<Expr>> {
| Operator::Eq,
right,
}) => {
- let left_col = extract_column_from_expr(left);
- let right_col = extract_column_from_expr(right);
- if let (Some(col), Some(_)) = (&left_col, right.as_literal()) {
- column_predicates.entry(col.clone()).or_default().push(pred);
- } else if let (Some(_), Some(col)) = (left.as_literal(), &right_col) {
- column_predicates.entry(col.clone()).or_default().push(pred);
+ if let (Some(col), Some(_)) =
+ (extract_column_from_expr(left), right.as_literal())
+ {
+ column_predicates.entry(col).or_default().push(pred);
+ } else if let (Some(_), Some(col)) =
+ (left.as_literal(), extract_column_from_expr(right))
+ {
+ column_predicates.entry(col).or_default().push(pred);
} else {
other_predicates.push(pred);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org