diff --git a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs index 26e6e0c74c49d..f559d33a5e4b0 100644 --- a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs +++ b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs @@ -1601,6 +1601,403 @@ 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}"); + } +} + +/// 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 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 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_preserves_name_resolution() { + 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), 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@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 +/// 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/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..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}; @@ -57,7 +57,8 @@ use datafusion_common::config::ConfigOptions; use datafusion_common::stats::Precision; 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; @@ -66,7 +67,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 +317,29 @@ 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 { + if parent_filters.is_empty() { + return Ok(ChildFilterDescription::empty()); + } + 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 @@ -697,21 +721,18 @@ impl ExecutionPlan for FilterExec { parent_filters: Vec>, _config: &ConfigOptions, ) -> Result { - if phase != FilterPushdownPhase::Pre { - let child = - ChildFilterDescription::from_child(&parent_filters, self.input())?; - 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 = ChildFilterDescription::from_child(&parent_filters, self.input())? - .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( @@ -735,12 +756,25 @@ 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() { - let input_schema = self.input().schema(); + // 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() + && !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| reassign_expr_columns(expr, &input_schema)) + .map(|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:?}" + ) + }) + }) .collect::>>()?; } diff --git a/datafusion/physical-plan/src/filter_pushdown.rs b/datafusion/physical-plan/src/filter_pushdown.rs index dfaec62d062ec..5eee4e2cdb8e5 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; @@ -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,55 +313,71 @@ 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: -/// 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 /// -/// `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. +/// 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, - /// 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, + mapping: ColumnMapping, } 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. + /// 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 new(child_schema: SchemaRef) -> Self { - let allowed_indices = (0..child_schema.fields().len()).collect(); Self { child_schema, - allowed_indices, + mapping: ColumnMapping::Identity, } } - /// 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. - fn with_allowed_indices( + /// Create a remapper with an explicit parent-output to child-input mapping. + pub(crate) fn with_column_mapping( child_schema: SchemaRef, - allowed_indices: HashSet, + column_mapping: HashMap, ) -> Self { Self { child_schema, - allowed_indices, + 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.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)?; + Some(Column::new(field.name(), index)) + } + /// 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 +386,8 @@ 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, - )))) + if let Some(remapped) = self.remap_column(col) { + Ok(Transformed::yes(Arc::new(remapped))) } else { all_valid = false; Ok(Transformed::complete(expr)) @@ -390,21 +402,19 @@ 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()); } @@ -412,29 +422,59 @@ impl ChildFilterDescription { Self::remap_filters(parent_filters, &remapper) } - /// Like [`Self::from_child`], but restricts which parent-level columns are - /// considered reachable through this child. + /// Forwards filters whose columns all appear in `allowed_indices` and + /// resolve by name in the child schema. /// - /// `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`. - /// - /// 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. + /// 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 = "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 { - // See [`Self::from_child`]: nothing to remap, nothing to index. + 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(move |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) + } + + /// Remap parent filters using an explicit parent-output to child-input + /// column mapping. Columns absent from the mapping cannot be pushed down. + /// + /// 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, + child: &Arc, + ) -> Result { if parent_filters.is_empty() { return Ok(Self::empty()); } let remapper = - FilterRemapper::with_allowed_indices(child.schema(), allowed_indices); + FilterRemapper::with_column_mapping(child.schema(), column_mapping); Self::remap_filters(parent_filters, &remapper) } @@ -459,7 +499,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/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index b72e180543f9a..088376bc349c1 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}; @@ -1819,13 +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); - // Build the set of allowed column indices for each side + // Map each output position to its input position, accounting for the + // join's projection. let column_indices: Vec = match self.projection.as_ref() { Some(projection) => projection .iter() @@ -1834,59 +1835,56 @@ 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 (output_side, other_mapping) = match self.join_type { + JoinType::LeftSemi => (JoinSide::Left, &mut right_mapping), + _ => (JoinSide::Right, &mut left_mapping), + }; + 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); } } - _ => {} } 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 +1892,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/projection.rs b/datafusion/physical-plan/src/projection.rs index 7851e5705934e..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, @@ -49,7 +48,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; @@ -304,29 +303,6 @@ impl ProjectionExec { fn overrides_metadata(&self) -> bool { 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( - &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) - } } impl DisplayAs for ProjectionExec { @@ -545,17 +521,16 @@ 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_schema = self.schema(); let remapper = FilterRemapper::new(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 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/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt index 514ae81095a49..6a6bad99f0840 100644 --- a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt +++ b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt @@ -904,3 +904,275 @@ 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. +# 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; + +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'; + +# 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 +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 + +# 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, 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 ( + 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 + +# The same queries must return the same rows with join dynamic filtering +# enabled. +statement ok +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 +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 + +# 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, 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 ( + 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 + +statement ok +DROP TABLE issue_25244_a; + +statement ok +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; 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..c7932e031de08 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,46 @@ 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 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:** + +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, +)?; +```