Leon-Weigel opened a new issue, #25244:
URL: https://github.com/apache/datafusion/issues/25244
## 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<Schema>, 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<Expr>| {
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.<b.id> -- 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<dyn
datafusion::physical_plan::ExecutionPlan>, Vec<RecordBatch>) {
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::<usize>();
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
```
## Caveat on the reproducer's planning path
The reproducer builds the join tree with `LogicalPlanBuilder` and bypasses
the LOGICAL optimizer, so the
hand-built tree is not reordered; the physical optimizer, including
dynamic-filter pushdown, runs normally.
That is deliberate: plain SQL cannot reference the *second* of two
same-named output columns, so SQL cannot
express the join key that triggers this. The same misroute occurs through
the ordinary end-to-end path in the
graph engine described at the bottom, where the duplicate `id` names arise
naturally, so the bypass is a way to
make the case small rather than a way to make it happen.
## 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")]`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
