Unparsing of CROSS JOINs with filters is generating incorrect queries

[nuno-faria]

Describe the bug

The unparsing of CROSS JOINS with filters is generating invalid SQL, when building from an optimized plan.

To Reproduce

Example:

-- original
select t1.k, t2.v
from t1, t2
where t2.k = 0;

-- optimized plan
Cross Join:
  TableScan: t1 projection=[k]
  Projection: t2.v
    Filter: t2.k = Int32(0)
      TableScan: t2 projection=[k, v]

-- generated; the repeated filter is not valid since k is not projected from the subquery
SELECT t1.k, t2.v FROM t1 CROSS JOIN (SELECT t2.v FROM t2 WHERE (t2.k = 0)) WHERE (t2.k = 0)

If we filter by a projected column (t2.v) it works correctly (although it would make more sense for the filter to be done on a WHERE instead of on the JOIN, but it works):

-- original
select t1.k, t2.v
from t1, t2
where t2.v = 0;

-- optimized plan
Cross Join: 
  TableScan: t1 projection=[k]
  Filter: t2.v = Int32(0)
    TableScan: t2 projection=[v]

-- generated
SELECT t1.k, t2.v FROM t1 INNER JOIN t2 ON (t2.v = 0)

Expected behavior

Generate valid SQL.

Additional context

No response

[jonathanc-n]

I'm a bit confused, I believe the logical plan should have already pushed down the filter into the subquery, removing the filter from the entire join. When I ran a test in datafusion/sql/tests/cases/plan_to_sql.rs, it passed fine:

#[test]
fn roundtrip_cross_join_right_filter_bug_dialect2() -> Result<(), DataFusionError> {
   // Repros the buggy output (outer WHERE duplicated), formatted like dialect 2.
   roundtrip_statement_with_dialect_helper!(
       sql: "select j1.j1_id, j2.j2_string from j1, j2 where j2.j2_id = 0;",
       parser_dialect: GenericDialect {},
       unparser_dialect: UnparserDefaultDialect {},
       expected: @"SELECT j1.j1_id, j2.j2_string FROM j1 CROSS JOIN j2 WHERE (j2.j2_id = 0)",
   );
   Ok(())
}

If you manually create the plan like so below the test will reproduce but this shouldn't happen in the first place from the original query but I think i'm missing something:

#[test]
fn test_unparse_cross_join() -> Result<()> {
    let schema = Schema::new(vec![
        Field::new("k", DataType::Int32, false),
        Field::new("v", DataType::Int32, false),
    ]);

    // Cross Join:
    //   TableScan: t1 projection=[k]
    //   Projection: t2.v
    //     Filter: t2.k = Int32(0)
    //       TableScan: t2 projection=[k, v]
    let t1_scan = table_scan(Some("t1"), &schema, Some(vec![0]))?.build()?;

    let t2_scan = table_scan(Some("t2"), &schema, Some(vec![0, 1]))?.build()?;
    let t2_proj = LogicalPlanBuilder::from(t2_scan)
        .filter(col("t2.k").eq(lit(0)))?
        .project(vec![col("t2.v")])?
        .build()?;
    let plan = LogicalPlanBuilder::from(t1_scan)
        .cross_join(t2_proj)?
        .filter(col("t2.k").eq(lit(0)))?
        .project(vec![col("t1.k"), col("t2.v")])?
        .build()?;

    let unparser = Unparser::new(&UnparserPostgreSqlDialect {});
    let sql = unparser.plan_to_sql(&plan)?;

    assert_snapshot!(
        sql,
        @r#"SELECT "t1"."k", "t2"."v" FROM "t1" CROSS JOIN (SELECT "t2"."v" FROM "t2" WHERE ("t2"."k" = 0))"#
    );

    Ok(())
}

However this would be incorrect as the outside filter is still kept even though it would have been pushed down. Are you able to make a specific test that I can run somewhere that reproduces this?

[nuno-faria]

@jonathanc-n thanks for looking into it. Here is a complete example:

use datafusion::error::Result;
use datafusion::prelude::SessionContext;
use datafusion::sql::unparser::plan_to_sql;

#[tokio::main]
async fn main() -> Result<()> {
    let ctx = SessionContext::new();

    ctx.sql("create table j1 (j1_id int, j1_string varchar)")
        .await?
        .collect()
        .await?;
    ctx.sql("create table j2 (j2_id int, j2_string varchar)")
        .await?
        .collect()
        .await?;

    let df = ctx
        .sql(
            "select j1.j1_id, j2.j2_string
            from j1, j2
            where j2.j2_id = 0;",
        )
        .await?;

    // unoptimized
    println!("unoptimized: {}", plan_to_sql(df.logical_plan())?);

    // optimized
    let optimized_plan = df.into_optimized_plan()?;
    println!("optimized: {}", plan_to_sql(&optimized_plan)?);

    Ok(())
}

unoptimized: SELECT j1.j1_id, j2.j2_string FROM j1 CROSS JOIN j2 WHERE (j2.j2_id = 0)
optimized: SELECT j1.j1_id, j2.j2_string FROM j1 CROSS JOIN (SELECT j2.j2_string FROM j2 WHERE (j2.j2_id = 0)) WHERE (j2.j2_id = 0)

The problem occurs when unparsing an optimized plan, while unparsing the original unoptimized plan works well.

Here are the plans for reference:

-- unoptimized
Projection: j1.j1_id, j2.j2_string
  Filter: j2.j2_id = Int64(0)
    Cross Join:
      TableScan: j1
      TableScan: j2

-- optimized
Cross Join:
  TableScan: j1 projection=[j1_id]
  Projection: j2.j2_string
    Filter: j2.j2_id = Int32(0)
      TableScan: j2 projection=[j2_id, j2_string]

[chenkovsky]

take

[jonathanc-n]

@chenkovsky Is it possible for me to do this one? I had forgotten to 'take', but I was about to get around to it. It is fine if that is not possible, sorry about that.

[chenkovsky]

untake