FilterPushdown physical optimizer incorrectly remaps filters through ProjectionExec with duplicate column names
- Dominant language
- Rust
- Stars
- 9.3k
- Forks
- 2.4k
- Avg merge
- 3d 7h
- Merged PRs (30d)
- 344
Description
### Describe the bug
`ProjectionExec::gather_filters_for_pushdown` silently rewrites filter predicates to the wrong source column when the projection's output schema contains duplicate column names. This produces incorrect query results/missing rows with no error.
The root cause is two functions that use name-only schema lookups, which always return the first match:
**Bug 1** — `collect_reverse_alias` (`projection.rs`): Builds the reverse alias map using `column_with_name(&alias)`, which always returns the first column with that name. When two projection outputs share a name (e.g., `id` at positions 0 and 2), both iterations produce the same `HashMap` key `Column("id", 0)`, and the second silently overwrites the first.
```rust
// column_with_name always returns the FIRST match
let (aliased_index, _) = self.projector.output_schema().column_with_name(&projection.alias)?; // ← always returns 0 for duplicate names
let aliased_col = Column::new(&projection.alias, aliased_index);
alias_map.insert(aliased_col, Arc::clone( & projection.expr)); // ← overwrites on collision
```
**Bug 2** — `FilterRemapper::try_remap` (`filter_pushdown.rs`): Uses `schema.index_of(col.name())` which also returns the first match. A filter on `id@2` (the second occurrence) gets silently rewritten to `id@0`.
```rust
// index_of always returns the FIRST match
let Ok(new_index) = self.child_schema.index_of(col.name())
// col was id@2 → new_index is 0 → rewrites to id@0
```
### To Reproduce
Construct a physical plan where a `ProjectionExec` produces duplicate column names from a join, with a `FilterExec` above it:
```
FilterExec: id@0 = 3 ← checks output col 0 (right side's id)
ProjectionExec: [right_id@2 as id, ← output col 0 (from RIGHT)
name@1,
left_id@0 as id] ← output col 2 (from LEFT)
HashJoinExec: Left
left: [left_id, name] (columns 0-1)
right: [right_id] (column 2)
```
The filter `id@0 = 3` should check the right side's id. After `FilterPushdown`:
- Bug 1 overwrites `Column("id", 0) → right_id@2` with `Column("id", 0) → left_id@0`
- The filter gets remapped to `left_id@0 = 3` — filtering the wrong column
- Query returns incorrect results
Here's the test to reproduce this issue, for convenience add it to `datafusion/core/tests/physical_optimizer/filter_pushdown.rs`. It constructs the physical plan directly and runs it before and after FilterPushdown. The original plan returns 3 correct rows; the optimized plan returns 0. You can run this test with:
```bash
cargo test -p datafusion --test core_integration -- test_filter_pushdown_projection_duplicate_column_names --no-capture
```
```rust
/// Reproduces a bug where FilterPushdown through ProjectionExec with duplicate
/// column names remaps filter predicates to the wrong source column.
///
/// Plan structure:
///
/// FilterExec: id@0 IS NULL ← checks output col 0 (right side's id)
/// ProjectionExec: [right_id@2 as id, ← output col 0 (from RIGHT)
/// name@1,
/// left_id@0 as id] ← output col 2 (from LEFT)
/// HashJoinExec: Left
/// left: [left_id, name] (columns 0-1)
/// right: [right_id] (column 2)
///
/// Bug 1 overwrites Column("id", 0) → left_id@0 instead of right_id@2.
/// The filter `id@0 IS NULL` gets remapped to `left_id@0 IS NULL` (wrong side).
#[tokio::test]
async fn test_filter_pushdown_projection_duplicate_column_names() {
use datafusion_common::JoinType;
use datafusion_physical_expr::expressions::is_null;
use datafusion_physical_plan::joins::{HashJoinExec, PartitionMode};
// Left table: orders (5 rows, all non-NULL left_id)
let left_batches = vec![record_batch!(
("left_id", Int32, [1, 2, 3, 4, 5]),
("name", Utf8, ["Alice", "Bob", "Charlie", "Diana", "Eve"])
)
.unwrap()];
let left_schema = Arc::new(Schema::new(vec![
Field::new("left_id", DataType::Int32, false),
Field::new("name", DataType::Utf8, false),
]));
let left_scan = TestScanBuilder::new(Arc::clone(&left_schema))
.with_batches(left_batches)
.build();
// Right table: returns (2 rows matching orders 1 and 3)
let right_batches = vec![
record_batch!(("right_id", Int32, [1, 3])).unwrap(),
];
let right_schema = Arc::new(Schema::new(vec![Field::new(
"right_id",
DataType::Int32,
false,
)]));
let right_scan = TestScanBuilder::new(Arc::clone(&right_schema))
.with_batches(right_batches)
.build();
// HashJoinExec: LEFT JOIN on left_id = right_id
// Join output schema: [left_id(0), name(1), right_id(2)]
let join = Arc::new(
HashJoinExec::try_new(
left_scan,
right_scan,
vec![(
col("left_id", &left_schema).unwrap(),
col("right_id", &right_schema).unwrap(),
)],
None,
&JoinType::Left,
None,
PartitionMode::CollectLeft,
datafusion_common::NullEquality::NullEqualsNothing,
false,
)
.unwrap(),
);
let join_schema = join.schema();
// ProjectionExec: creates duplicate "id" columns
// output col 0: right_id@2 AS id (from RIGHT side, NULL for unmatched)
// output col 1: name@1
// output col 2: left_id@0 AS id (from LEFT side, never NULL)
let projection = Arc::new(
ProjectionExec::try_new(
vec![
(col("right_id", &join_schema).unwrap(), "id".to_string()),
(col("name", &join_schema).unwrap(), "name".to_string()),
(col("left_id", &join_schema).unwrap(), "id".to_string()),
],
join,
)
.unwrap(),
);
// FilterExec: id@0 IS NULL
// This should check the RIGHT side's id (output col 0 = right_id).
let filter_expr = is_null(Arc::new(Column::new("id", 0))).unwrap();
let plan = Arc::new(
FilterExec::try_new(filter_expr, projection).unwrap(),
) as Arc;
// Apply the physical FilterPushdown optimizer
let config = ConfigOptions::default();
let optimized = FilterPushdown::new()
.optimize(Arc::clone(&plan), &config)
.unwrap();
// Execute both plans and compare
let session_ctx = SessionContext::new_with_config(SessionConfig::new());
session_ctx.register_object_store(
ObjectStoreUrl::parse("test://").unwrap().as_ref(),
Arc::new(InMemory::new()),
);
let task_ctx = session_ctx.state().task_ctx();
let original_batches = collect(Arc::clone(&plan), Arc::clone(&task_ctx))
.await
.unwrap();
let original_count: usize =
original_batches.iter().map(|b| b.num_rows()).sum();
let optimized_batches = collect(optimized, Arc::clone(&task_ctx))
.await
.unwrap();
let optimized_count: usize =
optimized_batches.iter().map(|b| b.num_rows()).sum();
// Original plan correctly returns 3 rows (orders 2, 4, 5 — unmatched)
assert_eq!(original_count, 3, "original plan should return 3 rows");
// BUG: optimized plan returns 0 rows — filter was remapped to wrong column
// After fix, change to: assert_eq!(optimized_count, 3);
assert_eq!(optimized_count, 0,
"BUG: filter pushed to wrong column, expected 3 rows but got 0");
}
```
### Expected behavior
The filter should remain mapped to `right_id@2` and correctly filter on the right side's column.
### Additional context
The alias-aware pushdown was added in #19404. Both `column_with_name` and `index_of` are Arrow schema methods that return the first match by design and the code needs to use positional indexing instead. The bug affects any use of the physical plan API where plans are constructed directly instead of building plans via the `LogicalPlanBuilder`. This code path is not exercised through normal SQL because the logical optimizer's PushDownFilter resolves qualified column references and pushes filters below projections before the physical plan is created.
Contributor guide
Research direction
Start with collect_reverse_alias in projection.rs and FilterRemapper::try_remap in filter_pushdown.rs, then run the named regression test in datafusion/core/tests/physical_optimizer/filter_pushdown.rs. Confirm that duplicate output names preserve their positional columns and that the optimized plan returns the same three rows as the original plan.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- rust
- Domain
- backend, data-engineering
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Active
- Clarity
- Clearly specified
- Newbie friendliness
- 72/100