`enable_join_dynamic_filter_pushdown` silently drops rows: the pushed filter is remapped by column NAME and lands on the wrong same-named column

## One-line summary

With `datafusion.optimizer.enable_join_dynamic_filter_pushdown = true`, dynamic-filter
pushdown remaps the filter's column **by name** into the probe-side child, so when a
join key references the *second* of two output columns both named `id`, the filter is
silently moved to the *first* `id` column and parquet row-group pruning then discards
all rows. The same plan with the flag `false` returns the correct row.

## DataFusion version

`datafusion 55.1.0` / `arrow 59.3.0` (default features). Also observed through a graph
engine on `datafusion 54.0.0`; the flags/plan shape are unchanged.

## Reproducer

Pure DataFusion — no third-party engine. Two one-row parquet files and a hand-built
logical plan (built with the API rather than SQL because SQL cannot reference the
second of two same-named columns). Save as a test/example with dev-deps
`datafusion = "55.1"`, `arrow = "59"`, `parquet = "59"`, `tempfile`, `tokio`.

```rust
use std::sync::Arc;

use arrow::array::{ArrayRef, StringArray};
use arrow::datatypes::{DataType, Field, Schema};
use arrow::record_batch::RecordBatch;
use datafusion::execution::context::SessionConfig;
use datafusion::logical_expr::{Expr, JoinType, LogicalPlan, LogicalPlanBuilder};
use datafusion::physical_optimizer::PhysicalOptimizerRule;
use datafusion::prelude::*;
use parquet::arrow::ArrowWriter;

fn write_parquet(path: &std::path::Path, schema: Arc, batch: &RecordBatch) {
let file = std::fs::File::create(path).unwrap();
let mut w = ArrowWriter::try_new(file, schema, None).unwrap();
w.write(batch).unwrap();
w.close().unwrap();
}

// ta: single row (id = "a1", ty = "x1") -> id range [a1,a1] does NOT contain x1
// tb: single row (id = "x1")
fn make_tables() -> (tempfile::TempDir, String, String) {
let dir = tempfile::tempdir().unwrap();

let sa = Arc::new(Schema::new(vec![
Field::new("id", DataType::Utf8, false),
Field::new("ty", DataType::Utf8, false),
]));
let b = RecordBatch::try_new(
Arc::clone(&sa),
vec![
Arc::new(StringArray::from(vec!["a1"])) as ArrayRef,
Arc::new(StringArray::from(vec!["x1"])) as ArrayRef,
],
)
.unwrap();
let ta = dir.path().join("ta.parquet");
write_parquet(&ta, sa, &b);

let sb = Arc::new(Schema::new(vec![Field::new("id", DataType::Utf8, false)]));
let b = RecordBatch::try_new(
Arc::clone(&sb),
vec![Arc::new(StringArray::from(vec!["x1"])) as ArrayRef],
)
.unwrap();
let tb = dir.path().join("tb.parquet");
write_parquet(&tb, sb, &b);

(
dir,
ta.to_string_lossy().into_owned(),
tb.to_string_lossy().into_owned(),
)
}

async fn ctx(dynamic: bool, ta: &str, tb: &str) -> SessionContext {
let mut config = SessionConfig::new();
config.options_mut().execution.target_partitions = 1;
config.options_mut().execution.collect_statistics = true;
config.options_mut().execution.parquet.schema_force_view_types = false;
config.options_mut().optimizer.join_reordering = false;
config.options_mut().optimizer.enable_join_dynamic_filter_pushdown = dynamic;
let ctx = SessionContext::new_with_config(config);
ctx.register_parquet("ta", ta, ParquetReadOptions::default()).await.unwrap();
ctx.register_parquet("tb", tb, ParquetReadOptions::default()).await.unwrap();
ctx
}

fn c(rel: &str, name: &str) -> Expr {
Expr::Column(datafusion::common::Column::new(Some(rel), name))
}
fn k(rel: &str, name: &str) -> datafusion::common::Column {
datafusion::common::Column::new(Some(rel), name)
}

fn build_logical(scan_a: LogicalPlan, scan_b: LogicalPlan) -> LogicalPlan {
let aliased = |scan: LogicalPlan, name: &str, exprs: Vec| {
LogicalPlanBuilder::from(scan)
.alias(name).unwrap()
.project(exprs).unwrap()
.build().unwrap()
};
let a = aliased(scan_a, "a", vec![c("a", "id"), c("a", "ty")]);
let b = aliased(scan_b.clone(), "b", vec![c("b", "id")]);
let cc = aliased(scan_b.clone(), "c", vec![c("c", "id")]);

// A = a JOIN b ON a.ty = b.id -> [a.id, a.ty, b.id]
let a_join = LogicalPlanBuilder::from(a)
.join_detailed(
b, JoinType::Inner,
(vec![k("a", "ty")], vec![k("b", "id")]),
None, datafusion::common::NullEquality::NullEqualsNothing,
).unwrap().build().unwrap();

// A2 = [a.id, b.id] -> TWO output columns both named `id`
let a2 = LogicalPlanBuilder::from(a_join)
.project(vec![c("a", "id"), c("b", "id")])
.unwrap().build().unwrap();

// J = A2 JOIN c ON b.id = c.id -> [a.id, b.id, c.id]
let j = LogicalPlanBuilder::from(a2)
.join_detailed(
cc, JoinType::Inner,
(vec![k("b", "id")], vec![k("c", "id")]),
None, datafusion::common::NullEquality::NullEqualsNothing,
).unwrap().build().unwrap();

// s = filter(tb, id = 'x1') -> [s.id]
let s = LogicalPlanBuilder::from(scan_b)
.alias("s").unwrap()
.filter(c("s", "id").eq(lit("x1"))).unwrap()
.project(vec![c("s", "id")]).unwrap()
.build().unwrap();

// Top = s JOIN J ON s.id = J. -- J's SECOND `id` (index 1)
LogicalPlanBuilder::from(s)
.join_detailed(
j, JoinType::Inner,
(vec![k("s", "id")], vec![k("b", "id")]),
None, datafusion::common::NullEquality::NullEqualsNothing,
).unwrap().build().unwrap()
}

async fn run(dynamic: bool) -> (Arc, Vec) {
let (_dir, ta, tb) = make_tables();
let ctx = ctx(dynamic, &ta, &tb).await;
let scan_a = ctx.table("ta").await.unwrap().logical_plan().clone();
let scan_b = ctx.table("tb").await.unwrap().logical_plan().clone();
let logical = build_logical(scan_a, scan_b);

// NOTE: logical optimization is bypassed so the hand-built join tree is not
// reordered; the physical optimizer (incl. dynamic-filter pushdown) still runs.
let state = ctx.state();
let mut phys = state.query_planner()
.create_physical_plan(&logical, &state).await.unwrap();
for rule in state.physical_optimizers() {
phys = rule.optimize(phys, state.config_options()).unwrap();
}
let batches = datafusion::physical_plan::collect(
Arc::clone(&phys), ctx.task_ctx(),
).await.unwrap();
(phys, batches)
}

#[tokio::test]
async fn dynamic_filter_must_not_change_the_answer() {
let (_, on) = run(true).await;
let (_, off) = run(false).await;
let rows = |b: &[RecordBatch]| b.iter().map(|x| x.num_rows()).sum::();
assert_eq!(rows(&on), rows(&off),
"a physical optimizer flag changed the answer: ON={} OFF={}",
rows(&on), rows(&off));
}
```

## Expected output

Both settings return the single matching row (`s.id = x1`, `a.id = a1`, `b.id = x1`,
`c.id = x1`):

```
ROWS(ON) = 1
ROWS(OFF) = 1
```

## Actual output (datafusion 55.1.0)

```
--- flag ON ---
ROWS(ON) = 0
--- flag OFF ---
ROWS(OFF) = 1
["x1", "a1", "x1", "x1"]
```

## Physical plans

### Flag ON

A `DynamicFilter` has been pushed onto `ta.parquet` (which contains only
`id = a1`); the filter's values come from `s.id = x1`, so parquet row-group pruning
(relying on the wrong column) removes the only `ta` row and the query returns zero rows.

```
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, id@1)]
FilterExec: id@0 = x1
DataSourceExec: file_groups={1 group: [[.../tb.parquet]]}, projection=[id], file_type=parquet, predicate=id@0 = x1 AND id@0 = x1, ...
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@1, id@0)]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(ty@1, id@0)], projection=[id@0, id@2]
DataSourceExec: file_groups={1 group: [[.../ta.parquet]]}, projection=[id, ty], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
DataSourceExec: file_groups={1 group: [[.../tb.parquet]]}, projection=[id], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
DataSourceExec: file_groups={1 group: [[.../tb.parquet]]}, projection=[id], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
```

### Flag OFF

Identical plan with **no** `DynamicFilter` anywhere:

```
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, id@1)]
FilterExec: id@0 = x1
DataSourceExec: file_groups={1 group: [[.../tb.parquet]]}, projection=[id], file_type=parquet, predicate=id@0 = x1 AND id@0 = x1, ...
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@1, id@0)]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(ty@1, id@0)], projection=[id@0, id@2]
DataSourceExec: file_groups={1 group: [[.../ta.parquet]]}, projection=[id, ty], file_type=parquet
DataSourceExec: file_groups={1 group: [[.../tb.parquet]]}, projection=[id], file_type=parquet
DataSourceExec: file_groups={1 group: [[.../tb.parquet]]}, projection=[id], file_type=parquet
```

## Scope: reachable through the `DataFrame` / `LogicalPlanBuilder` API, not through `ctx.sql`

I checked this deliberately, because the reproducer below builds its plan with `LogicalPlanBuilder`
and a reader could reasonably wonder whether it only happens to hand-built plans.

**It is not an artefact of skipping optimization.** The originating case is a production query engine
that builds plans through the `DataFrame` API and then calls `SessionState::create_physical_plan` on
the result: the full pipeline, analyzer and logical optimizer included, nothing bypassed. The misroute
happens there, with optimization fully enabled:

```
HashJoinExec: on=[(id@0, id@1)], projection=[id@1, id@3]
FilterExec: id@0 = x1
HashJoinExec: on=[(end_id@2, id@0)], projection=[id@0, id@1, id@3]
...
DataSourceExec: test.parquet, projection=[id], predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
```

Flag on returns no rows; flag off returns the correct one. The minimal reproducer below skips the
logical optimizer only to keep the case small, not to make it happen.

**Plain SQL does NOT reproduce it.** I tried five shapes through `ctx.sql` — a left-deep three-join
chain keyed on a qualifier, a derived table carrying a duplicate `id` with the top join on `j.id`,
three left-deep joins on `b.id`, and two parenthesized variants placing the chain on either side.
All five returned the same row count with the option on and off. The SQL path attaches a
`projection=[id@1]` to the join and drops the duplicate `id` before the downstream join, so the second
same-named physical field the misroute needs never survives; derived tables also lose the qualifier
needed to name the second `id`.

So the precondition is **two same-named columns surviving into a downstream join key**, which SQL's
projection pushdown happens to eliminate and the builder API does not. That is unremarkable for a graph
engine, where nearly every output column is named `id`, so duplicate names are the normal case rather
than a contrived one.

## Probable root cause

`datafusion-physical-plan/src/filter_pushdown.rs`, `FilterRemapper::try_remap`
(~lines 365-385). When routing a pushed-down filter into a child it validates the
column position against `allowed_indices` but then **remaps by name**:

```rust
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))))
}
```

`Schema::index_of(name)` returns the *first* field with that name. The join key created
by `HashJoinExec::gather_filters_for_pushdown` is
`Column { name: "id", index: 1 }` (the second `id` of `A2`). Pushing it into `A2`/`A`
resolves the name `id` to index 0 (`a.id`) instead of index 1 (`b.id`), so the dynamic
filter is attached to the wrong parquet scan. `dynamic_rg_pruning` then evaluates
`x1` against `ta.id` min/max `[a1,a1]` and prunes the row group, returning no rows.
The `allowed_indices` guard added for same-named join sides does not help because it
only gates *which child* is eligible; the actual index selection is still name-based.

A positional remap (e.g. carry the parent-output index → child-input index mapping from
the join's `column_indices`/`projection`) would fix it.

## Same shape in a graph engine (context, not required to reproduce)

The originating incident was a Cypher multi-hop expansion whose endpoint (`w.id`) was a
join-output column equated to a separately bound node (`mid.id`). There every output
column is named `id`, and the same mechanism pushed the dynamic filter from the top
join down to the *base* `test.parquet` scan:

```
HashJoinExec: on=[(id@0, id@1)], projection=[id@1, id@3]
FilterExec: id@0 = x1 <- spec WHERE id='x1'
HashJoinExec: on=[(end_id@2, id@0)], projection=[id@0, id@1, id@3]
...
DataSourceExec: test.parquet, projection=[id], predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
```

`ROWS(ON) = []` vs `ROWS(OFF) = [("t2","g1")]`.