From a48b44dd9b67a2ad59d42a3533b2f32178d61ee5 Mon Sep 17 00:00:00 2001 From: haohuaijin Date: Sun, 13 Sep 2026 18:17:35 +0800 Subject: [PATCH 1/8] fix: preserve column indices in join dynamic filter pushdown --- .../physical_optimizer/filter_pushdown.rs | 137 ++++++++++++++++++ .../physical-plan/src/filter_pushdown.rs | 120 ++++++++++----- .../physical-plan/src/joins/hash_join/exec.rs | 91 ++++++------ .../physical-plan/src/repartition/mod.rs | 11 +- .../dynamic_filter_pushdown_config.slt | 120 +++++++++++++++ 5 files changed, 390 insertions(+), 89 deletions(-) diff --git a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs index 26e6e0c74c49d..b07603a1180b8 100644 --- a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs +++ b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs @@ -1601,6 +1601,143 @@ fn test_hashjoin_parent_filter_pushdown_same_column_names() { ); } +/// Repartition must preserve the second of two same-named columns when +/// forwarding a predicate, in both physical pushdown phases. +#[test] +fn test_repartition_filter_pushdown_preserves_duplicate_column_indices() { + use datafusion_physical_plan::filter_pushdown::{FilterPushdownPhase, PushedDown}; + + let schema = Arc::new(Schema::new(vec![ + Field::new("id", DataType::Utf8, false), + Field::new("id", DataType::Utf8, false), + ])); + let input = TestScanBuilder::new(schema).build(); + let repartition = + RepartitionExec::try_new(input, Partitioning::RoundRobinBatch(4)).unwrap(); + let predicate: Arc = Arc::new(BinaryExpr::new( + Arc::new(Column::new("id", 1)), + Operator::Eq, + Arc::new(Literal::new(ScalarValue::from("x"))), + )); + for phase in [FilterPushdownPhase::Pre, FilterPushdownPhase::Post] { + let filters = repartition + .gather_filters_for_pushdown( + phase, + vec![Arc::clone(&predicate)], + &ConfigOptions::default(), + ) + .unwrap() + .parent_filters(); + assert_eq!(filters.len(), 1); + assert!(matches!(filters[0][0].discriminant, PushedDown::Yes)); + assert_eq!(filters[0][0].predicate.to_string(), "id@1 = x", "{phase}"); + } +} + +/// A join's output projection must map to child positions even when a child +/// contains multiple columns with the same name. +#[test] +fn test_hashjoin_parent_filter_pushdown_duplicate_child_columns() { + use datafusion_physical_plan::filter_pushdown::{FilterPushdownPhase, PushedDown}; + + let schema = Arc::new(Schema::new(vec![ + Field::new("id", DataType::Utf8, false), + Field::new("id", DataType::Utf8, false), + ])); + for projection in [None, Some(vec![3, 1, 2, 0])] { + let output_indices = projection.clone().unwrap_or_else(|| vec![0, 1, 2, 3]); + let join = HashJoinExec::try_new( + TestScanBuilder::new(Arc::clone(&schema)).build(), + TestScanBuilder::new(Arc::clone(&schema)).build(), + vec![( + Arc::new(Column::new("id", 0)), + Arc::new(Column::new("id", 0)), + )], + None, + &JoinType::Inner, + projection, + PartitionMode::Partitioned, + datafusion_common::NullEquality::NullEqualsNothing, + false, + ) + .unwrap(); + for (output_index, input_index) in output_indices.into_iter().enumerate() { + let predicate = Arc::new(BinaryExpr::new( + Arc::new(Column::new("id", output_index)), + Operator::Eq, + Arc::new(Literal::new(ScalarValue::from("x"))), + )); + let filters = join + .gather_filters_for_pushdown( + FilterPushdownPhase::Pre, + vec![predicate], + &ConfigOptions::default(), + ) + .unwrap() + .parent_filters(); + let side = input_index / 2; + assert!(matches!(filters[side][0].discriminant, PushedDown::Yes)); + assert!(matches!(filters[1 - side][0].discriminant, PushedDown::No)); + assert_eq!( + filters[side][0].predicate.to_string(), + format!("id@{} = x", input_index % 2) + ); + } + } +} + +/// Semi joins must use the paired join key, not a same-named non-key column +/// on the other side. Exercise both directions and a reordered projection. +#[test] +fn test_hashjoin_parent_filter_pushdown_semi_join_key_mapping() { + use datafusion_physical_plan::filter_pushdown::{FilterPushdownPhase, PushedDown}; + + let schema = Arc::new(Schema::new(vec![ + Field::new("left_key", DataType::Utf8, false), + Field::new("right_key", DataType::Utf8, false), + ])); + for (join_type, output_key_index) in + [(JoinType::LeftSemi, 1), (JoinType::RightSemi, 0)] + { + let join = HashJoinExec::try_new( + TestScanBuilder::new(Arc::clone(&schema)).build(), + TestScanBuilder::new(Arc::clone(&schema)).build(), + vec![( + Arc::new(Column::new("left_key", 0)), + Arc::new(Column::new("right_key", 1)), + )], + None, + &join_type, + Some(vec![1, 0]), + PartitionMode::Partitioned, + datafusion_common::NullEquality::NullEqualsNothing, + false, + ) + .unwrap(); + let predicate = Arc::new(BinaryExpr::new( + Arc::new(Column::new( + join.schema().field(output_key_index).name(), + output_key_index, + )), + Operator::Eq, + Arc::new(Literal::new(ScalarValue::from("x"))), + )); + let filters = join + .gather_filters_for_pushdown( + FilterPushdownPhase::Pre, + vec![predicate], + &ConfigOptions::default(), + ) + .unwrap() + .parent_filters(); + for side in &filters { + assert!(matches!(side[0].discriminant, PushedDown::Yes)); + } + assert_eq!(filters[0][0].predicate.to_string(), "left_key@0 = x"); + assert_eq!(filters[1][0].predicate.to_string(), "right_key@1 = x"); + } +} + #[test] fn test_hashjoin_parent_filter_pushdown_mark_join() { let left_schema = Arc::new(Schema::new(vec![ diff --git a/datafusion/physical-plan/src/filter_pushdown.rs b/datafusion/physical-plan/src/filter_pushdown.rs index dfaec62d062ec..0010dec07f9dc 100644 --- a/datafusion/physical-plan/src/filter_pushdown.rs +++ b/datafusion/physical-plan/src/filter_pushdown.rs @@ -34,7 +34,7 @@ //! //! See also datafusion/physical-optimizer/src/filter_pushdown.rs. -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use std::sync::Arc; use arrow_schema::SchemaRef; @@ -318,17 +318,20 @@ pub struct ChildFilterDescription { /// 1. Verify that all columns referenced by the filter exist in the target /// 2. Remap column indices to match the target schema /// -/// `allowed_indices` controls which column indices (in the parent schema) are -/// considered valid. For single-input nodes this defaults to -/// `0..child_schema.len()` (all columns are reachable). For join nodes it is -/// restricted to the subset of output columns that map to the target child, -/// which is critical when different sides have same-named columns. -pub(crate) struct FilterRemapper { - /// The target schema to remap column indices into. - child_schema: SchemaRef, - /// Only columns at these indices (in the *parent* schema) are considered - /// valid. For non-join nodes this defaults to `0..child_schema.len()`. - allowed_indices: HashSet, +/// Use an explicit positional mapping when the parent can reorder columns or +/// the child can contain duplicate names, such as when pushing through a join. +pub(crate) enum FilterRemapper { + ByName { + child_schema: SchemaRef, + /// Reachable column indices in the parent schema. + allowed_indices: HashSet, + }, + ByIndex { + child_schema: SchemaRef, + /// Parent output index to child input index. Missing entries cannot + /// be pushed to this child. + column_mapping: HashMap, + }, } impl FilterRemapper { @@ -336,31 +339,26 @@ impl FilterRemapper { /// `0..child_schema.len()` and whose name exists in the target schema. pub(crate) fn new(child_schema: SchemaRef) -> Self { let allowed_indices = (0..child_schema.fields().len()).collect(); - Self { + Self::ByName { child_schema, allowed_indices, } } - /// Create a remapper that only accepts columns at the given indices. - /// This is used by join nodes to restrict pushdown to one side of the - /// join when both sides have same-named columns. + /// Create a name-based remapper that only accepts columns at the given + /// parent indices. fn with_allowed_indices( child_schema: SchemaRef, allowed_indices: HashSet, ) -> Self { - Self { + Self::ByName { child_schema, allowed_indices, } } /// Try to remap a filter's column references to the target schema. - /// - /// Validates and remaps in a single tree traversal: for each column, - /// checks that its index is in the allowed set and that - /// its name exists in the target schema, then remaps the index. - /// Returns `Some(remapped)` if all columns are valid, or `None` if any + /// Returns `Some(remapped)` if all columns are reachable, or `None` if any /// column fails validation. pub(crate) fn try_remap( &self, @@ -369,13 +367,27 @@ impl FilterRemapper { let mut all_valid = true; let transformed = Arc::clone(filter).transform_down(|expr| { if let Some(col) = expr.downcast_ref::() { - if self.allowed_indices.contains(&col.index()) - && let Ok(new_index) = self.child_schema.index_of(col.name()) - { - Ok(Transformed::yes(Arc::new(Column::new( - col.name(), - new_index, - )))) + let remapped = match self { + Self::ByName { + child_schema, + allowed_indices, + } => allowed_indices + .contains(&col.index()) + .then(|| child_schema.index_of(col.name()).ok()) + .flatten() + .map(|index| Column::new(col.name(), index)), + Self::ByIndex { + child_schema, + column_mapping, + } => column_mapping.get(&col.index()).and_then(|&index| { + child_schema + .fields() + .get(index) + .map(|field| Column::new(field.name(), index)) + }), + }; + if let Some(remapped) = remapped { + Ok(Transformed::yes(Arc::new(remapped))) } else { all_valid = false; Ok(Transformed::complete(expr)) @@ -412,18 +424,29 @@ impl ChildFilterDescription { Self::remap_filters(parent_filters, &remapper) } + /// Forward parent filters through a node that preserves column positions. + /// Columns are resolved at the same index in the child schema, even if + /// multiple fields share a name. Nodes that project or reorder columns + /// must instead provide an explicit mapping. + pub fn from_child_preserving_indices( + parent_filters: &[Arc], + child: &Arc, + ) -> Result { + if parent_filters.is_empty() { + return Ok(Self::empty()); + } + let column_mapping = (0..child.schema().fields().len()) + .map(|index| (index, index)) + .collect(); + Self::from_child_with_column_mapping(parent_filters, column_mapping, child) + } + /// Like [`Self::from_child`], but restricts which parent-level columns are - /// considered reachable through this child. - /// - /// `allowed_indices` is the set of column indices (in the *parent* - /// schema) that map to this child's side of a join. A filter is only - /// eligible for pushdown when **every** column index it references - /// appears in `allowed_indices`. + /// considered reachable through this child. All columns in a filter must + /// appear in `allowed_indices`; their child positions are resolved by name. /// - /// This prevents incorrect pushdown when different join sides have - /// columns with the same name: matching on index ensures a filter - /// referencing the right side's `k@2` is not pushed to the left side - /// which also has a column named `k` but at a different index. + /// Use [`Self::from_child_with_column_mapping`] when the child can contain + /// duplicate names or when column identity must be preserved across a join. pub fn from_child_with_allowed_indices( parent_filters: &[Arc], allowed_indices: HashSet, @@ -438,6 +461,25 @@ impl ChildFilterDescription { Self::remap_filters(parent_filters, &remapper) } + /// Remap parent filters using an explicit parent-output to child-input + /// column mapping. Columns absent from the mapping cannot be pushed down. + /// Unlike name-based remapping, this preserves column identity when a child + /// has duplicate field names, and supports differently named join keys. + pub fn from_child_with_column_mapping( + parent_filters: &[Arc], + column_mapping: HashMap, + child: &Arc, + ) -> Result { + if parent_filters.is_empty() { + return Ok(Self::empty()); + } + let remapper = FilterRemapper::ByIndex { + child_schema: child.schema(), + column_mapping, + }; + Self::remap_filters(parent_filters, &remapper) + } + fn remap_filters( parent_filters: &[Arc], remapper: &FilterRemapper, diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index b72e180543f9a..12461ec106b5f 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -use std::collections::HashSet; +use std::collections::HashMap; use std::fmt; use std::mem::size_of; use std::sync::{Arc, OnceLock}; @@ -1825,7 +1825,9 @@ impl ExecutionPlan for HashJoinExec { // (e.g. nested mark joins both producing "mark" columns). let (left_preserved, right_preserved) = lr_is_preserved(self.join_type); - // Build the set of allowed column indices for each side + // Map each output position to its input position, accounting for the + // join's projection. Looking up child columns by name is ambiguous when + // a nested join produces multiple fields with the same name. let column_indices: Vec = match self.projection.as_ref() { Some(projection) => projection .iter() @@ -1834,59 +1836,52 @@ impl ExecutionPlan for HashJoinExec { None => self.column_indices.clone(), }; - let (mut left_allowed, mut right_allowed) = (HashSet::new(), HashSet::new()); - column_indices - .iter() - .enumerate() - .for_each(|(output_idx, ci)| { - match ci.side { - JoinSide::Left => left_allowed.insert(output_idx), - JoinSide::Right => right_allowed.insert(output_idx), - // Mark columns - don't allow pushdown to either side - JoinSide::None => false, - }; - }); - - // For semi joins, filters on output join keys can also be pushed to the - // non-output side: every emitted row has an equal key there. This is not - // true for anti joins, whose emitted rows have no match. - match self.join_type { - JoinType::LeftSemi => { - let left_key_indices: HashSet = self - .on - .iter() - .filter_map(|(left_key, _)| { - left_key.downcast_ref::().map(|c| c.index()) - }) - .collect(); - for (output_idx, ci) in column_indices.iter().enumerate() { - if ci.side == JoinSide::Left && left_key_indices.contains(&ci.index) { - right_allowed.insert(output_idx); - } + let (mut left_mapping, mut right_mapping) = (HashMap::new(), HashMap::new()); + for (output_idx, ci) in column_indices.iter().enumerate() { + match ci.side { + JoinSide::Left => { + left_mapping.insert(output_idx, ci.index); } + JoinSide::Right => { + right_mapping.insert(output_idx, ci.index); + } + // Mark columns cannot be pushed to either side. + JoinSide::None => {} } - JoinType::RightSemi => { - let right_key_indices: HashSet = self - .on - .iter() - .filter_map(|(_, right_key)| { - right_key.downcast_ref::().map(|c| c.index()) + } + + // Semi joins also allow filters on output join keys to reach the + // non-output side. Map to the paired key's position, which need not + // have the same name. Only direct column pairs can be remapped this way. + // Anti joins cannot do this: their emitted rows have no matching key. + if matches!(self.join_type, JoinType::LeftSemi | JoinType::RightSemi) { + let key_mapping: HashMap = self + .on + .iter() + .filter_map(|(left_key, right_key)| { + let left = left_key.downcast_ref::()?; + let right = right_key.downcast_ref::()?; + Some(match self.join_type { + JoinType::LeftSemi => (left.index(), right.index()), + _ => (right.index(), left.index()), }) - .collect(); - for (output_idx, ci) in column_indices.iter().enumerate() { - if ci.side == JoinSide::Right && right_key_indices.contains(&ci.index) - { - left_allowed.insert(output_idx); - } + }) + .collect(); + let other_mapping = match self.join_type { + JoinType::LeftSemi => &mut right_mapping, + _ => &mut left_mapping, + }; + for (output_idx, ci) in column_indices.iter().enumerate() { + if let Some(&input_idx) = key_mapping.get(&ci.index) { + other_mapping.insert(output_idx, input_idx); } } - _ => {} } let left_child = if left_preserved { - ChildFilterDescription::from_child_with_allowed_indices( + ChildFilterDescription::from_child_with_column_mapping( &parent_filters, - left_allowed, + left_mapping, self.left(), )? } else { @@ -1894,9 +1889,9 @@ impl ExecutionPlan for HashJoinExec { }; let mut right_child = if right_preserved { - ChildFilterDescription::from_child_with_allowed_indices( + ChildFilterDescription::from_child_with_column_mapping( &parent_filters, - right_allowed, + right_mapping, self.right(), )? } else { diff --git a/datafusion/physical-plan/src/repartition/mod.rs b/datafusion/physical-plan/src/repartition/mod.rs index b653a74fa5d27..68fe014f872a1 100644 --- a/datafusion/physical-plan/src/repartition/mod.rs +++ b/datafusion/physical-plan/src/repartition/mod.rs @@ -75,7 +75,7 @@ use datafusion_physical_expr_common::sort_expr::{ use datafusion_proto_models::protobuf; use crate::filter_pushdown::{ - ChildPushdownResult, FilterDescription, FilterPushdownPhase, + ChildFilterDescription, ChildPushdownResult, FilterDescription, FilterPushdownPhase, FilterPushdownPropagation, }; use crate::joins::SeededRandomState; @@ -2003,7 +2003,14 @@ impl ExecutionPlan for RepartitionExec { parent_filters: Vec>, _config: &ConfigOptions, ) -> Result { - FilterDescription::from_children(parent_filters, &self.children()) + // Repartition changes row placement, not column positions. Preserve + // indices so a nested join's same-named columns remain distinct. + Ok(FilterDescription::new().with_child( + ChildFilterDescription::from_child_preserving_indices( + &parent_filters, + self.input(), + )?, + )) } fn handle_child_pushdown_result( diff --git a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt index 514ae81095a49..fcfccd7966215 100644 --- a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt +++ b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt @@ -904,3 +904,123 @@ SET datafusion.optimizer.enable_dynamic_filter_pushdown = true; statement ok RESET datafusion.execution.parquet.max_row_group_size; + +# Regression for https://github.com/apache/datafusion/issues/25244. +# Keep both a.id and b.id in the nested join output. A dynamic filter on b.id +# must not be remapped by name to a.id, which has a different value. LEFT JOINs +# preserve this join tree through logical optimization. +statement ok +SET datafusion.execution.target_partitions = 1; + +statement ok +SET datafusion.execution.collect_statistics = true; + +statement ok +SET datafusion.execution.parquet.schema_force_view_types = false; + +statement ok +SET datafusion.optimizer.join_reordering = false; + +query I +COPY (SELECT 'a1' AS id, 'x1' AS ty) +TO 'test_files/scratch/dynamic_filter_pushdown_config/issue_25244_a.parquet' +STORED AS PARQUET; +---- +1 + +query I +COPY (SELECT 'x1' AS id) +TO 'test_files/scratch/dynamic_filter_pushdown_config/issue_25244_b.parquet' +STORED AS PARQUET; +---- +1 + +statement ok +CREATE EXTERNAL TABLE issue_25244_a +STORED AS PARQUET +LOCATION 'test_files/scratch/dynamic_filter_pushdown_config/issue_25244_a.parquet'; + +statement ok +CREATE EXTERNAL TABLE issue_25244_b +STORED AS PARQUET +LOCATION 'test_files/scratch/dynamic_filter_pushdown_config/issue_25244_b.parquet'; + +statement ok +SET datafusion.optimizer.enable_join_dynamic_filter_pushdown = false; + +query TTTT +SELECT s.id AS sid, a.id AS aid, b.id AS bid, c.id AS cid +FROM issue_25244_b s +JOIN ( + (issue_25244_a a LEFT JOIN issue_25244_b b ON a.ty = b.id) + LEFT JOIN issue_25244_b c ON b.id = c.id +) +ON s.id = b.id; +---- +x1 a1 x1 x1 + +statement ok +SET datafusion.optimizer.enable_join_dynamic_filter_pushdown = true; + +query TTTT +SELECT s.id AS sid, a.id AS aid, b.id AS bid, c.id AS cid +FROM issue_25244_b s +JOIN ( + (issue_25244_a a LEFT JOIN issue_25244_b b ON a.ty = b.id) + LEFT JOIN issue_25244_b c ON b.id = c.id +) +ON s.id = b.id; +---- +x1 a1 x1 x1 + +# Parallel execution inserts RepartitionExec above the nested joins. It must +# preserve the second id's position when forwarding the dynamic filter. +statement ok +SET datafusion.execution.target_partitions = 4; + +statement ok +SET datafusion.optimizer.enable_join_dynamic_filter_pushdown = false; + +query TTTT +SELECT s.id AS sid, a.id AS aid, b.id AS bid, c.id AS cid +FROM issue_25244_b s +JOIN ( + (issue_25244_a a LEFT JOIN issue_25244_b b ON a.ty = b.id) + LEFT JOIN issue_25244_b c ON b.id = c.id +) +ON s.id = b.id; +---- +x1 a1 x1 x1 + +statement ok +SET datafusion.optimizer.enable_join_dynamic_filter_pushdown = true; + +query TTTT +SELECT s.id AS sid, a.id AS aid, b.id AS bid, c.id AS cid +FROM issue_25244_b s +JOIN ( + (issue_25244_a a LEFT JOIN issue_25244_b b ON a.ty = b.id) + LEFT JOIN issue_25244_b c ON b.id = c.id +) +ON s.id = b.id; +---- +x1 a1 x1 x1 + +statement ok +DROP TABLE issue_25244_a; + +statement ok +DROP TABLE issue_25244_b; + +# The SLT runner uses four target partitions. +statement ok +SET datafusion.execution.target_partitions = 4; + +statement ok +RESET datafusion.execution.collect_statistics; + +statement ok +RESET datafusion.execution.parquet.schema_force_view_types; + +statement ok +RESET datafusion.optimizer.join_reordering; From 82ab235c445466ac61f3b192b79b1b7d2327160a Mon Sep 17 00:00:00 2001 From: Huaijin Date: Sun, 13 Sep 2026 20:29:09 +0800 Subject: [PATCH 2/8] fix: map pushed-down filter columns by position instead of by name Physical filter pushdown resolved a pushed filter's columns in the child schema by name. When the child contains duplicate column names, such as a.id and b.id from a nested join, a join dynamic filter on the second column was rewritten onto the first and silently dropped matching rows. Make FilterRemapper positional only: an identity mapping for nodes that preserve their input schema, or an explicit parent-output to child-input mapping. HashJoin builds the mapping from its column indices and output projection (semi joins map output keys to the paired key), FilterExec maps through its embedded projection, ProjectionExec substitutes the expression at each output position, and AggregateExec maps grouping outputs to the input column each grouping expression reads. `from_child_with_allowed_indices` is kept as a deprecated wrapper that resolves the allowed indices positionally. --- .../physical_optimizer/filter_pushdown.rs | 118 +++++++++++++ .../physical-plan/src/aggregates/mod.rs | 35 ++-- datafusion/physical-plan/src/filter.rs | 55 +++++- .../physical-plan/src/filter_pushdown.rs | 160 ++++++++---------- .../physical-plan/src/joins/hash_join/exec.rs | 11 +- datafusion/physical-plan/src/projection.rs | 49 +++--- .../physical-plan/src/repartition/mod.rs | 11 +- .../dynamic_filter_pushdown_config.slt | 113 ++++++++----- 8 files changed, 361 insertions(+), 191 deletions(-) diff --git a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs index b07603a1180b8..f621f2fd98d7b 100644 --- a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs +++ b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs @@ -1634,6 +1634,124 @@ fn test_repartition_filter_pushdown_preserves_duplicate_column_indices() { } } +/// Schema with two same-named columns, as produced by a nested join. +fn duplicate_id_schema() -> SchemaRef { + Arc::new(Schema::new(vec![ + Field::new("id", DataType::Utf8, false), + Field::new("id", DataType::Utf8, false), + ])) +} + +fn id_eq_x(index: usize) -> Arc { + Arc::new(BinaryExpr::new( + Arc::new(Column::new("id", index)), + Operator::Eq, + Arc::new(Literal::new(ScalarValue::from("x"))), + )) +} + +/// A filter with an embedded projection must map a parent predicate through +/// the projection by position, in both physical pushdown phases. +#[test] +fn test_filter_with_projection_pushdown_preserves_duplicate_column_indices() { + use datafusion_physical_plan::filter_pushdown::{FilterPushdownPhase, PushedDown}; + + let input = TestScanBuilder::new(duplicate_id_schema()).build(); + let filter = FilterExecBuilder::new(id_eq_x(0), input) + .apply_projection(Some(vec![1, 0])) + .unwrap() + .build() + .unwrap(); + for phase in [FilterPushdownPhase::Pre, FilterPushdownPhase::Post] { + let filters = filter + .gather_filters_for_pushdown( + phase, + vec![id_eq_x(0), id_eq_x(1)], + &ConfigOptions::default(), + ) + .unwrap() + .parent_filters(); + assert_eq!(filters.len(), 1); + assert!( + matches!(filters[0][0].discriminant, PushedDown::Yes), + "{phase}" + ); + assert!( + matches!(filters[0][1].discriminant, PushedDown::Yes), + "{phase}" + ); + assert_eq!(filters[0][0].predicate.to_string(), "id@1 = x", "{phase}"); + assert_eq!(filters[0][1].predicate.to_string(), "id@0 = x", "{phase}"); + } +} + +/// A projection whose outputs share an alias must expand a parent predicate +/// to the expression at that output position, not the first same-named one. +#[test] +fn test_projection_pushdown_preserves_duplicate_aliases() { + use datafusion_physical_plan::filter_pushdown::{FilterPushdownPhase, PushedDown}; + + let input = TestScanBuilder::new(duplicate_id_schema()).build(); + let projection = ProjectionExec::try_new( + vec![ + (Arc::new(Column::new("id", 1)) as _, "id".to_string()), + (Arc::new(Column::new("id", 0)) as _, "id".to_string()), + ], + input, + ) + .unwrap(); + let filters = projection + .gather_filters_for_pushdown( + FilterPushdownPhase::Pre, + vec![id_eq_x(0), id_eq_x(1), id_eq_x(2)], + &ConfigOptions::default(), + ) + .unwrap() + .parent_filters(); + assert_eq!(filters.len(), 1); + assert!(matches!(filters[0][0].discriminant, PushedDown::Yes)); + assert!(matches!(filters[0][1].discriminant, PushedDown::Yes)); + assert!(matches!(filters[0][2].discriminant, PushedDown::No)); + assert_eq!(filters[0][0].predicate.to_string(), "id@1 = x"); + assert_eq!(filters[0][1].predicate.to_string(), "id@0 = x"); +} + +/// An aggregate grouping on two same-named columns must map a parent +/// predicate on a grouping output to the input column that grouping reads. +#[test] +fn test_aggregate_pushdown_preserves_duplicate_grouping_columns() { + use datafusion_physical_plan::filter_pushdown::{FilterPushdownPhase, PushedDown}; + + let schema = duplicate_id_schema(); + let input = TestScanBuilder::new(Arc::clone(&schema)).build(); + let group_by = PhysicalGroupBy::new_single(vec![ + (Arc::new(Column::new("id", 1)) as _, "id".to_string()), + (Arc::new(Column::new("id", 0)) as _, "id".to_string()), + ]); + let aggregate = AggregateExec::try_new( + AggregateMode::Partial, + group_by, + vec![], + vec![], + input, + schema, + ) + .unwrap(); + let filters = aggregate + .gather_filters_for_pushdown( + FilterPushdownPhase::Pre, + vec![id_eq_x(0), id_eq_x(1)], + &ConfigOptions::default(), + ) + .unwrap() + .parent_filters(); + assert_eq!(filters.len(), 1); + assert!(matches!(filters[0][0].discriminant, PushedDown::Yes)); + assert!(matches!(filters[0][1].discriminant, PushedDown::Yes)); + assert_eq!(filters[0][0].predicate.to_string(), "id@1 = x"); + assert_eq!(filters[0][1].predicate.to_string(), "id@0 = x"); +} + /// A join's output projection must map to child positions even when a child /// contains multiple columns with the same name. #[test] diff --git a/datafusion/physical-plan/src/aggregates/mod.rs b/datafusion/physical-plan/src/aggregates/mod.rs index 3ed93e09ce4f4..314cfe310b3f1 100644 --- a/datafusion/physical-plan/src/aggregates/mod.rs +++ b/datafusion/physical-plan/src/aggregates/mod.rs @@ -178,7 +178,7 @@ use crate::{ }; use datafusion_common::config::ConfigOptions; use parking_lot::Mutex; -use std::collections::{HashMap, HashSet}; +use std::collections::HashMap; use arrow::array::{ArrayRef, UInt8Array, UInt16Array, UInt32Array, UInt64Array}; use arrow::datatypes::{DataType, Field, Schema, SchemaRef}; @@ -2268,13 +2268,27 @@ impl ExecutionPlan for AggregateExec { // the result of SUM or COUNT), as those require computing all groups first. // Grouping columns are output before aggregate columns, in the same order - // as the grouping expressions. A grouping-set null mask marks grouping - // columns that are not available in that set. - let mut allowed_indices: HashSet = - (0..self.group_by.expr().len()).collect(); - for null_mask in self.group_by.groups() { - allowed_indices.retain(|idx| null_mask.get(*idx) != Some(&true)); - } + // as the grouping expressions. Map each grouping output position to the + // input column it reads, by position rather than by name, so that + // same-named grouping columns stay distinct. Only grouping expressions + // that are plain input columns can be mapped; a grouping-set null mask + // marks grouping columns that are not available in that set. + let column_mapping: HashMap = self + .group_by + .expr() + .iter() + .enumerate() + .filter(|(idx, _)| { + self.group_by + .groups() + .iter() + .all(|null_mask| null_mask.get(*idx) != Some(&true)) + }) + .filter_map(|(idx, (expr, _))| { + expr.downcast_ref::() + .map(|column| (idx, column.index())) + }) + .collect(); let child = self.children()[0]; // Global aggregates and grouping sets containing an empty grouping set @@ -2290,9 +2304,9 @@ impl ExecutionPlan for AggregateExec { let mut child_desc = if may_emit_on_empty_input { ChildFilterDescription::all_unsupported(&parent_filters) } else { - ChildFilterDescription::from_child_with_allowed_indices( + ChildFilterDescription::from_child_with_column_mapping( &parent_filters, - allowed_indices, + column_mapping, child, )? }; @@ -3238,6 +3252,7 @@ pub fn evaluate_group_by( #[cfg(test)] mod tests { + use std::collections::HashSet; use std::task::{Context, Poll}; use super::*; diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs index 12771eec78470..e6e5600b79478 100644 --- a/datafusion/physical-plan/src/filter.rs +++ b/datafusion/physical-plan/src/filter.rs @@ -55,7 +55,7 @@ use arrow::record_batch::RecordBatch; use datafusion_common::cast::as_boolean_array; use datafusion_common::config::ConfigOptions; use datafusion_common::stats::Precision; -use datafusion_common::tree_node::TreeNodeRecursion; +use datafusion_common::tree_node::{Transformed, TreeNode, TreeNodeRecursion}; use datafusion_common::{ DataFusionError, Result, ScalarValue, internal_err, plan_err, project_schema, }; @@ -66,7 +66,7 @@ use datafusion_physical_expr::expressions::{ BinaryExpr, Column, InListExpr, IsNotNullExpr, Literal, lit, }; use datafusion_physical_expr::intervals::utils::check_support; -use datafusion_physical_expr::utils::{collect_columns, reassign_expr_columns}; +use datafusion_physical_expr::utils::collect_columns; use datafusion_physical_expr::{ AcrossPartitions, AnalysisContext, ConstExpr, ExprBoundaries, PhysicalExpr, analyze, conjunction, split_conjunction, @@ -316,6 +316,26 @@ impl FilterExec { self.default_selectivity } + /// Describe which parent filters (in this node's output coordinates) can + /// be forwarded to the input, remapped into input coordinates. + /// + /// With an embedded projection the output position `i` reads input column + /// `projection[i]`; without one the positions are identical. Mapping by + /// position keeps same-named input columns distinct. + fn parent_filters_for_input( + &self, + parent_filters: &[Arc], + ) -> Result { + match self.projection.as_ref() { + Some(projection) => ChildFilterDescription::from_child_with_column_mapping( + parent_filters, + projection.iter().copied().enumerate().collect(), + self.input(), + ), + None => ChildFilterDescription::from_child(parent_filters, self.input()), + } + } + /// Projection pub fn projection(&self) -> &Option { &self.projection @@ -698,12 +718,12 @@ impl ExecutionPlan for FilterExec { _config: &ConfigOptions, ) -> Result { if phase != FilterPushdownPhase::Pre { - let child = - ChildFilterDescription::from_child(&parent_filters, self.input())?; + let child = self.parent_filters_for_input(&parent_filters)?; return Ok(FilterDescription::new().with_child(child)); } - let child = ChildFilterDescription::from_child(&parent_filters, self.input())? + let child = self + .parent_filters_for_input(&parent_filters)? .with_self_filters( split_conjunction(&self.predicate) .into_iter() @@ -735,12 +755,31 @@ impl ExecutionPlan for FilterExec { // If this FilterExec has a projection, the unsupported parent filters // are in the output schema (after projection) coordinates. We need to - // remap them to the input schema coordinates before combining with self filters. - if self.projection.is_some() { + // remap them to the input schema coordinates before combining with self + // filters. Map by position through the projection: the input may + // contain several columns with the same name. + if let Some(projection) = self.projection.as_ref() { let input_schema = self.input().schema(); unsupported_parent_filters = unsupported_parent_filters .into_iter() - .map(|expr| reassign_expr_columns(expr, &input_schema)) + .map(|expr| { + expr.transform_down(|expr| { + if let Some(column) = expr.downcast_ref::() { + let Some(&index) = projection.get(column.index()) else { + return internal_err!( + "Parent filter column {column} is not in the FilterExec projection {projection:?}" + ); + }; + let field = input_schema.field(index); + return Ok(Transformed::yes(Arc::new(Column::new( + field.name(), + index, + )))); + } + Ok(Transformed::no(expr)) + }) + .map(|transformed| transformed.data) + }) .collect::>>()?; } diff --git a/datafusion/physical-plan/src/filter_pushdown.rs b/datafusion/physical-plan/src/filter_pushdown.rs index 0010dec07f9dc..830e4963a9b0b 100644 --- a/datafusion/physical-plan/src/filter_pushdown.rs +++ b/datafusion/physical-plan/src/filter_pushdown.rs @@ -315,48 +315,60 @@ pub struct ChildFilterDescription { /// Validates and remaps filter column references to a target schema in one step. /// /// When pushing filters from a parent to a child node, we need to: -/// 1. Verify that all columns referenced by the filter exist in the target -/// 2. Remap column indices to match the target schema +/// 1. Verify that every column referenced by the filter is reachable in the child +/// 2. Remap column indices to match the child schema /// -/// Use an explicit positional mapping when the parent can reorder columns or -/// the child can contain duplicate names, such as when pushing through a join. -pub(crate) enum FilterRemapper { - ByName { - child_schema: SchemaRef, - /// Reachable column indices in the parent schema. - allowed_indices: HashSet, - }, - ByIndex { - child_schema: SchemaRef, - /// Parent output index to child input index. Missing entries cannot - /// be pushed to this child. - column_mapping: HashMap, - }, +/// Columns are always resolved by position, never by name: a child schema can +/// contain several fields with the same name (for example the output of nested +/// joins), so a name lookup could silently redirect a predicate to the wrong +/// column. +pub(crate) struct FilterRemapper { + /// The target schema to remap column indices into. + child_schema: SchemaRef, + /// Parent output index to child input index. `None` maps every parent + /// index to the same child index, for nodes that preserve their input + /// schema. Parent columns absent from an explicit mapping cannot be + /// pushed to this child. + column_mapping: Option>, } impl FilterRemapper { - /// Create a remapper that accepts any column whose index falls within - /// `0..child_schema.len()` and whose name exists in the target schema. - pub(crate) fn new(child_schema: SchemaRef) -> Self { - let allowed_indices = (0..child_schema.fields().len()).collect(); - Self::ByName { + /// Create a remapper for a node whose output has the same column positions + /// as `child_schema`. Each column is resolved to the same index, provided + /// the child field at that position has the same name. + pub(crate) fn identity(child_schema: SchemaRef) -> Self { + Self { child_schema, - allowed_indices, + column_mapping: None, } } - /// Create a name-based remapper that only accepts columns at the given - /// parent indices. - fn with_allowed_indices( + /// Create a remapper with an explicit parent-output to child-input mapping. + fn with_column_mapping( child_schema: SchemaRef, - allowed_indices: HashSet, + column_mapping: HashMap, ) -> Self { - Self::ByName { + Self { child_schema, - allowed_indices, + column_mapping: Some(column_mapping), } } + /// Resolve a parent column to its position in the child schema. + fn remap_column(&self, col: &Column) -> Option { + let index = match &self.column_mapping { + None => col.index(), + Some(mapping) => *mapping.get(&col.index())?, + }; + let field = self.child_schema.fields().get(index)?; + // With an identity mapping the parent and child names must agree; + // a mismatch means the caller does not actually preserve positions. + if self.column_mapping.is_none() && field.name() != col.name() { + return None; + } + Some(Column::new(field.name(), index)) + } + /// Try to remap a filter's column references to the target schema. /// Returns `Some(remapped)` if all columns are reachable, or `None` if any /// column fails validation. @@ -367,26 +379,7 @@ impl FilterRemapper { let mut all_valid = true; let transformed = Arc::clone(filter).transform_down(|expr| { if let Some(col) = expr.downcast_ref::() { - let remapped = match self { - Self::ByName { - child_schema, - allowed_indices, - } => allowed_indices - .contains(&col.index()) - .then(|| child_schema.index_of(col.name()).ok()) - .flatten() - .map(|index| Column::new(col.name(), index)), - Self::ByIndex { - child_schema, - column_mapping, - } => column_mapping.get(&col.index()).and_then(|&index| { - child_schema - .fields() - .get(index) - .map(|field| Column::new(field.name(), index)) - }), - }; - if let Some(remapped) = remapped { + if let Some(remapped) = self.remap_column(col) { Ok(Transformed::yes(Arc::new(remapped))) } else { all_valid = false; @@ -402,69 +395,54 @@ impl FilterRemapper { } impl ChildFilterDescription { - /// Build a child filter description by analyzing which parent filters can be pushed to a specific child. + /// Build a child filter description for a node whose output has the same + /// column positions as `child`, such as a filter, sort or repartition. /// - /// 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 + /// Every column referenced by a filter is resolved at the same index in the + /// child schema, so same-named columns stay distinct. A filter is only + /// pushed down when all of its columns resolve. Nodes that project or + /// reorder columns must use [`Self::from_child_with_column_mapping`]. /// /// See [`FilterDescription::from_children`] for more details pub fn from_child( parent_filters: &[Arc], child: &Arc, ) -> Result { - // Building the remapper indexes every column of the child's schema, so - // with no filters to remap it is pure cost for the empty description - // `remap_filters` would return anyway. On a wide schema, and once per - // child, that is worth not paying. if parent_filters.is_empty() { return Ok(Self::empty()); } - let remapper = FilterRemapper::new(child.schema()); + let remapper = FilterRemapper::identity(child.schema()); Self::remap_filters(parent_filters, &remapper) } - /// Forward parent filters through a node that preserves column positions. - /// Columns are resolved at the same index in the child schema, even if - /// multiple fields share a name. Nodes that project or reorder columns - /// must instead provide an explicit mapping. - pub fn from_child_preserving_indices( - parent_filters: &[Arc], - child: &Arc, - ) -> Result { - if parent_filters.is_empty() { - return Ok(Self::empty()); - } - let column_mapping = (0..child.schema().fields().len()) - .map(|index| (index, index)) - .collect(); - Self::from_child_with_column_mapping(parent_filters, column_mapping, child) - } - - /// Like [`Self::from_child`], but restricts which parent-level columns are - /// considered reachable through this child. All columns in a filter must - /// appear in `allowed_indices`; their child positions are resolved by name. + /// Like [`Self::from_child`], but only forwards filters whose columns all + /// appear in `allowed_indices`. /// - /// Use [`Self::from_child_with_column_mapping`] when the child can contain - /// duplicate names or when column identity must be preserved across a join. + /// Columns are resolved at the same position in the child schema. Earlier + /// versions resolved them by name, which is ambiguous when the child has + /// duplicate field names. Nodes whose output positions differ from the + /// child's must use [`Self::from_child_with_column_mapping`] instead. + #[deprecated( + since = "55.0.0", + note = "use `from_child` or `from_child_with_column_mapping`" + )] pub fn from_child_with_allowed_indices( parent_filters: &[Arc], allowed_indices: HashSet, child: &Arc, ) -> Result { - // See [`Self::from_child`]: nothing to remap, nothing to index. - if parent_filters.is_empty() { - return Ok(Self::empty()); - } - let remapper = - FilterRemapper::with_allowed_indices(child.schema(), allowed_indices); - Self::remap_filters(parent_filters, &remapper) + let column_mapping = allowed_indices.into_iter().map(|i| (i, i)).collect(); + Self::from_child_with_column_mapping(parent_filters, column_mapping, child) } /// Remap parent filters using an explicit parent-output to child-input /// column mapping. Columns absent from the mapping cannot be pushed down. - /// Unlike name-based remapping, this preserves column identity when a child - /// has duplicate field names, and supports differently named join keys. + /// + /// Joins, aggregates and filters with an embedded projection use this: + /// their output positions differ from the child's, and a child can contain + /// duplicate field names, so positions cannot be recovered from names. + /// Join keys may also be mapped to a differently named column on the + /// other side. pub fn from_child_with_column_mapping( parent_filters: &[Arc], column_mapping: HashMap, @@ -473,10 +451,8 @@ impl ChildFilterDescription { if parent_filters.is_empty() { return Ok(Self::empty()); } - let remapper = FilterRemapper::ByIndex { - child_schema: child.schema(), - column_mapping, - }; + let remapper = + FilterRemapper::with_column_mapping(child.schema(), column_mapping); Self::remap_filters(parent_filters, &remapper) } diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index 12461ec106b5f..6af48cb1fe018 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -1819,15 +1819,14 @@ impl ExecutionPlan for HashJoinExec { // 1. `lr_is_preserved` gates whether a side is eligible at all. // 2. For each filter, we check that all column references belong to the // target child (using `column_indices` to map output column positions - // to join sides). This is critical for correctness: name-based matching - // alone (as done by `ChildFilterDescription::from_child`) can incorrectly - // push filters when different join sides have columns with the same name - // (e.g. nested mark joins both producing "mark" columns). + // to join sides). Columns are mapped by position, never by name: + // different join sides, or a nested join on one side, can produce + // columns with the same name (e.g. nested mark joins both producing + // "mark" columns, or several `id` columns). let (left_preserved, right_preserved) = lr_is_preserved(self.join_type); // Map each output position to its input position, accounting for the - // join's projection. Looking up child columns by name is ambiguous when - // a nested join produces multiple fields with the same name. + // join's projection. let column_indices: Vec = match self.projection.as_ref() { Some(projection) => projection .iter() diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs index 7851e5705934e..ee60a1a79aed5 100644 --- a/datafusion/physical-plan/src/projection.rs +++ b/datafusion/physical-plan/src/projection.rs @@ -49,7 +49,7 @@ use datafusion_common::config::ConfigOptions; use datafusion_common::tree_node::{ Transformed, TransformedResult, TreeNode, TreeNodeRecursion, }; -use datafusion_common::{DataFusionError, JoinSide, Result, internal_err, plan_err}; +use datafusion_common::{JoinSide, Result, internal_err, plan_err}; use datafusion_execution::TaskContext; use datafusion_expr::ExpressionPlacement; use datafusion_physical_expr::EquivalenceProperties; @@ -305,27 +305,23 @@ impl ProjectionExec { self.overrides_metadata } - /// Collect reverse alias mapping from projection expressions. - /// The result hash map is a map from aliased Column in parent to original expr. - fn collect_reverse_alias( + /// Map each output column, by position, to the expression that produces + /// it. The result is a map from the aliased `Column` in the parent to the + /// original expression. Output aliases are not unique, so this must not + /// go through the output schema by name. + fn output_column_to_expr( &self, - ) -> Result>> { - let mut alias_map = datafusion_common::HashMap::new(); - for projection in self.projection_expr().iter() { - let (aliased_index, _output_field) = self - .projector - .output_schema() - .column_with_name(&projection.alias) - .ok_or_else(|| { - DataFusionError::Internal(format!( - "Expr {} with alias {} not found in output schema", - projection.expr, projection.alias - )) - })?; - let aliased_col = Column::new(&projection.alias, aliased_index); - alias_map.insert(aliased_col, Arc::clone(&projection.expr)); - } - Ok(alias_map) + ) -> datafusion_common::HashMap> { + self.projection_expr() + .iter() + .enumerate() + .map(|(index, projection)| { + ( + Column::new(&projection.alias, index), + Arc::clone(&projection.expr), + ) + }) + .collect() } } @@ -545,16 +541,17 @@ impl ExecutionPlan for ProjectionExec { _config: &ConfigOptions, ) -> Result { // expand alias column to original expr in parent filters - let invert_alias_map = self.collect_reverse_alias()?; + let output_column_map = self.output_column_to_expr(); let output_schema = self.schema(); - let remapper = FilterRemapper::new(output_schema); + let remapper = FilterRemapper::identity(output_schema); let mut child_parent_filters = Vec::with_capacity(parent_filters.len()); for filter in parent_filters { - // Check that column exists in child, then reassign column indices to match child schema + // Check that every column is a valid output column of this + // projection, then replace each one with the expression at that + // output position. if let Some(reassigned) = remapper.try_remap(&filter)? { - // rewrite filter expression using invert alias map - let mut rewriter = PhysicalColumnRewriter::new(&invert_alias_map); + let mut rewriter = PhysicalColumnRewriter::new(&output_column_map); let rewritten = reassigned.rewrite(&mut rewriter)?.data; child_parent_filters.push(PushedDownPredicate::supported(rewritten)); } else { diff --git a/datafusion/physical-plan/src/repartition/mod.rs b/datafusion/physical-plan/src/repartition/mod.rs index 68fe014f872a1..b653a74fa5d27 100644 --- a/datafusion/physical-plan/src/repartition/mod.rs +++ b/datafusion/physical-plan/src/repartition/mod.rs @@ -75,7 +75,7 @@ use datafusion_physical_expr_common::sort_expr::{ use datafusion_proto_models::protobuf; use crate::filter_pushdown::{ - ChildFilterDescription, ChildPushdownResult, FilterDescription, FilterPushdownPhase, + ChildPushdownResult, FilterDescription, FilterPushdownPhase, FilterPushdownPropagation, }; use crate::joins::SeededRandomState; @@ -2003,14 +2003,7 @@ impl ExecutionPlan for RepartitionExec { parent_filters: Vec>, _config: &ConfigOptions, ) -> Result { - // Repartition changes row placement, not column positions. Preserve - // indices so a nested join's same-named columns remain distinct. - Ok(FilterDescription::new().with_child( - ChildFilterDescription::from_child_preserving_indices( - &parent_filters, - self.input(), - )?, - )) + FilterDescription::from_children(parent_filters, &self.children()) } fn handle_child_pushdown_result( diff --git a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt index fcfccd7966215..66287f611a028 100644 --- a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt +++ b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt @@ -906,18 +906,11 @@ statement ok RESET datafusion.execution.parquet.max_row_group_size; # Regression for https://github.com/apache/datafusion/issues/25244. -# Keep both a.id and b.id in the nested join output. A dynamic filter on b.id -# must not be remapped by name to a.id, which has a different value. LEFT JOINs -# preserve this join tree through logical optimization. -statement ok -SET datafusion.execution.target_partitions = 1; - -statement ok -SET datafusion.execution.collect_statistics = true; - -statement ok -SET datafusion.execution.parquet.schema_force_view_types = false; - +# Pushing a join dynamic filter through operators whose schema contains +# duplicate column names must map columns by position, not by name. Each +# query below keeps both a.id and b.id (and sometimes c.id) in the probe-side +# output; with dynamic filtering enabled the filter on b.id must not be +# redirected to a.id, which holds a different value and would drop the row. statement ok SET datafusion.optimizer.join_reordering = false; @@ -945,9 +938,12 @@ CREATE EXTERNAL TABLE issue_25244_b STORED AS PARQUET LOCATION 'test_files/scratch/dynamic_filter_pushdown_config/issue_25244_b.parquet'; +# Expected results, with join dynamic filtering disabled. statement ok SET datafusion.optimizer.enable_join_dynamic_filter_pushdown = false; +# 1. Nested joins, with a RepartitionExec between them at four partitions. +# LEFT JOINs keep this join tree through logical optimization. query TTTT SELECT s.id AS sid, a.id AS aid, b.id AS bid, c.id AS cid FROM issue_25244_b s @@ -959,28 +955,51 @@ ON s.id = b.id; ---- x1 a1 x1 x1 -statement ok -SET datafusion.optimizer.enable_join_dynamic_filter_pushdown = true; +# 2. A FilterExec with an embedded projection. +query TTT +SELECT s.id AS sid, a.id AS aid, b.id AS bid +FROM issue_25244_b s +JOIN ( + SELECT a.id, b.id + FROM issue_25244_a a LEFT JOIN issue_25244_b b ON a.ty = b.id + WHERE (a.ty || '!') IS DISTINCT FROM b.id +) +ON s.id = b.id; +---- +x1 a1 x1 +# 3. A ProjectionExec whose output has two columns aliased `id`. query TTTT -SELECT s.id AS sid, a.id AS aid, b.id AS bid, c.id AS cid +SELECT s.id AS sid, a.id AS aid, b.id AS bid, tag +FROM issue_25244_a s +JOIN ( + SELECT a.id, b.id, a.ty || '!' AS tag + FROM issue_25244_a a JOIN issue_25244_b b ON a.ty = b.id +) +ON s.id = a.id; +---- +a1 a1 x1 x1! + +# 4. An AggregateExec grouping on two same-named columns. +query TTT +SELECT s.id AS sid, a.id AS aid, b.id AS bid FROM issue_25244_b s JOIN ( - (issue_25244_a a LEFT JOIN issue_25244_b b ON a.ty = b.id) - LEFT JOIN issue_25244_b c ON b.id = c.id + SELECT a.id, b.id + FROM issue_25244_a a LEFT JOIN issue_25244_b b ON a.ty = b.id + GROUP BY a.id, b.id ) ON s.id = b.id; ---- -x1 a1 x1 x1 - -# Parallel execution inserts RepartitionExec above the nested joins. It must -# preserve the second id's position when forwarding the dynamic filter. -statement ok -SET datafusion.execution.target_partitions = 4; +x1 a1 x1 +# The same queries must return the same rows with join dynamic filtering +# enabled. statement ok -SET datafusion.optimizer.enable_join_dynamic_filter_pushdown = false; +SET datafusion.optimizer.enable_join_dynamic_filter_pushdown = true; +# 1. Nested joins, with a RepartitionExec between them at four partitions. +# LEFT JOINs keep this join tree through logical optimization. query TTTT SELECT s.id AS sid, a.id AS aid, b.id AS bid, c.id AS cid FROM issue_25244_b s @@ -992,19 +1011,43 @@ ON s.id = b.id; ---- x1 a1 x1 x1 -statement ok -SET datafusion.optimizer.enable_join_dynamic_filter_pushdown = true; +# 2. A FilterExec with an embedded projection. +query TTT +SELECT s.id AS sid, a.id AS aid, b.id AS bid +FROM issue_25244_b s +JOIN ( + SELECT a.id, b.id + FROM issue_25244_a a LEFT JOIN issue_25244_b b ON a.ty = b.id + WHERE (a.ty || '!') IS DISTINCT FROM b.id +) +ON s.id = b.id; +---- +x1 a1 x1 +# 3. A ProjectionExec whose output has two columns aliased `id`. query TTTT -SELECT s.id AS sid, a.id AS aid, b.id AS bid, c.id AS cid +SELECT s.id AS sid, a.id AS aid, b.id AS bid, tag +FROM issue_25244_a s +JOIN ( + SELECT a.id, b.id, a.ty || '!' AS tag + FROM issue_25244_a a JOIN issue_25244_b b ON a.ty = b.id +) +ON s.id = a.id; +---- +a1 a1 x1 x1! + +# 4. An AggregateExec grouping on two same-named columns. +query TTT +SELECT s.id AS sid, a.id AS aid, b.id AS bid FROM issue_25244_b s JOIN ( - (issue_25244_a a LEFT JOIN issue_25244_b b ON a.ty = b.id) - LEFT JOIN issue_25244_b c ON b.id = c.id + SELECT a.id, b.id + FROM issue_25244_a a LEFT JOIN issue_25244_b b ON a.ty = b.id + GROUP BY a.id, b.id ) ON s.id = b.id; ---- -x1 a1 x1 x1 +x1 a1 x1 statement ok DROP TABLE issue_25244_a; @@ -1012,15 +1055,5 @@ DROP TABLE issue_25244_a; statement ok DROP TABLE issue_25244_b; -# The SLT runner uses four target partitions. -statement ok -SET datafusion.execution.target_partitions = 4; - -statement ok -RESET datafusion.execution.collect_statistics; - -statement ok -RESET datafusion.execution.parquet.schema_force_view_types; - statement ok RESET datafusion.optimizer.join_reordering; From 05f3afdcd4046fd0fa3123276ef19f23e40fc3fb Mon Sep 17 00:00:00 2001 From: Huaijin Date: Sun, 13 Sep 2026 21:04:14 +0800 Subject: [PATCH 3/8] fix: check join side for semi join key mapping and test deprecated API --- .../physical_optimizer/filter_pushdown.rs | 22 +++++++++++++++++++ .../physical-plan/src/joins/hash_join/exec.rs | 10 +++++---- 2 files changed, 28 insertions(+), 4 deletions(-) diff --git a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs index f621f2fd98d7b..ea8806bcc8b29 100644 --- a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs +++ b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs @@ -1752,6 +1752,28 @@ fn test_aggregate_pushdown_preserves_duplicate_grouping_columns() { assert_eq!(filters[0][1].predicate.to_string(), "id@0 = x"); } +/// The deprecated allowed-indices API resolves columns by position too. +#[test] +#[expect(deprecated)] +fn test_from_child_with_allowed_indices_resolves_by_position() { + use datafusion_physical_plan::filter_pushdown::{ + ChildFilterDescription, FilterDescription, PushedDown, + }; + use std::collections::HashSet; + + let input = TestScanBuilder::new(duplicate_id_schema()).build(); + let child = ChildFilterDescription::from_child_with_allowed_indices( + &[id_eq_x(0), id_eq_x(1)], + HashSet::from([1]), + &input, + ) + .unwrap(); + let filters = FilterDescription::new().with_child(child).parent_filters(); + assert!(matches!(filters[0][0].discriminant, PushedDown::No)); + assert!(matches!(filters[0][1].discriminant, PushedDown::Yes)); + assert_eq!(filters[0][1].predicate.to_string(), "id@1 = x"); +} + /// A join's output projection must map to child positions even when a child /// contains multiple columns with the same name. #[test] diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index 6af48cb1fe018..7a2bf39002977 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -1866,12 +1866,14 @@ impl ExecutionPlan for HashJoinExec { }) }) .collect(); - let other_mapping = match self.join_type { - JoinType::LeftSemi => &mut right_mapping, - _ => &mut left_mapping, + let (output_side, other_mapping) = match self.join_type { + JoinType::LeftSemi => (JoinSide::Left, &mut right_mapping), + _ => (JoinSide::Right, &mut left_mapping), }; for (output_idx, ci) in column_indices.iter().enumerate() { - if let Some(&input_idx) = key_mapping.get(&ci.index) { + if ci.side == output_side + && let Some(&input_idx) = key_mapping.get(&ci.index) + { other_mapping.insert(output_idx, input_idx); } } From fa35c79355e696504134161ac26d470bbd02126b Mon Sep 17 00:00:00 2001 From: Huaijin Date: Sun, 13 Sep 2026 21:16:41 +0800 Subject: [PATCH 4/8] test: cover remaining filter remapping branches --- .../physical_optimizer/filter_pushdown.rs | 91 +++++++++++++++++++ datafusion/physical-plan/src/filter.rs | 25 +++-- .../physical-plan/src/filter_pushdown.rs | 4 +- .../physical-plan/src/joins/hash_join/exec.rs | 10 +- datafusion/physical-plan/src/projection.rs | 2 +- 5 files changed, 111 insertions(+), 21 deletions(-) diff --git a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs index ea8806bcc8b29..5fa52ab888fdf 100644 --- a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs +++ b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs @@ -1752,6 +1752,97 @@ fn test_aggregate_pushdown_preserves_duplicate_grouping_columns() { assert_eq!(filters[0][1].predicate.to_string(), "id@0 = x"); } +/// A semi join key that is not a plain column cannot be mapped to the other +/// side, so a filter on that output key only reaches the emitted side. +#[test] +fn test_hashjoin_parent_filter_pushdown_semi_join_expression_key() { + use datafusion_physical_expr::expressions::CastExpr; + use datafusion_physical_plan::filter_pushdown::{FilterPushdownPhase, PushedDown}; + + let schema = duplicate_id_schema(); + let key = || Arc::new(Column::new("id", 0)) as Arc; + let cast_key = + || Arc::new(CastExpr::new(key(), DataType::Utf8, None)) as Arc; + for on in [(cast_key(), key()), (key(), cast_key())] { + let join = HashJoinExec::try_new( + TestScanBuilder::new(Arc::clone(&schema)).build(), + TestScanBuilder::new(Arc::clone(&schema)).build(), + vec![on], + None, + &JoinType::LeftSemi, + None, + PartitionMode::Partitioned, + datafusion_common::NullEquality::NullEqualsNothing, + false, + ) + .unwrap(); + let filters = join + .gather_filters_for_pushdown( + FilterPushdownPhase::Pre, + vec![id_eq_x(0)], + &ConfigOptions::default(), + ) + .unwrap() + .parent_filters(); + assert!(matches!(filters[0][0].discriminant, PushedDown::Yes)); + assert!(matches!(filters[1][0].discriminant, PushedDown::No)); + assert_eq!(filters[0][0].predicate.to_string(), "id@0 = x"); + } +} + +/// The identity mapping used by schema-preserving nodes rejects a column +/// whose name differs from the child field at the same position. +#[test] +fn test_from_child_rejects_column_name_mismatch() { + use datafusion_physical_plan::filter_pushdown::{ + ChildFilterDescription, FilterDescription, PushedDown, + }; + + let input = TestScanBuilder::new(duplicate_id_schema()).build(); + let mismatched: Arc = Arc::new(BinaryExpr::new( + Arc::new(Column::new("other", 0)), + Operator::Eq, + Arc::new(Literal::new(ScalarValue::from("x"))), + )); + let child = + ChildFilterDescription::from_child(&[mismatched, id_eq_x(0)], &input).unwrap(); + let filters = FilterDescription::new().with_child(child).parent_filters(); + assert!(matches!(filters[0][0].discriminant, PushedDown::No)); + assert!(matches!(filters[0][1].discriminant, PushedDown::Yes)); +} + +/// An unsupported parent filter folded back into a projected FilterExec must +/// reference an output column of the projection. +#[test] +fn test_filter_with_projection_rejects_out_of_range_parent_filter() { + use datafusion_physical_plan::filter_pushdown::{ + ChildFilterPushdownResult, ChildPushdownResult, FilterPushdownPhase, PushedDown, + }; + + let input = TestScanBuilder::new(duplicate_id_schema()).build(); + let filter = FilterExecBuilder::new(id_eq_x(0), input) + .apply_projection(Some(vec![1])) + .unwrap() + .build() + .unwrap(); + let result = filter.handle_child_pushdown_result( + FilterPushdownPhase::Pre, + ChildPushdownResult { + parent_filters: vec![ChildFilterPushdownResult { + filter: id_eq_x(1), + child_results: vec![PushedDown::No], + }], + self_filters: vec![vec![]], + }, + &ConfigOptions::default(), + ); + let err = result.unwrap_err().to_string(); + assert!( + err.contains("is not in the FilterExec projection"), + "unexpected error: {err}" + ); +} + /// The deprecated allowed-indices API resolves columns by position too. #[test] #[expect(deprecated)] diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs index e6e5600b79478..485c85c9afc58 100644 --- a/datafusion/physical-plan/src/filter.rs +++ b/datafusion/physical-plan/src/filter.rs @@ -717,21 +717,18 @@ impl ExecutionPlan for FilterExec { parent_filters: Vec>, _config: &ConfigOptions, ) -> Result { - if phase != FilterPushdownPhase::Pre { - let child = self.parent_filters_for_input(&parent_filters)?; - return Ok(FilterDescription::new().with_child(child)); + let mut child = self.parent_filters_for_input(&parent_filters); + if phase == FilterPushdownPhase::Pre { + child = child.map(|child| { + child.with_self_filters( + split_conjunction(&self.predicate) + .into_iter() + .cloned() + .collect(), + ) + }); } - - let child = self - .parent_filters_for_input(&parent_filters)? - .with_self_filters( - split_conjunction(&self.predicate) - .into_iter() - .cloned() - .collect(), - ); - - Ok(FilterDescription::new().with_child(child)) + child.map(|child| FilterDescription::new().with_child(child)) } fn handle_child_pushdown_result( diff --git a/datafusion/physical-plan/src/filter_pushdown.rs b/datafusion/physical-plan/src/filter_pushdown.rs index 830e4963a9b0b..453ee7dfe94a0 100644 --- a/datafusion/physical-plan/src/filter_pushdown.rs +++ b/datafusion/physical-plan/src/filter_pushdown.rs @@ -336,7 +336,7 @@ impl FilterRemapper { /// Create a remapper for a node whose output has the same column positions /// as `child_schema`. Each column is resolved to the same index, provided /// the child field at that position has the same name. - pub(crate) fn identity(child_schema: SchemaRef) -> Self { + pub(crate) fn new(child_schema: SchemaRef) -> Self { Self { child_schema, column_mapping: None, @@ -411,7 +411,7 @@ impl ChildFilterDescription { if parent_filters.is_empty() { return Ok(Self::empty()); } - let remapper = FilterRemapper::identity(child.schema()); + let remapper = FilterRemapper::new(child.schema()); Self::remap_filters(parent_filters, &remapper) } diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index 7a2bf39002977..088376bc349c1 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -1870,10 +1870,12 @@ impl ExecutionPlan for HashJoinExec { JoinType::LeftSemi => (JoinSide::Left, &mut right_mapping), _ => (JoinSide::Right, &mut left_mapping), }; - for (output_idx, ci) in column_indices.iter().enumerate() { - if ci.side == output_side - && let Some(&input_idx) = key_mapping.get(&ci.index) - { + let emitted = column_indices + .iter() + .enumerate() + .filter(|(_, ci)| ci.side == output_side); + for (output_idx, ci) in emitted { + if let Some(&input_idx) = key_mapping.get(&ci.index) { other_mapping.insert(output_idx, input_idx); } } diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs index ee60a1a79aed5..d93854b9f6489 100644 --- a/datafusion/physical-plan/src/projection.rs +++ b/datafusion/physical-plan/src/projection.rs @@ -543,7 +543,7 @@ impl ExecutionPlan for ProjectionExec { // expand alias column to original expr in parent filters let output_column_map = self.output_column_to_expr(); let output_schema = self.schema(); - let remapper = FilterRemapper::identity(output_schema); + let remapper = FilterRemapper::new(output_schema); let mut child_parent_filters = Vec::with_capacity(parent_filters.len()); for filter in parent_filters { From d6db6a4d2408e98854dcab5e71640e54fdc8d2f1 Mon Sep 17 00:00:00 2001 From: Huaijin Date: Sun, 13 Sep 2026 22:48:35 +0800 Subject: [PATCH 5/8] update --- datafusion/physical-plan/src/filter.rs | 38 ++++++++-------- .../physical-plan/src/filter_pushdown.rs | 8 ++-- datafusion/physical-plan/src/projection.rs | 24 +--------- .../library-user-guide/upgrading/56.0.0.md | 45 +++++++++++++++++++ 4 files changed, 68 insertions(+), 47 deletions(-) diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs index 485c85c9afc58..ae87168ec7598 100644 --- a/datafusion/physical-plan/src/filter.rs +++ b/datafusion/physical-plan/src/filter.rs @@ -33,7 +33,7 @@ use crate::common::can_project; use crate::execution_plan::{CardinalityEffect, replace_children_if_necessary}; use crate::filter_pushdown::{ ChildFilterDescription, ChildPushdownResult, FilterDescription, FilterPushdownPhase, - FilterPushdownPropagation, PushedDown, + FilterPushdownPropagation, FilterRemapper, PushedDown, }; use crate::limit::LocalLimitExec; use crate::metrics::{MetricBuilder, MetricType}; @@ -55,9 +55,10 @@ use arrow::record_batch::RecordBatch; use datafusion_common::cast::as_boolean_array; use datafusion_common::config::ConfigOptions; use datafusion_common::stats::Precision; -use datafusion_common::tree_node::{Transformed, TreeNode, TreeNodeRecursion}; +use datafusion_common::tree_node::TreeNodeRecursion; use datafusion_common::{ - DataFusionError, Result, ScalarValue, internal_err, plan_err, project_schema, + DataFusionError, Result, ScalarValue, internal_datafusion_err, internal_err, + plan_err, project_schema, }; use datafusion_execution::TaskContext; use datafusion_expr::Operator; @@ -326,6 +327,9 @@ impl FilterExec { &self, parent_filters: &[Arc], ) -> Result { + if parent_filters.is_empty() { + return Ok(ChildFilterDescription::empty()); + } match self.projection.as_ref() { Some(projection) => ChildFilterDescription::from_child_with_column_mapping( parent_filters, @@ -755,27 +759,21 @@ impl ExecutionPlan for FilterExec { // remap them to the input schema coordinates before combining with self // filters. Map by position through the projection: the input may // contain several columns with the same name. - if let Some(projection) = self.projection.as_ref() { - let input_schema = self.input().schema(); + if let Some(projection) = self.projection.as_ref() + && !unsupported_parent_filters.is_empty() + { + let remapper = FilterRemapper::with_column_mapping( + self.input().schema(), + projection.iter().copied().enumerate().collect(), + ); unsupported_parent_filters = unsupported_parent_filters .into_iter() .map(|expr| { - expr.transform_down(|expr| { - if let Some(column) = expr.downcast_ref::() { - let Some(&index) = projection.get(column.index()) else { - return internal_err!( - "Parent filter column {column} is not in the FilterExec projection {projection:?}" - ); - }; - let field = input_schema.field(index); - return Ok(Transformed::yes(Arc::new(Column::new( - field.name(), - index, - )))); - } - Ok(Transformed::no(expr)) + remapper.try_remap(&expr)?.ok_or_else(|| { + internal_datafusion_err!( + "Parent filter {expr} references a column that is not in the FilterExec projection {projection:?}" + ) }) - .map(|transformed| transformed.data) }) .collect::>>()?; } diff --git a/datafusion/physical-plan/src/filter_pushdown.rs b/datafusion/physical-plan/src/filter_pushdown.rs index 453ee7dfe94a0..d457d9562e25b 100644 --- a/datafusion/physical-plan/src/filter_pushdown.rs +++ b/datafusion/physical-plan/src/filter_pushdown.rs @@ -344,7 +344,7 @@ impl FilterRemapper { } /// Create a remapper with an explicit parent-output to child-input mapping. - fn with_column_mapping( + pub(crate) fn with_column_mapping( child_schema: SchemaRef, column_mapping: HashMap, ) -> Self { @@ -423,8 +423,8 @@ impl ChildFilterDescription { /// duplicate field names. Nodes whose output positions differ from the /// child's must use [`Self::from_child_with_column_mapping`] instead. #[deprecated( - since = "55.0.0", - note = "use `from_child` or `from_child_with_column_mapping`" + since = "56.0.0", + note = "columns now resolve by position; use `from_child` for matching schemas or `from_child_with_column_mapping` when positions differ" )] pub fn from_child_with_allowed_indices( parent_filters: &[Arc], @@ -477,7 +477,7 @@ impl ChildFilterDescription { } /// A description carrying no filters in either direction. - fn empty() -> Self { + pub(crate) fn empty() -> Self { Self { parent_filters: vec![], self_filters: vec![], diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs index d93854b9f6489..9ec9b2ab1d786 100644 --- a/datafusion/physical-plan/src/projection.rs +++ b/datafusion/physical-plan/src/projection.rs @@ -26,7 +26,6 @@ use super::{ DisplayAs, ExecutionPlanProperties, PlanProperties, RecordBatchStream, SendableRecordBatchStream, SortOrderPushdownResult, Statistics, }; -use crate::column_rewriter::PhysicalColumnRewriter; use crate::execution_plan::{CardinalityEffect, replace_children_if_necessary}; use crate::filter_pushdown::{ ChildFilterDescription, ChildPushdownResult, FilterDescription, FilterPushdownPhase, @@ -304,25 +303,6 @@ impl ProjectionExec { fn overrides_metadata(&self) -> bool { self.overrides_metadata } - - /// Map each output column, by position, to the expression that produces - /// it. The result is a map from the aliased `Column` in the parent to the - /// original expression. Output aliases are not unique, so this must not - /// go through the output schema by name. - fn output_column_to_expr( - &self, - ) -> datafusion_common::HashMap> { - self.projection_expr() - .iter() - .enumerate() - .map(|(index, projection)| { - ( - Column::new(&projection.alias, index), - Arc::clone(&projection.expr), - ) - }) - .collect() - } } impl DisplayAs for ProjectionExec { @@ -541,7 +521,6 @@ impl ExecutionPlan for ProjectionExec { _config: &ConfigOptions, ) -> Result { // expand alias column to original expr in parent filters - let output_column_map = self.output_column_to_expr(); let output_schema = self.schema(); let remapper = FilterRemapper::new(output_schema); let mut child_parent_filters = Vec::with_capacity(parent_filters.len()); @@ -551,8 +530,7 @@ impl ExecutionPlan for ProjectionExec { // projection, then replace each one with the expression at that // output position. if let Some(reassigned) = remapper.try_remap(&filter)? { - let mut rewriter = PhysicalColumnRewriter::new(&output_column_map); - let rewritten = reassigned.rewrite(&mut rewriter)?.data; + let rewritten = self.projection_expr().unproject_expr(&reassigned)?; child_parent_filters.push(PushedDownPredicate::supported(rewritten)); } else { child_parent_filters.push(PushedDownPredicate::unsupported(filter)); diff --git a/docs/source/library-user-guide/upgrading/56.0.0.md b/docs/source/library-user-guide/upgrading/56.0.0.md index 8de2abc5a05fc..3434e92214348 100644 --- a/docs/source/library-user-guide/upgrading/56.0.0.md +++ b/docs/source/library-user-guide/upgrading/56.0.0.md @@ -277,3 +277,48 @@ let df = DataFrame::from_columns([ Most existing call sites using `Vec` require no changes. Code that relies on the exact non-generic function signature of `DataFrame::from_columns` may need to be updated to account for the new generic API. + +### Physical filter pushdown resolves columns by position + +`datafusion_physical_plan::filter_pushdown::ChildFilterDescription::from_child` +and `FilterDescription::from_children` now resolve filter columns by position +instead of looking them up by name. This prevents incorrect results when a +child schema contains duplicate column names. `from_child` requires the child +field at each referenced position to have the same name as the filter column. + +`ChildFilterDescription::from_child_with_allowed_indices` is deprecated and +also resolves columns by position. Its allowed indices now map to the same +indices in the child schema; it no longer preserves the previous name-based +mapping. Existing callers with different parent and child column positions +must migrate, even if all column names are unique, or filters may reference +the wrong child column. + +**Migration guide:** + +Use `from_child` (or `from_children` for multiple children) when the parent and +child schemas have matching column positions and names. When a node projects, +reorders, or renames columns, use `from_child_with_column_mapping` with an +explicit map from parent output indices to child input indices. Columns absent +from the mapping cannot be pushed down. + +For example, if the parent outputs `[a, b]` and the child outputs `[b, a]`, a +filter on `a@0` must map to child column `a@1`: + +```rust,ignore +use std::collections::{HashMap, HashSet}; +use datafusion_physical_plan::filter_pushdown::ChildFilterDescription; + +// Before: allow parent column 0 and resolve "a" by name in the child. +let description = ChildFilterDescription::from_child_with_allowed_indices( + &parent_filters, + HashSet::from([0]), + &child, +)?; + +// After: explicitly map parent column 0 to child column 1. +let description = ChildFilterDescription::from_child_with_column_mapping( + &parent_filters, + HashMap::from([(0, 1)]), + &child, +)?; +``` From 93e08cbe019049ad879cee50fe2c799884ce660a Mon Sep 17 00:00:00 2001 From: Huaijin Date: Mon, 14 Sep 2026 22:56:56 +0800 Subject: [PATCH 6/8] apply suggestion --- .../physical_optimizer/filter_pushdown.rs | 39 +++++++++-- .../physical-plan/src/filter_pushdown.rs | 68 ++++++++++++------- .../library-user-guide/upgrading/56.0.0.md | 10 ++- 3 files changed, 83 insertions(+), 34 deletions(-) diff --git a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs index 5fa52ab888fdf..f559d33a5e4b0 100644 --- a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs +++ b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs @@ -1843,10 +1843,11 @@ fn test_filter_with_projection_rejects_out_of_range_parent_filter() { ); } -/// The deprecated allowed-indices API resolves columns by position too. +/// The deprecated API preserves name lookup, including the first matching +/// field for duplicate names and parent indices outside the child schema. #[test] #[expect(deprecated)] -fn test_from_child_with_allowed_indices_resolves_by_position() { +fn test_from_child_with_allowed_indices_preserves_name_resolution() { use datafusion_physical_plan::filter_pushdown::{ ChildFilterDescription, FilterDescription, PushedDown, }; @@ -1854,15 +1855,43 @@ fn test_from_child_with_allowed_indices_resolves_by_position() { let input = TestScanBuilder::new(duplicate_id_schema()).build(); let child = ChildFilterDescription::from_child_with_allowed_indices( - &[id_eq_x(0), id_eq_x(1)], - HashSet::from([1]), + &[id_eq_x(0), id_eq_x(1), id_eq_x(3)], + HashSet::from([1, 3]), &input, ) .unwrap(); let filters = FilterDescription::new().with_child(child).parent_filters(); assert!(matches!(filters[0][0].discriminant, PushedDown::No)); assert!(matches!(filters[0][1].discriminant, PushedDown::Yes)); - assert_eq!(filters[0][1].predicate.to_string(), "id@1 = x"); + assert_eq!(filters[0][1].predicate.to_string(), "id@0 = x"); + assert!(matches!(filters[0][2].discriminant, PushedDown::Yes)); + assert_eq!(filters[0][2].predicate.to_string(), "id@0 = x"); +} + +/// An allowed position cannot make a column with an unknown name pushable. +#[test] +#[expect(deprecated)] +fn test_from_child_with_allowed_indices_rejects_unresolvable_name() { + use datafusion_physical_plan::filter_pushdown::{ + ChildFilterDescription, FilterDescription, PushedDown, + }; + use std::collections::HashSet; + + let input = TestScanBuilder::new(duplicate_id_schema()).build(); + let predicate: Arc = Arc::new(BinaryExpr::new( + Arc::new(Column::new("missing", 1)), + Operator::Eq, + Arc::new(Literal::new(ScalarValue::from("x"))), + )); + let child = ChildFilterDescription::from_child_with_allowed_indices( + &[Arc::clone(&predicate)], + HashSet::from([1]), + &input, + ) + .unwrap(); + let filters = FilterDescription::new().with_child(child).parent_filters(); + assert!(matches!(filters[0][0].discriminant, PushedDown::No)); + assert_eq!(filters[0][0].predicate.to_string(), predicate.to_string()); } /// A join's output projection must map to child positions even when a child diff --git a/datafusion/physical-plan/src/filter_pushdown.rs b/datafusion/physical-plan/src/filter_pushdown.rs index d457d9562e25b..7f54778400e07 100644 --- a/datafusion/physical-plan/src/filter_pushdown.rs +++ b/datafusion/physical-plan/src/filter_pushdown.rs @@ -43,6 +43,7 @@ use datafusion_common::{ tree_node::{Transformed, TreeNode}, }; use datafusion_physical_expr::expressions::Column; +use datafusion_physical_expr::utils::collect_columns; use datafusion_physical_expr_common::physical_expr::PhysicalExpr; #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -312,6 +313,18 @@ pub struct ChildFilterDescription { pub(crate) self_filters: Vec>, } +/// How a parent output position resolves to a child input position. +enum ColumnMapping { + /// Output position `i` reads child position `i`, and the child field at + /// that position must carry the same name. Used by schema-preserving + /// nodes such as sort, repartition and coalesce. + Identity, + /// Explicit output -> input positions supplied by a node that projects, + /// reorders or pairs columns (joins, aggregates, projected filters). + /// Names may differ, so the caller is trusted. + Explicit(HashMap), +} + /// Validates and remaps filter column references to a target schema in one step. /// /// When pushing filters from a parent to a child node, we need to: @@ -325,11 +338,7 @@ pub struct ChildFilterDescription { pub(crate) struct FilterRemapper { /// The target schema to remap column indices into. child_schema: SchemaRef, - /// Parent output index to child input index. `None` maps every parent - /// index to the same child index, for nodes that preserve their input - /// schema. Parent columns absent from an explicit mapping cannot be - /// pushed to this child. - column_mapping: Option>, + mapping: ColumnMapping, } impl FilterRemapper { @@ -339,7 +348,7 @@ impl FilterRemapper { pub(crate) fn new(child_schema: SchemaRef) -> Self { Self { child_schema, - column_mapping: None, + mapping: ColumnMapping::Identity, } } @@ -350,22 +359,20 @@ impl FilterRemapper { ) -> Self { Self { child_schema, - column_mapping: Some(column_mapping), + mapping: ColumnMapping::Explicit(column_mapping), } } /// Resolve a parent column to its position in the child schema. fn remap_column(&self, col: &Column) -> Option { - let index = match &self.column_mapping { - None => col.index(), - Some(mapping) => *mapping.get(&col.index())?, + let index = match &self.mapping { + ColumnMapping::Identity => { + let field = self.child_schema.fields().get(col.index())?; + (field.name() == col.name()).then_some(col.index())? + } + ColumnMapping::Explicit(mapping) => *mapping.get(&col.index())?, }; let field = self.child_schema.fields().get(index)?; - // With an identity mapping the parent and child names must agree; - // a mismatch means the caller does not actually preserve positions. - if self.column_mapping.is_none() && field.name() != col.name() { - return None; - } Some(Column::new(field.name(), index)) } @@ -415,23 +422,38 @@ impl ChildFilterDescription { Self::remap_filters(parent_filters, &remapper) } - /// Like [`Self::from_child`], but only forwards filters whose columns all - /// appear in `allowed_indices`. + /// Forwards filters whose columns all appear in `allowed_indices` and + /// resolve by name in the child schema. /// - /// Columns are resolved at the same position in the child schema. Earlier - /// versions resolved them by name, which is ambiguous when the child has - /// duplicate field names. Nodes whose output positions differ from the - /// child's must use [`Self::from_child_with_column_mapping`] instead. + /// Preserves the historical name-based resolution to the first matching + /// child field. This is ambiguous when the child has duplicate field names; + /// use [`Self::from_child_with_column_mapping`] to specify positions explicitly. #[deprecated( since = "56.0.0", - note = "columns now resolve by position; use `from_child` for matching schemas or `from_child_with_column_mapping` when positions differ" + note = "use `from_child` for matching schemas or `from_child_with_column_mapping` when positions differ" )] pub fn from_child_with_allowed_indices( parent_filters: &[Arc], allowed_indices: HashSet, child: &Arc, ) -> Result { - let column_mapping = allowed_indices.into_iter().map(|i| (i, i)).collect(); + if parent_filters.is_empty() { + return Ok(Self::empty()); + } + // Keep legacy name resolution local to this deprecated API. New callers + // must supply positions explicitly to avoid ambiguous column names. + let child_schema = child.schema(); + let column_mapping = parent_filters + .iter() + .flat_map(collect_columns) + .filter(|col| allowed_indices.contains(&col.index())) + .filter_map(|col| { + child_schema + .index_of(col.name()) + .ok() + .map(|child_index| (col.index(), child_index)) + }) + .collect(); Self::from_child_with_column_mapping(parent_filters, column_mapping, child) } diff --git a/docs/source/library-user-guide/upgrading/56.0.0.md b/docs/source/library-user-guide/upgrading/56.0.0.md index 3434e92214348..c7932e031de08 100644 --- a/docs/source/library-user-guide/upgrading/56.0.0.md +++ b/docs/source/library-user-guide/upgrading/56.0.0.md @@ -286,12 +286,10 @@ instead of looking them up by name. This prevents incorrect results when a child schema contains duplicate column names. `from_child` requires the child field at each referenced position to have the same name as the filter column. -`ChildFilterDescription::from_child_with_allowed_indices` is deprecated and -also resolves columns by position. Its allowed indices now map to the same -indices in the child schema; it no longer preserves the previous name-based -mapping. Existing callers with different parent and child column positions -must migrate, even if all column names are unique, or filters may reference -the wrong child column. +`ChildFilterDescription::from_child_with_allowed_indices` is deprecated but +preserves its previous name-based mapping to the first matching child field. +Migrate to `from_child_with_column_mapping` because name resolution is ambiguous +when the child schema contains duplicate field names. **Migration guide:** From 6d0e2b186b0b9bb0b65b2dc2c24abb8dbd3803dd Mon Sep 17 00:00:00 2001 From: Huaijin Date: Mon, 14 Sep 2026 22:57:34 +0800 Subject: [PATCH 7/8] add test case --- .../dynamic_filter_pushdown_config.slt | 119 ++++++++++++++++++ 1 file changed, 119 insertions(+) diff --git a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt index 66287f611a028..6a6bad99f0840 100644 --- a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt +++ b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt @@ -1057,3 +1057,122 @@ DROP TABLE issue_25244_b; statement ok RESET datafusion.optimizer.join_reordering; + +# Regression for https://github.com/apache/datafusion/issues/25296. +# Both joins expose columns named amount. The TopK filter on p.amount must +# not be remapped by name to o.amount in the lower join's output. Orders is +# larger than payments so it streams on the probe side. One row per batch +# and row group lets TopK update its filter before later orders are read. +# Name-based remapping prunes those orders and incorrectly returns (100, 30). +statement ok +SET datafusion.execution.target_partitions = 1; + +statement ok +SET datafusion.execution.batch_size = 1; + +statement ok +CREATE TABLE issue_25296_orders_src(id INT, amount INT) AS VALUES +(1, 100), (2, 200), (3, 300), (4, 400), (5, 500), (6, 600); + +statement ok +CREATE TABLE issue_25296_payments_src(id INT, amount INT) AS VALUES +(1, 30), (2, 20), (3, 10); + +statement ok +CREATE TABLE issue_25296_customers_src(id INT) AS VALUES (1), (2), (3); + +query I +COPY issue_25296_orders_src TO 'test_files/scratch/dynamic_filter_pushdown_config/issue_25296_orders.parquet' +STORED AS PARQUET OPTIONS ('format.max_row_group_size' '1'); +---- +6 + +query I +COPY issue_25296_payments_src TO 'test_files/scratch/dynamic_filter_pushdown_config/issue_25296_payments.parquet' +STORED AS PARQUET; +---- +3 + +query I +COPY issue_25296_customers_src TO 'test_files/scratch/dynamic_filter_pushdown_config/issue_25296_customers.parquet' +STORED AS PARQUET; +---- +3 + +statement ok +CREATE EXTERNAL TABLE issue_25296_orders(id INT, amount INT) STORED AS PARQUET +LOCATION 'test_files/scratch/dynamic_filter_pushdown_config/issue_25296_orders.parquet'; + +statement ok +CREATE EXTERNAL TABLE issue_25296_payments(id INT, amount INT) STORED AS PARQUET +LOCATION 'test_files/scratch/dynamic_filter_pushdown_config/issue_25296_payments.parquet'; + +statement ok +CREATE EXTERNAL TABLE issue_25296_customers(id INT) STORED AS PARQUET +LOCATION 'test_files/scratch/dynamic_filter_pushdown_config/issue_25296_customers.parquet'; + +# Establish the expected result without TopK dynamic filtering. +statement ok +SET datafusion.optimizer.enable_topk_dynamic_filter_pushdown = false; + +query II +SELECT o.amount, p.amount +FROM issue_25296_orders o +JOIN issue_25296_payments p ON o.id = p.id +JOIN issue_25296_customers c ON p.id = c.id +ORDER BY p.amount LIMIT 1; +---- +300 10 + +# Enabling TopK dynamic filtering must preserve the result. +statement ok +SET datafusion.optimizer.enable_topk_dynamic_filter_pushdown = true; + +query II +SELECT o.amount, p.amount +FROM issue_25296_orders o +JOIN issue_25296_payments p ON o.id = p.id +JOIN issue_25296_customers c ON p.id = c.id +ORDER BY p.amount LIMIT 1; +---- +300 10 + +# Also exercise Parquet row filtering in addition to row-group pruning. +statement ok +SET datafusion.execution.parquet.pushdown_filters = true; + +query II +SELECT o.amount, p.amount +FROM issue_25296_orders o +JOIN issue_25296_payments p ON o.id = p.id +JOIN issue_25296_customers c ON p.id = c.id +ORDER BY p.amount LIMIT 1; +---- +300 10 + +statement ok +DROP TABLE issue_25296_orders; + +statement ok +DROP TABLE issue_25296_payments; + +statement ok +DROP TABLE issue_25296_customers; + +statement ok +DROP TABLE issue_25296_orders_src; + +statement ok +DROP TABLE issue_25296_payments_src; + +statement ok +DROP TABLE issue_25296_customers_src; + +statement ok +RESET datafusion.execution.batch_size; + +statement ok +SET datafusion.execution.target_partitions = 4; + +statement ok +RESET datafusion.execution.parquet.pushdown_filters; From 85504605e8272c6ba21afcc6320d6695f8c59064 Mon Sep 17 00:00:00 2001 From: Huaijin Date: Mon, 14 Sep 2026 23:27:10 +0800 Subject: [PATCH 8/8] fix clippy --- datafusion/physical-plan/src/filter_pushdown.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/datafusion/physical-plan/src/filter_pushdown.rs b/datafusion/physical-plan/src/filter_pushdown.rs index 7f54778400e07..5eee4e2cdb8e5 100644 --- a/datafusion/physical-plan/src/filter_pushdown.rs +++ b/datafusion/physical-plan/src/filter_pushdown.rs @@ -446,7 +446,7 @@ impl ChildFilterDescription { let column_mapping = parent_filters .iter() .flat_map(collect_columns) - .filter(|col| allowed_indices.contains(&col.index())) + .filter(move |col| allowed_indices.contains(&col.index())) .filter_map(|col| { child_schema .index_of(col.name())