From 5bfdf5e608edca13f127870a50c8590b4f3eeddb Mon Sep 17 00:00:00 2001 From: Nimalan Date: Sun, 6 Sep 2026 20:04:20 +0530 Subject: [PATCH 1/2] feat: support filter pushdown through unnest operator --- .../physical_optimizer/filter_pushdown.rs | 280 ++++++++++++++++++ datafusion/physical-plan/src/unnest.rs | 56 ++++ 2 files changed, 336 insertions(+) diff --git a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs index 26e6e0c74c49d..69909e42410e8 100644 --- a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs +++ b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs @@ -810,6 +810,217 @@ fn test_pushdown_through_aggregates_preserves_parent_filter_order() { ); } +/// Schema for the unnest pushdown tests: two passthrough columns and a list +/// column that gets unnested. +fn unnest_input_schema() -> SchemaRef { + Arc::new(Schema::new(vec![ + Field::new("a", DataType::Utf8, false), + Field::new("b", DataType::Utf8, false), + Field::new( + "l", + DataType::List(Arc::new(Field::new("item", DataType::Utf8, true))), + true, + ), + ])) +} + +/// Output schema of unnesting `l` in [`unnest_input_schema`]: the list column +/// is replaced in place by its element type. +fn unnest_output_schema() -> SchemaRef { + Arc::new(Schema::new(vec![ + Field::new("a", DataType::Utf8, false), + Field::new("b", DataType::Utf8, false), + Field::new("l", DataType::Utf8, true), + ])) +} + +fn unnest_over(input: Arc) -> Arc { + use datafusion_common::UnnestOptions; + use datafusion_physical_plan::unnest::{ListUnnest, UnnestExec}; + Arc::new( + UnnestExec::new( + input, + vec![ListUnnest { + index_in_input_schema: 2, + depth: 1, + }], + vec![], + unnest_output_schema(), + UnnestOptions::default(), + ) + .unwrap(), + ) +} + +#[test] +fn test_pushdown_through_unnest_on_passthrough_columns() { + // A filter conjunct on a passthrough (non-unnested) column commutes with + // the unnest and gets absorbed into the scan; the conjunct on the unnested + // column must stay above the UnnestExec. + let scan = TestScanBuilder::new(unnest_input_schema()) + .with_support(true) + .build(); + let unnest = unnest_over(scan); + + let predicate = Arc::new(BinaryExpr::new( + col_lit_predicate("a", "foo", &unnest_output_schema()), + Operator::And, + col_lit_predicate("l", "bar", &unnest_output_schema()), + )); + let plan = Arc::new(FilterExec::try_new(predicate, unnest).unwrap()); + + insta::assert_snapshot!( + OptimizationTest::new(plan, FilterPushdown::new(), true), + @" + OptimizationTest: + input: + - FilterExec: a@0 = foo AND l@2 = bar + - UnnestExec + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, l], file_type=test, pushdown_supported=true + output: + Ok: + - FilterExec: l@2 = bar + - UnnestExec + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, l], file_type=test, pushdown_supported=true, predicate=a@0 = foo + " + ); +} + +/// Probe-side batch with a list column, so the unnest has real rows to expand. +fn unnest_probe_batch() -> RecordBatch { + use arrow::array::{ListBuilder, StringArray, StringBuilder}; + let mut list = ListBuilder::new(StringBuilder::new()); + for items in [["x", "y"].as_slice(), &["z"], &["w"], &["v"]] { + for item in items { + list.values().append_value(item); + } + list.append(true); + } + RecordBatch::try_new( + unnest_input_schema(), + vec![ + Arc::new(StringArray::from(vec!["aa", "ab", "ac", "ad"])), + Arc::new(StringArray::from(vec!["ba", "bb", "bc", "bd"])), + Arc::new(list.finish()), + ], + ) + .unwrap() +} + +#[tokio::test] +async fn test_hashjoin_dynamic_filter_pushdown_through_unnest() { + // A CollectLeft hash join's dynamic filter on a passthrough column must + // reach the scan below an UnnestExec on the probe side, and must actually + // prune there: the bounds come from the build side, and the probe scan + // applies them *before* the unnest multiplies its rows. + let build_side_schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Utf8, false), + Field::new("b", DataType::Utf8, false), + Field::new("c", DataType::Float64, false), + ])); + let build_scan = TestScanBuilder::new(Arc::clone(&build_side_schema)) + .with_support(true) + .with_batches(vec![ + record_batch!( + ("a", Utf8, ["aa", "ab"]), + ("b", Utf8, ["ba", "bb"]), + ("c", Float64, [1.0, 2.0]) + ) + .unwrap(), + ]) + .build(); + + let probe_scan = TestScanBuilder::new(unnest_input_schema()) + .with_support(true) + .with_batches(vec![unnest_probe_batch()]) + .build(); + let probe_unnest = unnest_over(Arc::clone(&probe_scan)); + + let on = vec![( + col("a", &build_side_schema).unwrap(), + col("a", &unnest_output_schema()).unwrap(), + )]; + let hash_join = Arc::new( + HashJoinExec::try_new( + build_scan, + probe_unnest, + on, + None, + &JoinType::Inner, + None, + PartitionMode::CollectLeft, + datafusion_common::NullEquality::NullEqualsNothing, + false, + ) + .unwrap(), + ); + + // Sort on the unnested column for deterministic output. + let plan = Arc::new(SortExec::new( + LexOrdering::new(vec![PhysicalSortExpr::new( + col("l", &hash_join.schema()).unwrap(), + SortOptions::new(false, false), + )]) + .unwrap(), + Arc::clone(&hash_join) as Arc, + )) as Arc; + + insta::assert_snapshot!( + OptimizationTest::new(Arc::clone(&plan), FilterPushdown::new_post_optimization(), true), + @" + OptimizationTest: + input: + - SortExec: expr=[l@5 ASC NULLS LAST], preserve_partitioning=[false] + - HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, a@0)] + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true + - UnnestExec + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, l], file_type=test, pushdown_supported=true + output: + Ok: + - SortExec: expr=[l@5 ASC NULLS LAST], preserve_partitioning=[false] + - HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, a@0)] + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true + - UnnestExec + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, l], file_type=test, pushdown_supported=true, predicate=DynamicFilter [ empty ] + " + ); + + let mut config = ConfigOptions::default(); + config.execution.parquet.pushdown_filters = true; + config.optimizer.enable_dynamic_filter_pushdown = true; + let (plan, batches) = optimize_and_collect_pushdown_plan(plan, config).await; + + // The filter that reached the scan below the unnest carries the build + // side's real bounds on the passthrough column, not an empty placeholder. + insta::assert_snapshot!( + format!("{}", format_plan_for_test(&plan)), + @" + - SortExec: expr=[l@5 ASC NULLS LAST], preserve_partitioning=[false] + - HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, a@0)] + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true + - UnnestExec + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, l], file_type=test, pushdown_supported=true, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND a@0 IN (SET) ([aa, ab]) ] + " + ); + + // Only the two rows whose `a` matches the build side survive, and they are + // dropped before the unnest expands them. + assert_eq!(probe_scan.metrics().unwrap().output_rows().unwrap(), 2); + + insta::assert_snapshot!( + format!("{}", pretty_format_batches(&batches).unwrap()), + @r" + +----+----+-----+----+----+---+ + | a | b | c | a | b | l | + +----+----+-----+----+----+---+ + | aa | ba | 1.0 | aa | ba | x | + | aa | ba | 1.0 | aa | ba | y | + | ab | bb | 2.0 | ab | bb | z | + +----+----+-----+----+----+---+ + " + ); +} + /// Test various combinations of handling of child pushdown results /// in an ExecutionPlan in combination with support/not support in a DataSource. #[test] @@ -3853,3 +4064,72 @@ fn post_phase_is_idempotent_on_hash_join() { "second invocation of FilterPushdown::new_post_optimization mutated the plan", ); } + +#[test] +fn test_pushdown_through_struct_unnest_shifts_column_indices() { + // A struct expands into *multiple* output columns in place, so passthrough + // columns after it sit at a different index above the unnest than below it + // (`b` is 3 in the output, 2 in the input). The pushed filter must be + // rewritten to the input's indices. + use datafusion_common::UnnestOptions; + use datafusion_physical_plan::unnest::UnnestExec; + + let input_schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Utf8, false), + Field::new( + "s", + DataType::Struct( + vec![ + Field::new("f1", DataType::Utf8, true), + Field::new("f2", DataType::Utf8, true), + ] + .into(), + ), + true, + ), + Field::new("b", DataType::Utf8, false), + ])); + let output_schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Utf8, false), + Field::new("s.f1", DataType::Utf8, true), + Field::new("s.f2", DataType::Utf8, true), + Field::new("b", DataType::Utf8, false), + ])); + + let scan = TestScanBuilder::new(Arc::clone(&input_schema)) + .with_support(true) + .build(); + let unnest = Arc::new( + UnnestExec::new( + scan, + vec![], + vec![1], + Arc::clone(&output_schema), + UnnestOptions::default(), + ) + .unwrap(), + ); + + let predicate = Arc::new(BinaryExpr::new( + col_lit_predicate("b", "x", &output_schema), + Operator::And, + col_lit_predicate("s.f1", "y", &output_schema), + )); + let plan = Arc::new(FilterExec::try_new(predicate, unnest).unwrap()); + + insta::assert_snapshot!( + OptimizationTest::new(plan, FilterPushdown::new(), true), + @" + OptimizationTest: + input: + - FilterExec: b@3 = x AND s.f1@1 = y + - UnnestExec + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, s, b], file_type=test, pushdown_supported=true + output: + Ok: + - FilterExec: s.f1@1 = y + - UnnestExec + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, s, b], file_type=test, pushdown_supported=true, predicate=b@2 = x + " + ); +} diff --git a/datafusion/physical-plan/src/unnest.rs b/datafusion/physical-plan/src/unnest.rs index d93e0280515c6..d59bc27ac2d2c 100644 --- a/datafusion/physical-plan/src/unnest.rs +++ b/datafusion/physical-plan/src/unnest.rs @@ -26,6 +26,10 @@ use super::metrics::{ MetricsSet, SplitMetrics, }; use super::{DisplayAs, ExecutionPlanProperties, PlanProperties}; +use crate::filter_pushdown::{ + ChildFilterDescription, ChildPushdownResult, FilterDescription, FilterPushdownPhase, + FilterPushdownPropagation, +}; use crate::stream::{BatchSplitStream, EmptyRecordBatchStream, ObservedStream}; use crate::{ ChildrenPropertiesMode, DisplayFormatType, Distribution, ExecutionPlan, @@ -45,6 +49,7 @@ use arrow::datatypes::{DataType, Int64Type, Schema, SchemaRef}; use arrow::record_batch::RecordBatch; use arrow_ord::cmp::lt; use async_trait::async_trait; +use datafusion_common::config::ConfigOptions; use datafusion_common::tree_node::TreeNodeRecursion; use datafusion_common::{ Constraints, HashMap, HashSet, Result, UnnestOptions, exec_datafusion_err, exec_err, @@ -328,6 +333,57 @@ impl ExecutionPlan for UnnestExec { Some(self.metrics.clone_inner()) } + fn gather_filters_for_pushdown( + &self, + _phase: FilterPushdownPhase, + parent_filters: Vec>, + _config: &ConfigOptions, + ) -> Result { + // Filters that reference only non-unnested (passthrough) columns commute + // with the unnest. Non-unnest values values are replicated onto every row + // produced from an input row, so dropping an input row before the unnest + // removes exactly the output rows the filter would have dropped after it. + // Filters referencing unnested list/struct columns must stay above this + // node. + let input_schema = self.input.schema(); + let unnested_input_indices: HashSet = self + .list_column_indices + .iter() + .map(|list_unnest| list_unnest.index_in_input_schema) + .chain(self.struct_column_indices.iter().copied()) + .collect(); + + // Output indices of passthrough columns, resolved by name the same way + // `compute_properties` builds its projection mapping. + let allowed_output_indices: std::collections::HashSet = + (0..input_schema.fields().len()) + .filter(|input_idx| !unnested_input_indices.contains(input_idx)) + .filter_map(|input_idx| { + let name = input_schema.field(input_idx).name(); + self.schema + .fields() + .iter() + .position(|output_field| output_field.name() == name) + }) + .collect(); + + let child = ChildFilterDescription::from_child_with_allowed_indices( + &parent_filters, + allowed_output_indices, + &self.input, + )?; + Ok(FilterDescription::new().with_child(child)) + } + + fn handle_child_pushdown_result( + &self, + _phase: FilterPushdownPhase, + child_pushdown_result: ChildPushdownResult, + _config: &ConfigOptions, + ) -> Result>> { + Ok(FilterPushdownPropagation::if_all(child_pushdown_result)) + } + #[cfg(feature = "proto")] fn try_to_proto( &self, From 8080395cc32ba219bc9b8e5c977afcfafa4ca6b5 Mon Sep 17 00:00:00 2001 From: Nimalan Date: Mon, 7 Sep 2026 10:44:18 +0530 Subject: [PATCH 2/2] fix: Resolve pushdown by using indices rather than names to prevent name collision --- .../physical_optimizer/filter_pushdown.rs | 92 +++++++ datafusion/physical-plan/src/unnest.rs | 230 ++++++++++++------ 2 files changed, 254 insertions(+), 68 deletions(-) diff --git a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs index 69909e42410e8..09fa5564e31aa 100644 --- a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs +++ b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs @@ -4133,3 +4133,95 @@ fn test_pushdown_through_struct_unnest_shifts_column_indices() { " ); } + +#[test] +fn test_no_pushdown_when_unnested_struct_field_name_collides() { + // Unnesting struct `s` synthesizes an output column named `s.f1`, which + // collides with the genuine passthrough column physically named `s.f1`. + // The output then holds two `s.f1` fields, distinguished only by index. + // A filter on the *generated* one (`s.f1@0`) must not be pushed: resolving + // by name would hand it the unrelated passthrough input column instead. + use datafusion_common::UnnestOptions; + use datafusion_physical_plan::unnest::UnnestExec; + + let input_schema = Arc::new(Schema::new(vec![ + Field::new( + "s", + DataType::Struct(vec![Field::new("f1", DataType::Utf8, true)].into()), + true, + ), + Field::new("s.f1", DataType::Utf8, false), + ])); + // `s` expands in place, so the generated `s.f1` lands at index 0 and the + // passthrough column of the same name is pushed to index 1. + let output_schema = Arc::new(Schema::new(vec![ + Field::new("s.f1", DataType::Utf8, true), + Field::new("s.f1", DataType::Utf8, false), + ])); + + let scan = TestScanBuilder::new(Arc::clone(&input_schema)) + .with_support(true) + .build(); + let unnest = Arc::new( + UnnestExec::new( + scan, + vec![], + vec![0], + Arc::clone(&output_schema), + UnnestOptions::default(), + ) + .unwrap(), + ); + + // Filter on the generated column: must stay above the unnest. + let generated = Arc::new( + FilterExec::try_new( + col_lit_predicate("s.f1", "x", &output_schema), + Arc::clone(&unnest) as Arc, + ) + .unwrap(), + ); + insta::assert_snapshot!( + OptimizationTest::new(generated, FilterPushdown::new(), true), + @" + OptimizationTest: + input: + - FilterExec: s.f1@0 = x + - UnnestExec + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[s, s.f1], file_type=test, pushdown_supported=true + output: + Ok: + - FilterExec: s.f1@0 = x + - UnnestExec + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[s, s.f1], file_type=test, pushdown_supported=true + " + ); + + // Filter on the real passthrough column at index 1: safe to push, and must + // be rewritten to that column's index in the input schema. + let passthrough = Arc::new( + FilterExec::try_new( + Arc::new(BinaryExpr::new( + Arc::new(Column::new("s.f1", 1)), + Operator::Eq, + Arc::new(Literal::new(ScalarValue::from("x"))), + )), + unnest, + ) + .unwrap(), + ); + insta::assert_snapshot!( + OptimizationTest::new(passthrough, FilterPushdown::new(), true), + @" + OptimizationTest: + input: + - FilterExec: s.f1@1 = x + - UnnestExec + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[s, s.f1], file_type=test, pushdown_supported=true + output: + Ok: + - UnnestExec + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[s, s.f1], file_type=test, pushdown_supported=true, predicate=s.f1@1 = x + " + ); +} diff --git a/datafusion/physical-plan/src/unnest.rs b/datafusion/physical-plan/src/unnest.rs index d59bc27ac2d2c..5f9d0808824cf 100644 --- a/datafusion/physical-plan/src/unnest.rs +++ b/datafusion/physical-plan/src/unnest.rs @@ -38,9 +38,9 @@ use crate::{ }; use arrow::array::{ - Array, ArrayRef, AsArray, BooleanBufferBuilder, FixedSizeListArray, Int64Array, - LargeListArray, LargeListViewArray, ListArray, ListViewArray, PrimitiveArray, Scalar, - StructArray, new_null_array, + Array, ArrayRef, AsArray, FixedSizeListArray, Int64Array, LargeListArray, + LargeListViewArray, ListArray, ListViewArray, PrimitiveArray, Scalar, StructArray, + new_null_array, }; use arrow::compute::kernels::length::length; use arrow::compute::kernels::zip::zip; @@ -120,48 +120,32 @@ impl UnnestExec { struct_column_indices: &[usize], schema: &SchemaRef, ) -> Result { - // Find out which indices are not unnested, such that they can be copied over from the input plan let input_schema = input.schema(); - let mut unnested_indices = BooleanBufferBuilder::new(input_schema.fields().len()); - unnested_indices.append_n(input_schema.fields().len(), false); - for list_unnest in list_column_indices { - unnested_indices.set_bit(list_unnest.index_in_input_schema, true); - } - for struct_unnest in struct_column_indices { - unnested_indices.set_bit(*struct_unnest, true) - } - let unnested_indices = unnested_indices.finish(); - let non_unnested_indices: Vec = (0..input_schema.fields().len()) - .filter(|idx| !unnested_indices.value(*idx)) - .collect(); - // Manually build projection mapping from non-unnested input columns to their positions in the output - let input_schema = input.schema(); - let projection_mapping: ProjectionMapping = non_unnested_indices - .iter() - .map(|&input_idx| { - // Find what index the input column has in the output schema - let input_field = input_schema.field(input_idx); - let output_idx = schema - .fields() - .iter() - .position(|output_field| output_field.name() == input_field.name()) - .ok_or_else(|| { - exec_datafusion_err!( - "Non-unnested column '{}' must exist in output schema", - input_field.name() - ) - })?; - - let input_col = Arc::new(Column::new(input_field.name(), input_idx)) - as Arc; - let target_col = Arc::new(Column::new(input_field.name(), output_idx)) - as Arc; - // Use From, usize)>> for ProjectionTargets - let targets = vec![(target_col, output_idx)].into(); - Ok((input_col, targets)) - }) - .collect::>()?; + // Passthrough columns keep their input equivalences, at whatever index + // the unnest's in-place expansion leaves them. Derived positionally a + // struct expansion synthesizes names like `s.f1` that can collide with a + // real column of that name, and resolving by name would then map an + // input column onto an unrelated output one. + let projection_mapping: ProjectionMapping = Self::passthrough_columns( + &input_schema, + schema, + list_column_indices, + struct_column_indices, + ) + .unwrap_or_default() + .into_iter() + .map(|(input_idx, output_idx)| { + let name = input_schema.field(input_idx).name(); + let input_col = + Arc::new(Column::new(name, input_idx)) as Arc; + let target_col = + Arc::new(Column::new(name, output_idx)) as Arc; + // Use From, usize)>> for ProjectionTargets + let targets = vec![(target_col, output_idx)].into(); + (input_col, targets) + }) + .collect(); // Create the unnest's equivalence properties by copying the input plan's equivalence properties // for the unaffected columns. Except for the constraints, which are removed entirely because @@ -202,6 +186,56 @@ impl UnnestExec { pub fn options(&self) -> &UnnestOptions { &self.options } + + /// The columns that pass through this unnest untouched with + /// input_schema_index = output_schema_index. This is needed to + /// find out columns which potentially can be pushdown and to also make + /// sure these pushdown columns do not conflict with the new columns + /// expanded through unnest. + /// + /// Each input column is replaced in place by however many columns it + /// expands into, so the output index of a passthrough column is the running + /// total of the widths before it. Physical expressions are index-based + /// precisely so such duplicates stay distinguishable. + /// + /// Returns `None` when the reconstruction disagrees with the real output + /// schema. We then then fall back to pushing down nothing rather than acting on a guess. + fn passthrough_columns( + input_schema: &SchemaRef, + output_schema: &SchemaRef, + list_column_indices: &[ListUnnest], + struct_column_indices: &[usize], + ) -> Option> { + let input_len = input_schema.fields().len(); + + // A list column is replaced by one column per `ListUnnest` entry naming + // it, several when it is unnested at multiple depths. + let mut list_counts = vec![0usize; input_len]; + for list_unnest in list_column_indices { + *list_counts.get_mut(list_unnest.index_in_input_schema)? += 1; + } + + let mut passthrough = Vec::with_capacity(input_len); + let mut output_idx = 0; + for (input_idx, &list_count) in list_counts.iter().enumerate() { + if list_count > 0 { + // Unnested into `list_count` generated columns. + output_idx += list_count; + } else if struct_column_indices.contains(&input_idx) { + // Flattened into one column per struct field. + let DataType::Struct(fields) = input_schema.field(input_idx).data_type() + else { + return None; + }; + output_idx += fields.len(); + } else { + passthrough.push((input_idx, output_idx)); + output_idx += 1; + } + } + + (output_idx == output_schema.fields().len()).then_some(passthrough) + } } impl DisplayAs for UnnestExec { @@ -339,34 +373,38 @@ impl ExecutionPlan for UnnestExec { parent_filters: Vec>, _config: &ConfigOptions, ) -> Result { - // Filters that reference only non-unnested (passthrough) columns commute - // with the unnest. Non-unnest values values are replicated onto every row - // produced from an input row, so dropping an input row before the unnest - // removes exactly the output rows the filter would have dropped after it. - // Filters referencing unnested list/struct columns must stay above this - // node. + // Filters that reference only passthrough columns commute with unnest. + // So dropping an input row before the unnest removes exactly the output + // rows the filter would have dropped after it. Filters referencing unnested + // list/struct columns must stay above this node. + let Some(passthrough) = Self::passthrough_columns( + &self.input.schema(), + &self.schema, + &self.list_column_indices, + &self.struct_column_indices, + ) else { + return Ok(FilterDescription::all_unsupported( + &parent_filters, + &[&self.input], + )); + }; + let input_schema = self.input.schema(); - let unnested_input_indices: HashSet = self - .list_column_indices - .iter() - .map(|list_unnest| list_unnest.index_in_input_schema) - .chain(self.struct_column_indices.iter().copied()) + let allowed_output_indices: std::collections::HashSet = passthrough + .into_iter() + // `FilterRemapper` rewrites a pushed column by looking its *name* up + // in the input schema, so a column is only safe to admit when that + // lookup lands back on the very column it came from. Duplicate input + // names would otherwise redirect the filter to an unrelated column. + .filter(|&(input_idx, _)| { + matches!( + input_schema.index_of(input_schema.field(input_idx).name()), + Ok(found) if found == input_idx + ) + }) + .map(|(_, output_idx)| output_idx) .collect(); - // Output indices of passthrough columns, resolved by name the same way - // `compute_properties` builds its projection mapping. - let allowed_output_indices: std::collections::HashSet = - (0..input_schema.fields().len()) - .filter(|input_idx| !unnested_input_indices.contains(input_idx)) - .filter_map(|input_idx| { - let name = input_schema.field(input_idx).name(); - self.schema - .fields() - .iter() - .position(|output_field| output_field.name() == name) - }) - .collect(); - let child = ChildFilterDescription::from_child_with_allowed_indices( &parent_filters, allowed_output_indices, @@ -2329,6 +2367,62 @@ mod tests { .collect() } + /// Unnesting struct `s` synthesizes an output column named `s.f1`, colliding + /// with the genuine passthrough column of that name. The equivalence + /// properties must follow the *positional* expansion. The passthrough column + /// lands at output index 1, so an input ordering on it must be reported on + /// `s.f1@1`. Matching by name would attach it to the generated `s.f1@0` and + /// could wrongly elide a sort on an unrelated column. + #[test] + fn test_compute_properties_with_colliding_struct_field_name() -> Result<()> { + let input_schema = Arc::new(Schema::new(vec![ + Field::new( + "s", + DataType::Struct(vec![Field::new("f1", DataType::Utf8, true)].into()), + true, + ), + Field::new("s.f1", DataType::Utf8, false), + ])); + // `s` expands in place, so the generated `s.f1` takes index 0 and the + // passthrough column of the same name is pushed to index 1. + let output_schema = Arc::new(Schema::new(vec![ + Field::new("s.f1", DataType::Utf8, true), + Field::new("s.f1", DataType::Utf8, false), + ])); + + assert_eq!( + UnnestExec::passthrough_columns(&input_schema, &output_schema, &[], &[0]), + Some(vec![(1, 1)]), + ); + + use datafusion_physical_expr::{LexOrdering, PhysicalSortExpr}; + let ordering = LexOrdering::new(vec![PhysicalSortExpr::new_default(Arc::new( + Column::new("s.f1", 1), + ) + as Arc)]) + .expect("single-element ordering"); + let source = crate::test::TestMemoryExec::try_new( + &[vec![]], + Arc::clone(&input_schema), + None, + )? + .try_with_sort_information(vec![ordering])?; + let input = + Arc::new(crate::test::TestMemoryExec::update_cache(&Arc::new(source))); + + let unnest = UnnestExec::new( + input, + vec![], + vec![0], + output_schema, + UnnestOptions::default(), + )?; + + let orderings = unnest.properties().equivalence_properties().oeq_class(); + assert_snapshot!(orderings.to_string(), @"[[s.f1@1 ASC]]"); + Ok(()) + } + /// Output batch sizes are fully determined by the input lengths and `batch_size`, so /// assert the exact shapes rather than just the `<= batch_size` bound. Each case pins a /// distinct path through `next_chunk_rows`.