Improve TableScan with the pushdown filter unparsing in Join #13156

Activity

1. goldmedal

   added

   enhancementNew feature or requestNew feature or request

   on Oct 29, 2024
2. pull

   added a commit that references this issue on Apr 28, 2026

   feat(unparser): Keep inner join `Filter → TableScan` predicates to `W…

   ...

   c2c0773
3. greg-semvia commented on Aug 5, 2026

   

   greg-semvia

   on Aug 5, 2026

   More actions

   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"] }.

   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:

   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.