greg-semvia commented on issue #13156:
URL: https://github.com/apache/datafusion/issues/13156#issuecomment-5196792301
This is filed as a performance improvement, but for outer joins it is also a
correctness bug, and I think that is worth recording here since the issue is
the tracking point for the fix.
Folding a filter into `ON` is equivalence-preserving only when the filtered
input is *not* a preserved side of the join. On a preserved side the two
placements mean different things: below the join the predicate restricts the
input, in `ON` it only gates match eligibility, so every row that fails it
survives NULL-padded instead of disappearing. For `LEFT JOIN` that is the left
input, for `RIGHT JOIN` the right, and for `FULL JOIN` both. The `INNER JOIN`
case in the issue body is sound, and the non-preserved side of an outer join is
sound too.
#21694 fixed the inner-join half and states that "outer joins require
additional work" and that for LEFT/RIGHT/FULL "the existing behavior is
preserved" — so the case below is still live on 53.1.0 and on `main` (I checked
the source; the most recent change to `datafusion/sql/src/unparser/plan.rs` is
unrelated).
### Repro
Standalone, no external data. `datafusion = "53.1.0"`, `tokio = { version =
"1", features = ["macros", "rt-multi-thread"] }`.
```rust
use std::sync::Arc;
use datafusion::arrow::array::StringArray;
use datafusion::arrow::datatypes::{DataType, Field, Schema};
use datafusion::arrow::record_batch::RecordBatch;
use datafusion::common::JoinType;
use datafusion::error::Result;
use datafusion::logical_expr::{col, lit};
use datafusion::prelude::SessionContext;
use datafusion::sql::unparser::dialect::PostgreSqlDialect;
use datafusion::sql::unparser::Unparser;
#[tokio::main]
async fn main() -> Result<()> {
let ctx = SessionContext::new();
// Three left rows, exactly one of which satisfies the filter.
let left_schema = Arc::new(Schema::new(vec![
Field::new("id", DataType::Utf8, false),
Field::new("target", DataType::Utf8, false),
]));
let left = RecordBatch::try_new(
left_schema,
vec![
Arc::new(StringArray::from(vec!["a", "b", "c"])),
Arc::new(StringArray::from(vec!["keep", "drop", "drop"])),
],
)?;
let right_schema = Arc::new(Schema::new(vec![
Field::new("id", DataType::Utf8, false),
Field::new("name", DataType::Utf8, true),
]));
let right = RecordBatch::try_new(
right_schema,
vec![
Arc::new(StringArray::from(vec!["a", "b"])),
Arc::new(StringArray::from(vec!["alpha", "beta"])),
],
)?;
ctx.register_batch("l", left)?;
ctx.register_batch("r", right)?;
// Filter on the PRESERVED side of a LEFT JOIN.
let plan = ctx
.table("l")
.await?
.filter(col("target").eq(lit("keep")))?
.join_on(
ctx.table("r").await?,
JoinType::Left,
[col("l.id").eq(col("r.id"))],
)?
.select(vec![col("l.id"), col("r.name")])?
.into_unoptimized_plan();
println!("--- LogicalPlan ---\n{}", plan.display_indent());
let via_plan =
ctx.execute_logical_plan(plan.clone()).await?.collect().await?;
let plan_rows: usize = via_plan.iter().map(|b| b.num_rows()).sum();
let sql = Unparser::new(&PostgreSqlDialect {})
.plan_to_sql(&plan)?
.to_string();
println!("--- plan_to_sql ---\n{sql}");
let via_sql = ctx.sql(&sql).await?.collect().await?;
let sql_rows: usize = via_sql.iter().map(|b| b.num_rows()).sum();
println!("--- row counts ---\nplan: {plan_rows}\nsql: {sql_rows}");
Ok(())
}
```
Output:
```
--- LogicalPlan ---
Projection: l.id, r.name
Left Join: Filter: l.id = r.id
Filter: l.target = Utf8("keep")
TableScan: l
TableScan: r
--- plan_to_sql ---
SELECT "l"."id", "r"."name" FROM "l" LEFT OUTER JOIN "r" ON (("l"."id" =
"r"."id") AND ("l"."target" = 'keep'))
--- row counts ---
plan: 1
sql: 3
```
The plan and its unparsing disagree on the answer, not on the shape.
Two things narrow it further. The plan here is **unoptimized** — a plain
`Filter` node above the scan, with nothing folded into `TableScan.filters` — so
the defect is not confined to the pushdown path the issue title names, and
there is no plan-preparation precondition that avoids it. And executing the
same `LogicalPlan` directly returns the right answer, so this is purely SQL
generation.
### The proposed fix also fixes this
Wrapping the preserved side in a `SubqueryAlias` produces the derived table
this issue asks for, and with it the correct row count:
```rust
let aliased = ctx
.table("l")
.await?
.filter(col("target").eq(lit("keep")))?
.select(vec![col("id")])?
.alias("preserved")?
.join_on(
ctx.table("r").await?,
JoinType::Left,
[col("preserved.id").eq(col("r.id"))],
)?
.select(vec![col("preserved.id"), col("r.name")])?
.into_unoptimized_plan();
```
```
SELECT "preserved"."id", "r"."name" FROM (SELECT "l"."id" FROM "l" WHERE
("l"."target" = 'keep')) AS "preserved" LEFT OUTER JOIN "r" ON
("preserved"."id" = "r"."id")
sql: 1
```
So the derived-table rendering is not only the better plan for the source,
it is the only correct one when the filtered input is preserved. A narrower
version of the same fix — fold into `ON` only when the filtered input is not a
preserved side of the join type, and emit a derived table otherwise — would
close the correctness half without changing inner-join output at all.
### Why the row counts matter
I hit this generating a single SQL statement from a plan whose `LEFT JOIN`
preserved side carried a ground equality. On the real source the hoisted form
returned **482,824 rows** — the full extent of the preserved relation — where
the correct answer is **219**. Nothing errors and nothing looks malformed; the
result set is well-typed, plausibly shaped, and wrong by three orders of
magnitude. That failure mode is why I would rather see this treated as a
correctness issue than a performance one: a slow query announces itself, and
this does not.
We route around it locally by emitting the preserved side as an explicit
`SubqueryAlias`, as above, so this is not blocking on our end — posting the
repro in case it is useful to whoever picks up the outer-join half.
--
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]
