You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by al...@apache.org on 2025/07/11 20:30:50 UTC
(datafusion) branch main updated: Refactor filter pushdown APIs to enable joins to pass through filters (#16732)
This is an automated email from the ASF dual-hosted git repository.
alamb 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 4bc66c80d5 Refactor filter pushdown APIs to enable joins to pass through filters (#16732)
4bc66c80d5 is described below
commit 4bc66c80d560581f527dee5774fb4f0479786d3e
Author: Adrian Garcia Badaracco <17...@users.noreply.github.com>
AuthorDate: Fri Jul 11 15:30:45 2025 -0500
Refactor filter pushdown APIs to enable joins to pass through filters (#16732)
* refactor filter pushdown to remove assumptions blocking hashjoinexec pushdown
* fix assertion
* fix
* fix any/all
* remove hashjoinexec impl
* lint
* resolve merge
* fix imports
* fix merge
* fix typo
* lint
* Improve some documentation in filter pushdown (#32)
* refactor to a struct with a field instead of enum
* better docs, rename methods
* fix cargo doc
---------
Co-authored-by: Andrew Lamb <an...@nerdnetworks.org>
---
.../physical_optimizer/filter_pushdown/util.rs | 43 +++--
datafusion/datasource-parquet/src/source.rs | 40 ++---
datafusion/datasource/src/file.rs | 11 +-
datafusion/datasource/src/source.rs | 35 ++--
.../physical-optimizer/src/filter_pushdown.rs | 166 +++++++++--------
datafusion/physical-plan/src/coalesce_batches.rs | 4 +-
datafusion/physical-plan/src/execution_plan.rs | 21 ++-
datafusion/physical-plan/src/filter.rs | 50 +++---
datafusion/physical-plan/src/filter_pushdown.rs | 200 +++++++++++++++++----
datafusion/physical-plan/src/repartition/mod.rs | 4 +-
.../src/engines/postgres_engine/mod.rs | 2 +-
11 files changed, 346 insertions(+), 230 deletions(-)
diff --git a/datafusion/core/tests/physical_optimizer/filter_pushdown/util.rs b/datafusion/core/tests/physical_optimizer/filter_pushdown/util.rs
index d4318235ba..ea12bea1cf 100644
--- a/datafusion/core/tests/physical_optimizer/filter_pushdown/util.rs
+++ b/datafusion/core/tests/physical_optimizer/filter_pushdown/util.rs
@@ -29,13 +29,13 @@ use datafusion_datasource::{
use datafusion_physical_expr::conjunction;
use datafusion_physical_expr_common::physical_expr::fmt_sql;
use datafusion_physical_optimizer::PhysicalOptimizerRule;
-use datafusion_physical_plan::filter_pushdown::FilterPushdownPhase;
+use datafusion_physical_plan::filter_pushdown::{FilterPushdownPhase, PushedDown};
use datafusion_physical_plan::{
displayable,
filter::FilterExec,
filter_pushdown::{
ChildFilterDescription, ChildPushdownResult, FilterDescription,
- FilterPushdownPropagation, PredicateSupport,
+ FilterPushdownPropagation,
},
metrics::ExecutionPlanMetricsSet,
DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties,
@@ -227,19 +227,13 @@ impl FileSource for TestSource {
predicate: Some(conjunction(filters.clone())),
..self.clone()
});
- Ok(FilterPushdownPropagation {
- filters: filters
- .into_iter()
- .map(PredicateSupport::Supported)
- .collect(),
- updated_node: Some(new_node),
- })
+ Ok(FilterPushdownPropagation::with_parent_pushdown_result(
+ vec![PushedDown::Yes; filters.len()],
+ )
+ .with_updated_node(new_node))
} else {
- Ok(FilterPushdownPropagation::with_filters(
- filters
- .into_iter()
- .map(PredicateSupport::Unsupported)
- .collect(),
+ Ok(FilterPushdownPropagation::with_parent_pushdown_result(
+ vec![PushedDown::No; filters.len()],
))
}
}
@@ -547,26 +541,29 @@ impl ExecutionPlan for TestNode {
assert_eq!(self_pushdown_result.len(), 1);
let self_pushdown_result: Vec<_> = self_pushdown_result.into_iter().collect();
- match &self_pushdown_result[0] {
- PredicateSupport::Unsupported(filter) => {
+ let first_pushdown_result = self_pushdown_result[0].clone();
+
+ match &first_pushdown_result.discriminant {
+ PushedDown::No => {
// We have a filter to push down
- let new_child =
- FilterExec::try_new(Arc::clone(filter), Arc::clone(&self.input))?;
+ let new_child = FilterExec::try_new(
+ Arc::clone(&first_pushdown_result.predicate),
+ Arc::clone(&self.input),
+ )?;
let new_self =
TestNode::new(false, Arc::new(new_child), self.predicate.clone());
let mut res =
- FilterPushdownPropagation::transparent(child_pushdown_result);
+ FilterPushdownPropagation::if_all(child_pushdown_result);
res.updated_node = Some(Arc::new(new_self) as Arc<dyn ExecutionPlan>);
Ok(res)
}
- PredicateSupport::Supported(_) => {
- let res =
- FilterPushdownPropagation::transparent(child_pushdown_result);
+ PushedDown::Yes => {
+ let res = FilterPushdownPropagation::if_all(child_pushdown_result);
Ok(res)
}
}
} else {
- let res = FilterPushdownPropagation::transparent(child_pushdown_result);
+ let res = FilterPushdownPropagation::if_all(child_pushdown_result);
Ok(res)
}
}
diff --git a/datafusion/datasource-parquet/src/source.rs b/datafusion/datasource-parquet/src/source.rs
index 8ca36e7cd3..0c73661a6d 100644
--- a/datafusion/datasource-parquet/src/source.rs
+++ b/datafusion/datasource-parquet/src/source.rs
@@ -41,8 +41,9 @@ use datafusion_datasource::file_scan_config::FileScanConfig;
use datafusion_physical_expr::conjunction;
use datafusion_physical_expr_common::physical_expr::fmt_sql;
use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
+use datafusion_physical_plan::filter_pushdown::PushedDown;
use datafusion_physical_plan::filter_pushdown::{
- FilterPushdownPropagation, PredicateSupport,
+ FilterPushdownPropagation, PushedDownPredicate,
};
use datafusion_physical_plan::metrics::Count;
use datafusion_physical_plan::metrics::ExecutionPlanMetricsSet;
@@ -622,11 +623,8 @@ impl FileSource for ParquetSource {
config: &ConfigOptions,
) -> datafusion_common::Result<FilterPushdownPropagation<Arc<dyn FileSource>>> {
let Some(file_schema) = self.file_schema.clone() else {
- return Ok(FilterPushdownPropagation::with_filters(
- filters
- .into_iter()
- .map(PredicateSupport::Unsupported)
- .collect(),
+ return Ok(FilterPushdownPropagation::with_parent_pushdown_result(
+ vec![PushedDown::No; filters.len()],
));
};
// Determine if based on configs we should push filters down.
@@ -641,29 +639,31 @@ impl FileSource for ParquetSource {
let pushdown_filters = table_pushdown_enabled || config_pushdown_enabled;
let mut source = self.clone();
- let filters: Vec<PredicateSupport> = filters
+ let filters: Vec<PushedDownPredicate> = filters
.into_iter()
.map(|filter| {
if can_expr_be_pushed_down_with_schemas(&filter, &file_schema) {
- PredicateSupport::Supported(filter)
+ PushedDownPredicate::supported(filter)
} else {
- PredicateSupport::Unsupported(filter)
+ PushedDownPredicate::unsupported(filter)
}
})
.collect();
if filters
.iter()
- .all(|f| matches!(f, PredicateSupport::Unsupported(_)))
+ .all(|f| matches!(f.discriminant, PushedDown::No))
{
// No filters can be pushed down, so we can just return the remaining filters
// and avoid replacing the source in the physical plan.
- return Ok(FilterPushdownPropagation::with_filters(filters));
+ return Ok(FilterPushdownPropagation::with_parent_pushdown_result(
+ vec![PushedDown::No; filters.len()],
+ ));
}
let allowed_filters = filters
.iter()
- .filter_map(|f| match f {
- PredicateSupport::Supported(expr) => Some(Arc::clone(expr)),
- PredicateSupport::Unsupported(_) => None,
+ .filter_map(|f| match f.discriminant {
+ PushedDown::Yes => Some(Arc::clone(&f.predicate)),
+ PushedDown::No => None,
})
.collect_vec();
let predicate = match source.predicate {
@@ -678,15 +678,15 @@ impl FileSource for ParquetSource {
// If pushdown_filters is false we tell our parents that they still have to handle the filters,
// even if we updated the predicate to include the filters (they will only be used for stats pruning).
if !pushdown_filters {
- return Ok(FilterPushdownPropagation::with_filters(
- filters
- .into_iter()
- .map(|f| PredicateSupport::Unsupported(f.into_inner()))
- .collect_vec(),
+ return Ok(FilterPushdownPropagation::with_parent_pushdown_result(
+ vec![PushedDown::No; filters.len()],
)
.with_updated_node(source));
}
- Ok(FilterPushdownPropagation::with_filters(filters).with_updated_node(source))
+ Ok(FilterPushdownPropagation::with_parent_pushdown_result(
+ filters.iter().map(|f| f.discriminant).collect(),
+ )
+ .with_updated_node(source))
}
fn with_schema_adapter_factory(
diff --git a/datafusion/datasource/src/file.rs b/datafusion/datasource/src/file.rs
index a95c07cc31..29fa38a8ee 100644
--- a/datafusion/datasource/src/file.rs
+++ b/datafusion/datasource/src/file.rs
@@ -30,9 +30,7 @@ use arrow::datatypes::SchemaRef;
use datafusion_common::config::ConfigOptions;
use datafusion_common::{not_impl_err, Result, Statistics};
use datafusion_physical_expr::{LexOrdering, PhysicalExpr};
-use datafusion_physical_plan::filter_pushdown::{
- FilterPushdownPropagation, PredicateSupport,
-};
+use datafusion_physical_plan::filter_pushdown::{FilterPushdownPropagation, PushedDown};
use datafusion_physical_plan::metrics::ExecutionPlanMetricsSet;
use datafusion_physical_plan::DisplayFormatType;
@@ -122,11 +120,8 @@ pub trait FileSource: Send + Sync {
filters: Vec<Arc<dyn PhysicalExpr>>,
_config: &ConfigOptions,
) -> Result<FilterPushdownPropagation<Arc<dyn FileSource>>> {
- Ok(FilterPushdownPropagation::with_filters(
- filters
- .into_iter()
- .map(PredicateSupport::Unsupported)
- .collect(),
+ Ok(FilterPushdownPropagation::with_parent_pushdown_result(
+ vec![PushedDown::No; filters.len()],
))
}
diff --git a/datafusion/datasource/src/source.rs b/datafusion/datasource/src/source.rs
index 92c2555638..a002d4992c 100644
--- a/datafusion/datasource/src/source.rs
+++ b/datafusion/datasource/src/source.rs
@@ -42,7 +42,7 @@ use datafusion_physical_expr::{
use datafusion_physical_expr_common::sort_expr::LexOrdering;
use datafusion_physical_plan::filter::collect_columns_from_predicate;
use datafusion_physical_plan::filter_pushdown::{
- ChildPushdownResult, FilterPushdownPhase, FilterPushdownPropagation, PredicateSupport,
+ ChildPushdownResult, FilterPushdownPhase, FilterPushdownPropagation, PushedDown,
};
/// A source of data, typically a list of files or memory
@@ -172,11 +172,8 @@ pub trait DataSource: Send + Sync + Debug {
filters: Vec<Arc<dyn PhysicalExpr>>,
_config: &ConfigOptions,
) -> Result<FilterPushdownPropagation<Arc<dyn DataSource>>> {
- Ok(FilterPushdownPropagation::with_filters(
- filters
- .into_iter()
- .map(PredicateSupport::Unsupported)
- .collect(),
+ Ok(FilterPushdownPropagation::with_parent_pushdown_result(
+ vec![PushedDown::No; filters.len()],
))
}
}
@@ -324,17 +321,14 @@ impl ExecutionPlan for DataSourceExec {
config: &ConfigOptions,
) -> Result<FilterPushdownPropagation<Arc<dyn ExecutionPlan>>> {
// Push any remaining filters into our data source
- let res = self.data_source.try_pushdown_filters(
- child_pushdown_result
- .parent_filters
- .into_iter()
- .map(|f| match f {
- PredicateSupport::Supported(expr) => expr,
- PredicateSupport::Unsupported(expr) => expr,
- })
- .collect(),
- config,
- )?;
+ let parent_filters = child_pushdown_result
+ .parent_filters
+ .into_iter()
+ .map(|f| f.filter)
+ .collect_vec();
+ let res = self
+ .data_source
+ .try_pushdown_filters(parent_filters.clone(), config)?;
match res.updated_node {
Some(data_source) => {
let mut new_node = self.clone();
@@ -346,9 +340,10 @@ impl ExecutionPlan for DataSourceExec {
let filter = conjunction(
res.filters
.iter()
- .filter_map(|f| match f {
- PredicateSupport::Supported(expr) => Some(Arc::clone(expr)),
- PredicateSupport::Unsupported(_) => None,
+ .zip(parent_filters)
+ .filter_map(|(s, f)| match s {
+ PushedDown::Yes => Some(f),
+ PushedDown::No => None,
})
.collect_vec(),
);
diff --git a/datafusion/physical-optimizer/src/filter_pushdown.rs b/datafusion/physical-optimizer/src/filter_pushdown.rs
index eb7ec81b26..66ccc1a798 100644
--- a/datafusion/physical-optimizer/src/filter_pushdown.rs
+++ b/datafusion/physical-optimizer/src/filter_pushdown.rs
@@ -38,7 +38,8 @@ use crate::PhysicalOptimizerRule;
use datafusion_common::{config::ConfigOptions, Result};
use datafusion_physical_expr::PhysicalExpr;
use datafusion_physical_plan::filter_pushdown::{
- ChildPushdownResult, FilterPushdownPhase, FilterPushdownPropagation, PredicateSupport,
+ ChildFilterPushdownResult, ChildPushdownResult, FilterPushdownPhase,
+ FilterPushdownPropagation, PushedDown,
};
use datafusion_physical_plan::{with_new_children_if_necessary, ExecutionPlan};
@@ -434,24 +435,14 @@ impl PhysicalOptimizerRule for FilterPushdown {
}
}
-/// Support state of each predicate for the children of the node.
-/// These predicates are coming from the parent node.
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-enum ParentPredicateStates {
- NoChildren,
- Unsupported,
- Supported,
-}
-
fn push_down_filters(
node: Arc<dyn ExecutionPlan>,
parent_predicates: Vec<Arc<dyn PhysicalExpr>>,
config: &ConfigOptions,
phase: FilterPushdownPhase,
) -> Result<FilterPushdownPropagation<Arc<dyn ExecutionPlan>>> {
- // If the node has any child, these will be rewritten as supported or unsupported
- let mut parent_predicates_pushdown_states =
- vec![ParentPredicateStates::NoChildren; parent_predicates.len()];
+ let mut parent_filter_pushdown_supports: Vec<Vec<PushedDown>> =
+ vec![vec![]; parent_predicates.len()];
let mut self_filters_pushdown_supports = vec![];
let mut new_children = Vec::with_capacity(node.children().len());
@@ -459,45 +450,66 @@ fn push_down_filters(
let filter_description =
node.gather_filters_for_pushdown(phase, parent_predicates.clone(), config)?;
- for (child, parent_filters, self_filters) in izip!(
+ let filter_description_parent_filters = filter_description.parent_filters();
+ let filter_description_self_filters = filter_description.self_filters();
+ if filter_description_parent_filters.len() != children.len() {
+ return Err(datafusion_common::DataFusionError::Internal(
+ format!(
+ "Filter pushdown expected FilterDescription to have parent filters for {expected_num_children}, but got {actual_num_children} for node {node_name}",
+ expected_num_children = children.len(),
+ actual_num_children = filter_description_parent_filters.len(),
+ node_name = node.name(),
+ ),
+ ));
+ }
+ if filter_description_self_filters.len() != children.len() {
+ return Err(datafusion_common::DataFusionError::Internal(
+ format!(
+ "Filter pushdown expected FilterDescription to have self filters for {expected_num_children}, but got {actual_num_children} for node {node_name}",
+ expected_num_children = children.len(),
+ actual_num_children = filter_description_self_filters.len(),
+ node_name = node.name(),
+ ),
+ ));
+ }
+
+ for (child_idx, (child, parent_filters, self_filters)) in izip!(
children,
filter_description.parent_filters(),
filter_description.self_filters()
- ) {
+ )
+ .enumerate()
+ {
// Here, `parent_filters` are the predicates which are provided by the parent node of
// the current node, and tried to be pushed down over the child which the loop points
// currently. `self_filters` are the predicates which are provided by the current node,
// and tried to be pushed down over the child similarly.
let num_self_filters = self_filters.len();
- let mut parent_supported_predicate_indices = vec![];
- let mut all_predicates = self_filters;
+ let mut all_predicates = self_filters.clone();
+
+ // Track which parent filters are supported for this child
+ let mut parent_filter_indices = vec![];
// Iterate over each predicate coming from the parent
- for (idx, filter) in parent_filters.into_iter().enumerate() {
+ for (parent_filter_idx, filter) in parent_filters.into_iter().enumerate() {
// Check if we can push this filter down to our child.
// These supports are defined in `gather_filters_for_pushdown()`
- match filter {
- PredicateSupport::Supported(predicate) => {
+ match filter.discriminant {
+ PushedDown::Yes => {
// Queue this filter up for pushdown to this child
- all_predicates.push(predicate);
- parent_supported_predicate_indices.push(idx);
- // Mark this filter as supported by our children if no child has marked it as unsupported
- if parent_predicates_pushdown_states[idx]
- != ParentPredicateStates::Unsupported
- {
- parent_predicates_pushdown_states[idx] =
- ParentPredicateStates::Supported;
- }
+ all_predicates.push(filter.predicate);
+ parent_filter_indices.push(parent_filter_idx);
}
- PredicateSupport::Unsupported(_) => {
- // Mark as unsupported by our children
- parent_predicates_pushdown_states[idx] =
- ParentPredicateStates::Unsupported;
+ PushedDown::No => {
+ // This filter won't be pushed down to this child
+ // Will be marked as unsupported later in the initialization loop
}
}
}
+ let num_parent_filters = all_predicates.len() - num_self_filters;
+
// Any filters that could not be pushed down to a child are marked as not-supported to our parents
let result = push_down_filters(Arc::clone(child), all_predicates, config, phase)?;
@@ -513,57 +525,67 @@ fn push_down_filters(
// from our parents and filters that the current node injected. We need to de-entangle
// this since we do need to distinguish between them.
let mut all_filters = result.filters.into_iter().collect_vec();
- let parent_predicates = all_filters.split_off(num_self_filters);
- let self_predicates = all_filters;
- self_filters_pushdown_supports.push(self_predicates);
+ if all_filters.len() != num_self_filters + num_parent_filters {
+ return Err(datafusion_common::DataFusionError::Internal(
+ format!(
+ "Filter pushdown did not return the expected number of filters: expected {num_self_filters} self filters and {num_parent_filters} parent filters, but got {num_filters_from_child}. Likely culprit is {child}",
+ num_self_filters = num_self_filters,
+ num_parent_filters = num_parent_filters,
+ num_filters_from_child = all_filters.len(),
+ child = child.name(),
+ ),
+ ));
+ }
+ let parent_filters = all_filters
+ .split_off(num_self_filters)
+ .into_iter()
+ .collect_vec();
+ self_filters_pushdown_supports.push(
+ all_filters
+ .into_iter()
+ .zip(self_filters)
+ .map(|(s, f)| s.wrap_expression(f))
+ .collect(),
+ );
- for (idx, result) in parent_supported_predicate_indices
- .iter()
- .zip(parent_predicates)
+ // Start by marking all parent filters as unsupported for this child
+ for parent_filter_pushdown_support in parent_filter_pushdown_supports.iter_mut() {
+ parent_filter_pushdown_support.push(PushedDown::No);
+ assert_eq!(
+ parent_filter_pushdown_support.len(),
+ child_idx + 1,
+ "Parent filter pushdown supports should have the same length as the number of children"
+ );
+ }
+ // Map results from pushed-down filters back to original parent filter indices
+ for (result_idx, parent_filter_support) in parent_filters.into_iter().enumerate()
{
- let current_node_state = match result {
- PredicateSupport::Supported(_) => ParentPredicateStates::Supported,
- PredicateSupport::Unsupported(_) => ParentPredicateStates::Unsupported,
- };
- match (current_node_state, parent_predicates_pushdown_states[*idx]) {
- (r, ParentPredicateStates::NoChildren) => {
- // If we have no result, use the current state from this child
- parent_predicates_pushdown_states[*idx] = r;
- }
- (ParentPredicateStates::Supported, ParentPredicateStates::Supported) => {
- // If the current child and all previous children are supported,
- // the filter continues to support it
- parent_predicates_pushdown_states[*idx] =
- ParentPredicateStates::Supported;
- }
- _ => {
- // Either the current child or a previous child marked this filter as unsupported
- parent_predicates_pushdown_states[*idx] =
- ParentPredicateStates::Unsupported;
- }
- }
+ let original_parent_idx = parent_filter_indices[result_idx];
+ parent_filter_pushdown_supports[original_parent_idx][child_idx] =
+ parent_filter_support;
}
}
+
// Re-create this node with new children
let updated_node = with_new_children_if_necessary(Arc::clone(&node), new_children)?;
- // Remap the result onto the parent filters as they were given to us.
- // Any filters that were not pushed down to any children are marked as unsupported.
- let parent_pushdown_result = parent_predicates_pushdown_states
- .into_iter()
- .zip(parent_predicates)
- .map(|(state, filter)| match state {
- ParentPredicateStates::NoChildren => PredicateSupport::Unsupported(filter),
- ParentPredicateStates::Unsupported => PredicateSupport::Unsupported(filter),
- ParentPredicateStates::Supported => PredicateSupport::Supported(filter),
- })
- .collect();
+
// TODO: by calling `handle_child_pushdown_result` we are assuming that the
// `ExecutionPlan` implementation will not change the plan itself.
// Should we have a separate method for dynamic pushdown that does not allow modifying the plan?
let mut res = updated_node.handle_child_pushdown_result(
phase,
ChildPushdownResult {
- parent_filters: parent_pushdown_result,
+ parent_filters: parent_predicates
+ .into_iter()
+ .enumerate()
+ .map(
+ |(parent_filter_idx, parent_filter)| ChildFilterPushdownResult {
+ filter: parent_filter,
+ child_results: parent_filter_pushdown_supports[parent_filter_idx]
+ .clone(),
+ },
+ )
+ .collect(),
self_filters: self_filters_pushdown_supports,
},
config,
diff --git a/datafusion/physical-plan/src/coalesce_batches.rs b/datafusion/physical-plan/src/coalesce_batches.rs
index d008db48b8..d98530d28e 100644
--- a/datafusion/physical-plan/src/coalesce_batches.rs
+++ b/datafusion/physical-plan/src/coalesce_batches.rs
@@ -243,9 +243,7 @@ impl ExecutionPlan for CoalesceBatchesExec {
child_pushdown_result: ChildPushdownResult,
_config: &ConfigOptions,
) -> Result<FilterPushdownPropagation<Arc<dyn ExecutionPlan>>> {
- Ok(FilterPushdownPropagation::transparent(
- child_pushdown_result,
- ))
+ Ok(FilterPushdownPropagation::if_all(child_pushdown_result))
}
}
diff --git a/datafusion/physical-plan/src/execution_plan.rs b/datafusion/physical-plan/src/execution_plan.rs
index 375de7c738..6d51bf195d 100644
--- a/datafusion/physical-plan/src/execution_plan.rs
+++ b/datafusion/physical-plan/src/execution_plan.rs
@@ -18,7 +18,7 @@
pub use crate::display::{DefaultDisplay, DisplayAs, DisplayFormatType, VerboseDisplay};
use crate::filter_pushdown::{
ChildFilterDescription, ChildPushdownResult, FilterDescription, FilterPushdownPhase,
- FilterPushdownPropagation, PredicateSupport,
+ FilterPushdownPropagation, PushedDownPredicate,
};
pub use crate::metrics::Metric;
pub use crate::ordering::InputOrderMode;
@@ -525,7 +525,7 @@ pub trait ExecutionPlan: Debug + DisplayAs + Send + Sync {
let mut desc = FilterDescription::new();
let child_filters = parent_filters
.iter()
- .map(|f| PredicateSupport::Unsupported(Arc::clone(f)))
+ .map(|f| PushedDownPredicate::unsupported(Arc::clone(f)))
.collect_vec();
for _ in 0..self.children().len() {
desc = desc.with_child(ChildFilterDescription {
@@ -563,7 +563,7 @@ pub trait ExecutionPlan: Debug + DisplayAs + Send + Sync {
/// they have been handled.
/// - A `HashJoinExec` might ignore the pushdown result if filters need to
/// be applied during the join operation. It passes the parent filters back
- /// up wrapped in [`FilterPushdownPropagation::transparent`], discarding
+ /// up wrapped in [`FilterPushdownPropagation::if_any`], discarding
/// any self-filters from children.
///
/// **Example Walkthrough:**
@@ -597,10 +597,11 @@ pub trait ExecutionPlan: Debug + DisplayAs + Send + Sync {
///
/// **Helper Methods for Customization:**
/// There are various helper methods to simplify implementing this method:
- /// - [`FilterPushdownPropagation::transparent`]: Indicates that the node
- /// supports filter pushdown but does not modify it, simply transmitting
- /// the children's pushdown results back up to its parent.
- /// - [`FilterPushdownPropagation::with_filters`]: Allows adding filters
+ /// - [`FilterPushdownPropagation::if_any`]: Marks all parent filters as
+ /// supported as long as at least one child supports them.
+ /// - [`FilterPushdownPropagation::if_all`]: Marks all parent filters as
+ /// supported as long as all children support them.
+ /// - [`FilterPushdownPropagation::with_parent_pushdown_result`]: Allows adding filters
/// to the propagation result, indicating which filters are supported by
/// the current node.
/// - [`FilterPushdownPropagation::with_updated_node`]: Allows updating the
@@ -613,16 +614,14 @@ pub trait ExecutionPlan: Debug + DisplayAs + Send + Sync {
/// operator may or may not be allowed to modify the plan. See
/// [`FilterPushdownPhase`] for more details on phase-specific behavior.
///
- /// [`PredicateSupport::Supported`]: crate::filter_pushdown::PredicateSupport::Supported
+ /// [`PushedDownPredicate::supported`]: crate::filter_pushdown::PushedDownPredicate::supported
fn handle_child_pushdown_result(
&self,
_phase: FilterPushdownPhase,
child_pushdown_result: ChildPushdownResult,
_config: &ConfigOptions,
) -> Result<FilterPushdownPropagation<Arc<dyn ExecutionPlan>>> {
- Ok(FilterPushdownPropagation::transparent(
- child_pushdown_result,
- ))
+ Ok(FilterPushdownPropagation::if_all(child_pushdown_result))
}
/// Injects arbitrary run-time state into this execution plan, returning a new plan
diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs
index 66886c5766..8157e1b721 100644
--- a/datafusion/physical-plan/src/filter.rs
+++ b/datafusion/physical-plan/src/filter.rs
@@ -30,7 +30,7 @@ use crate::common::can_project;
use crate::execution_plan::CardinalityEffect;
use crate::filter_pushdown::{
ChildFilterDescription, ChildPushdownResult, FilterDescription, FilterPushdownPhase,
- FilterPushdownPropagation, PredicateSupport,
+ FilterPushdownPropagation, PushedDown, PushedDownPredicate,
};
use crate::projection::{
make_with_child, try_embed_projection, update_expr, EmbeddedProjection,
@@ -455,7 +455,7 @@ impl ExecutionPlan for FilterExec {
// For non-pre phase, filters pass through unchanged
let filter_supports = parent_filters
.into_iter()
- .map(PredicateSupport::Supported)
+ .map(PushedDownPredicate::supported)
.collect();
return Ok(FilterDescription::new().with_child(ChildFilterDescription {
parent_filters: filter_supports,
@@ -481,32 +481,28 @@ impl ExecutionPlan for FilterExec {
_config: &ConfigOptions,
) -> Result<FilterPushdownPropagation<Arc<dyn ExecutionPlan>>> {
if !matches!(phase, FilterPushdownPhase::Pre) {
- return Ok(FilterPushdownPropagation::transparent(
- child_pushdown_result,
- ));
+ return Ok(FilterPushdownPropagation::if_all(child_pushdown_result));
}
// We absorb any parent filters that were not handled by our children
- let mut unhandled_filters = child_pushdown_result
- .parent_filters
- .iter()
- .filter_map(|f| match f {
- PredicateSupport::Unsupported(expr) => Some(Arc::clone(expr)),
- PredicateSupport::Supported(_) => None,
- })
- .collect_vec();
- assert_eq!(
- child_pushdown_result.self_filters.len(),
- 1,
- "FilterExec should only have one child"
- );
- let unsupported_self_filters = child_pushdown_result.self_filters[0]
+ let unsupported_parent_filters =
+ child_pushdown_result.parent_filters.iter().filter_map(|f| {
+ matches!(f.all(), PushedDown::No).then_some(Arc::clone(&f.filter))
+ });
+ let unsupported_self_filters = child_pushdown_result
+ .self_filters
+ .first()
+ .expect("we have exactly one child")
.iter()
- .filter_map(|f| match f {
- PredicateSupport::Unsupported(expr) => Some(Arc::clone(expr)),
- PredicateSupport::Supported(_) => None,
+ .filter_map(|f| match f.discriminant {
+ PushedDown::Yes => None,
+ PushedDown::No => Some(&f.predicate),
})
+ .cloned();
+
+ let unhandled_filters = unsupported_parent_filters
+ .into_iter()
+ .chain(unsupported_self_filters)
.collect_vec();
- unhandled_filters.extend(unsupported_self_filters);
// If we have unhandled filters, we need to create a new FilterExec
let filter_input = Arc::clone(self.input());
@@ -555,15 +551,9 @@ impl ExecutionPlan for FilterExec {
};
Some(Arc::new(new) as _)
};
- // Mark all parent filters as supported since we absorbed them
- let supported_filters = child_pushdown_result
- .parent_filters
- .into_iter()
- .map(|f| PredicateSupport::Supported(f.into_inner()))
- .collect();
Ok(FilterPushdownPropagation {
- filters: supported_filters,
+ filters: vec![PushedDown::Yes; child_pushdown_result.parent_filters.len()],
updated_node,
})
}
diff --git a/datafusion/physical-plan/src/filter_pushdown.rs b/datafusion/physical-plan/src/filter_pushdown.rs
index b50f86382e..a3e94a75c8 100644
--- a/datafusion/physical-plan/src/filter_pushdown.rs
+++ b/datafusion/physical-plan/src/filter_pushdown.rs
@@ -86,25 +86,111 @@ impl std::fmt::Display for FilterPushdownPhase {
/// The result of a plan for pushing down a filter into a child node.
/// This contains references to filters so that nodes can mutate a filter
/// before pushing it down to a child node (e.g. to adjust a projection)
-/// or can directly take ownership of `Unsupported` filters that their children
+/// or can directly take ownership of filters that their children
/// could not handle.
#[derive(Debug, Clone)]
-pub enum PredicateSupport {
- Supported(Arc<dyn PhysicalExpr>),
- Unsupported(Arc<dyn PhysicalExpr>),
+pub struct PushedDownPredicate {
+ pub discriminant: PushedDown,
+ pub predicate: Arc<dyn PhysicalExpr>,
}
-impl PredicateSupport {
+impl PushedDownPredicate {
+ /// Return the wrapped [`PhysicalExpr`], discarding whether it is supported or unsupported.
pub fn into_inner(self) -> Arc<dyn PhysicalExpr> {
- match self {
- PredicateSupport::Supported(expr) | PredicateSupport::Unsupported(expr) => {
- expr
- }
+ self.predicate
+ }
+
+ /// Create a new [`PushedDownPredicate`] with supported pushdown.
+ pub fn supported(predicate: Arc<dyn PhysicalExpr>) -> Self {
+ Self {
+ discriminant: PushedDown::Yes,
+ predicate,
+ }
+ }
+
+ /// Create a new [`PushedDownPredicate`] with unsupported pushdown.
+ pub fn unsupported(predicate: Arc<dyn PhysicalExpr>) -> Self {
+ Self {
+ discriminant: PushedDown::No,
+ predicate,
+ }
+ }
+}
+
+/// Discriminant for the result of pushing down a filter into a child node.
+#[derive(Debug, Clone, Copy)]
+pub enum PushedDown {
+ /// The predicate was successfully pushed down into the child node.
+ Yes,
+ /// The predicate could not be pushed down into the child node.
+ No,
+}
+
+impl PushedDown {
+ /// Logical AND operation: returns `Yes` only if both operands are `Yes`.
+ pub fn and(self, other: PushedDown) -> PushedDown {
+ match (self, other) {
+ (PushedDown::Yes, PushedDown::Yes) => PushedDown::Yes,
+ _ => PushedDown::No,
+ }
+ }
+
+ /// Logical OR operation: returns `Yes` if either operand is `Yes`.
+ pub fn or(self, other: PushedDown) -> PushedDown {
+ match (self, other) {
+ (PushedDown::Yes, _) | (_, PushedDown::Yes) => PushedDown::Yes,
+ (PushedDown::No, PushedDown::No) => PushedDown::No,
+ }
+ }
+
+ /// Wrap a [`PhysicalExpr`] with this pushdown result.
+ pub fn wrap_expression(self, expr: Arc<dyn PhysicalExpr>) -> PushedDownPredicate {
+ PushedDownPredicate {
+ discriminant: self,
+ predicate: expr,
+ }
+ }
+}
+
+/// The result of pushing down a single parent filter into all children.
+#[derive(Debug, Clone)]
+pub struct ChildFilterPushdownResult {
+ pub filter: Arc<dyn PhysicalExpr>,
+ pub child_results: Vec<PushedDown>,
+}
+
+impl ChildFilterPushdownResult {
+ /// Combine all child results using OR logic.
+ /// Returns `Yes` if **any** child supports the filter.
+ /// Returns `No` if **all** children reject the filter or if there are no children.
+ pub fn any(&self) -> PushedDown {
+ if self.child_results.is_empty() {
+ // If there are no children, filters cannot be supported
+ PushedDown::No
+ } else {
+ self.child_results
+ .iter()
+ .fold(PushedDown::No, |acc, result| acc.or(*result))
+ }
+ }
+
+ /// Combine all child results using AND logic.
+ /// Returns `Yes` if **all** children support the filter.
+ /// Returns `No` if **any** child rejects the filter or if there are no children.
+ pub fn all(&self) -> PushedDown {
+ if self.child_results.is_empty() {
+ // If there are no children, filters cannot be supported
+ PushedDown::No
+ } else {
+ self.child_results
+ .iter()
+ .fold(PushedDown::Yes, |acc, result| acc.and(*result))
}
}
}
/// The result of pushing down filters into a child node.
+///
/// This is the result provided to nodes in [`ExecutionPlan::handle_child_pushdown_result`].
/// Nodes process this result and convert it into a [`FilterPushdownPropagation`]
/// that is returned to their parent.
@@ -112,51 +198,68 @@ impl PredicateSupport {
/// [`ExecutionPlan::handle_child_pushdown_result`]: crate::ExecutionPlan::handle_child_pushdown_result
#[derive(Debug, Clone)]
pub struct ChildPushdownResult {
- /// The combined result of pushing down each parent filter into each child.
- /// For example, given the fitlers `[a, b]` and children `[1, 2, 3]` the matrix of responses:
- ///
- // | filter | child 1 | child 2 | child 3 | result |
- // |--------|-------------|-----------|-----------|-------------|
- // | a | Supported | Supported | Supported | Supported |
- // | b | Unsupported | Supported | Supported | Unsupported |
- ///
- /// That is: if any child marks a filter as unsupported or if the filter was not pushed
- /// down into any child then the result is unsupported.
- /// If at least one children and all children that received the filter mark it as supported
- /// then the result is supported.
- pub parent_filters: Vec<PredicateSupport>,
+ /// The parent filters that were pushed down as received by the current node when [`ExecutionPlan::gather_filters_for_pushdown`](crate::ExecutionPlan::handle_child_pushdown_result) was called.
+ /// Note that this may *not* be the same as the filters that were passed to the children as the current node may have modified them
+ /// (e.g. by reassigning column indices) when it returned them from [`ExecutionPlan::gather_filters_for_pushdown`](crate::ExecutionPlan::handle_child_pushdown_result) in a [`FilterDescription`].
+ /// Attached to each filter is a [`PushedDown`] *per child* that indicates whether the filter was supported or unsupported by each child.
+ /// To get combined results see [`ChildFilterPushdownResult::any`] and [`ChildFilterPushdownResult::all`].
+ pub parent_filters: Vec<ChildFilterPushdownResult>,
/// The result of pushing down each filter this node provided into each of it's children.
- /// This is not combined with the parent filters so that nodes can treat each child independently.
- pub self_filters: Vec<Vec<PredicateSupport>>,
+ /// The outer vector corresponds to each child, and the inner vector corresponds to each filter.
+ /// Since this node may have generated a different filter for each child the inner vector may have different lengths or the expressions may not match at all.
+ /// It is up to each node to interpret this result based on the filters it provided for each child in [`ExecutionPlan::gather_filters_for_pushdown`](crate::ExecutionPlan::handle_child_pushdown_result).
+ pub self_filters: Vec<Vec<PushedDownPredicate>>,
}
-/// The result of pushing down filters into a node that it returns to its parent.
-/// This is what nodes return from [`ExecutionPlan::handle_child_pushdown_result`] to communicate
+/// The result of pushing down filters into a node.
+///
+/// Returned from [`ExecutionPlan::handle_child_pushdown_result`] to communicate
/// to the optimizer:
///
-/// 1. What to do with any parent filters that were not completely handled by the children.
+/// 1. What to do with any parent filters that were could not be pushed down into the children.
/// 2. If the node needs to be replaced in the execution plan with a new node or not.
///
/// [`ExecutionPlan::handle_child_pushdown_result`]: crate::ExecutionPlan::handle_child_pushdown_result
#[derive(Debug, Clone)]
pub struct FilterPushdownPropagation<T> {
- pub filters: Vec<PredicateSupport>,
+ /// What filters were pushed into the parent node.
+ pub filters: Vec<PushedDown>,
+ /// The updated node, if it was updated during pushdown
pub updated_node: Option<T>,
}
impl<T> FilterPushdownPropagation<T> {
- /// Create a new [`FilterPushdownPropagation`] that tells the parent node
- /// that echoes back up to the parent the result of pushing down the filters
- /// into the children.
- pub fn transparent(child_pushdown_result: ChildPushdownResult) -> Self {
+ /// Create a new [`FilterPushdownPropagation`] that tells the parent node that each parent filter
+ /// is supported if it was supported by *all* children.
+ pub fn if_all(child_pushdown_result: ChildPushdownResult) -> Self {
+ let filters = child_pushdown_result
+ .parent_filters
+ .into_iter()
+ .map(|result| result.all())
+ .collect();
+ Self {
+ filters,
+ updated_node: None,
+ }
+ }
+
+ /// Create a new [`FilterPushdownPropagation`] that tells the parent node that each parent filter
+ /// is supported if it was supported by *any* child.
+ pub fn if_any(child_pushdown_result: ChildPushdownResult) -> Self {
+ let filters = child_pushdown_result
+ .parent_filters
+ .into_iter()
+ .map(|result| result.any())
+ .collect();
Self {
- filters: child_pushdown_result.parent_filters,
+ filters,
updated_node: None,
}
}
/// Create a new [`FilterPushdownPropagation`] with the specified filter support.
- pub fn with_filters(filters: Vec<PredicateSupport>) -> Self {
+ /// This transmits up to our parent node what the result of pushing down the filters into our node and possibly our subtree was.
+ pub fn with_parent_pushdown_result(filters: Vec<PushedDown>) -> Self {
Self {
filters,
updated_node: None,
@@ -164,19 +267,25 @@ impl<T> FilterPushdownPropagation<T> {
}
/// Bind an updated node to the [`FilterPushdownPropagation`].
+ /// Use this when the current node wants to update iself in the tree or replace itself with a new node (e.g. one of it's children).
+ /// You do not need to call this if one of the children of the current node may have updated itself, that is handled by the optimizer.
pub fn with_updated_node(mut self, updated_node: T) -> Self {
self.updated_node = Some(updated_node);
self
}
}
+/// Describes filter pushdown for a single child node.
+///
+/// This structure contains two types of filters:
+/// - **Parent filters**: Filters received from the parent node, marked as supported or unsupported
+/// - **Self filters**: Filters generated by the current node to be pushed down to this child
#[derive(Debug, Clone)]
pub struct ChildFilterDescription {
/// Description of which parent filters can be pushed down into this node.
/// Since we need to transmit filter pushdown results back to this node's parent
/// we need to track each parent filter for each child, even those that are unsupported / won't be pushed down.
- /// We do this using a [`PredicateSupport`] which simplifies manipulating supported/unsupported filters.
- pub(crate) parent_filters: Vec<PredicateSupport>,
+ pub(crate) parent_filters: Vec<PushedDownPredicate>,
/// Description of which filters this node is pushing down to its children.
/// Since this is not transmitted back to the parents we can have variable sized inner arrays
/// instead of having to track supported/unsupported.
@@ -186,6 +295,10 @@ pub struct ChildFilterDescription {
impl ChildFilterDescription {
/// Build a child filter description by analyzing which parent filters can be pushed to a specific child.
///
+ /// This method performs column analysis to determine which filters can be pushed down:
+ /// - If all columns referenced by a filter exist in the child's schema, it can be pushed down
+ /// - Otherwise, it cannot be pushed down to that child
+ ///
/// See [`FilterDescription::from_children`] for more details
pub fn from_child(
parent_filters: &[Arc<dyn PhysicalExpr>],
@@ -217,11 +330,12 @@ impl ChildFilterDescription {
// Need to reassign column indices to match child schema
let reassigned_filter =
reassign_predicate_columns(Arc::clone(filter), &child_schema, false)?;
- child_parent_filters.push(PredicateSupport::Supported(reassigned_filter));
+ child_parent_filters
+ .push(PushedDownPredicate::supported(reassigned_filter));
} else {
// Some columns don't exist in child - cannot push down
child_parent_filters
- .push(PredicateSupport::Unsupported(Arc::clone(filter)));
+ .push(PushedDownPredicate::unsupported(Arc::clone(filter)));
}
}
@@ -244,6 +358,14 @@ impl ChildFilterDescription {
}
}
+/// Describes how filters should be pushed down to children.
+///
+/// This structure contains filter descriptions for each child node, specifying:
+/// - Which parent filters can be pushed down to each child
+/// - Which self-generated filters should be pushed down to each child
+///
+/// The filter routing is determined by column analysis - filters can only be pushed
+/// to children whose schemas contain all the referenced columns.
#[derive(Debug, Clone)]
pub struct FilterDescription {
/// A filter description for each child.
@@ -291,7 +413,7 @@ impl FilterDescription {
Ok(desc)
}
- pub fn parent_filters(&self) -> Vec<Vec<PredicateSupport>> {
+ pub fn parent_filters(&self) -> Vec<Vec<PushedDownPredicate>> {
self.child_filter_descriptions
.iter()
.map(|d| &d.parent_filters)
diff --git a/datafusion/physical-plan/src/repartition/mod.rs b/datafusion/physical-plan/src/repartition/mod.rs
index 37dc3a7675..754a208126 100644
--- a/datafusion/physical-plan/src/repartition/mod.rs
+++ b/datafusion/physical-plan/src/repartition/mod.rs
@@ -821,9 +821,7 @@ impl ExecutionPlan for RepartitionExec {
child_pushdown_result: ChildPushdownResult,
_config: &ConfigOptions,
) -> Result<FilterPushdownPropagation<Arc<dyn ExecutionPlan>>> {
- Ok(FilterPushdownPropagation::transparent(
- child_pushdown_result,
- ))
+ Ok(FilterPushdownPropagation::if_all(child_pushdown_result))
}
}
diff --git a/datafusion/sqllogictest/src/engines/postgres_engine/mod.rs b/datafusion/sqllogictest/src/engines/postgres_engine/mod.rs
index 68816626bf..375f06d34b 100644
--- a/datafusion/sqllogictest/src/engines/postgres_engine/mod.rs
+++ b/datafusion/sqllogictest/src/engines/postgres_engine/mod.rs
@@ -93,7 +93,7 @@ impl Postgres {
let spawned_task = SpawnedTask::spawn(async move {
if let Err(e) = connection.await {
- log::error!("Postgres connection error: {:?}", e);
+ log::error!("Postgres connection error: {e:?}");
}
});
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org