SQL case, the subquery plan inside the sub query expression does not get a chance to run all the optimizer rules

[jiangzhx]

Describe the bug

dataframe: TableScan node added a unexpected projection projection=[a]

dataframe:

+--------------+-------------------------------------------------------+
| plan_type    | plan                                                  |
+--------------+-------------------------------------------------------+
| logical_plan | Filter: EXISTS (<subquery>)                           |
|              |   Subquery:                                           |
|              |     Aggregate: groupBy=[[]], aggr=[[COUNT(UInt8(1))]] |
|              |       TableScan: t2 projection=[a]                    |
|              |   TableScan: t1 projection=[a, b]                     |
+--------------+-------------------------------------------------------+

sql

+--------------+---------------------------------------------------------+
| plan_type    | plan                                                    |
+--------------+---------------------------------------------------------+
| logical_plan | Filter: EXISTS (<subquery>)                             |
|              |   Subquery:                                             |
|              |     Projection: COUNT(UInt8(1))                         |
|              |       Aggregate: groupBy=[[]], aggr=[[COUNT(UInt8(1))]] |
|              |         TableScan: t2                                   |
|              |   TableScan: t1 projection=[a, b]                       |
+--------------+---------------------------------------------------------+

To Reproduce

#[tokio::test]
async fn test_count_wildcard_on_where_exist() -> Result<()> {
    // you can find create_join_context at datafusion/core/tests/dataframe.rs
    let ctx = create_join_context()?;
    let sql_results = ctx
        .sql("SELECT a, b FROM t1 WHERE EXISTS (SELECT count(*) FROM t2)")
        .await?
        .explain(false, false)?
        .collect()
        .await?;

    let df_results = ctx
        .table("t1")
        .await?
        .filter(Expr::Exists {
            subquery: Subquery {
                subquery: Arc::new(
                    ctx.table("t2")
                        .await?
                        .aggregate(vec![], vec![count(Expr::Wildcard)])?
                        .select(vec![count(Expr::Wildcard)])?
                        .into_optimized_plan()?,
                ),
                outer_ref_columns: vec![],
            },
            negated: false,
        })?
        .select(vec![col("a"), col("b")])?
        .explain(false, false)?
        .collect()
        .await?;

    assert_eq!(
        pretty_format_batches(&sql_results)?.to_string(),
        pretty_format_batches(&df_results)?.to_string()
    );
    Ok(())
}

Expected behavior

No response

Additional context

No response

[jiangzhx]

Filter: EXISTS ()
Subquery:
Projection: COUNT(UInt8(1)) //remove by eliminate_projection
Aggregate: groupBy=[[]], aggr=[[COUNT(UInt8(1))]]
TableScan: t2 projection=[a] //append by push_down_projection
TableScan: t1 projection=[a, b]

[mingmwang]

I think there are 2 issues here.

1. In the SQL case, the subquery plan inside the sub query expression does not get a chance to run all the optimizer rules,
   this is a known issue.
2. In the DataFrame case, after apply the push_down_projection rule, the subquery plan added unnecessary projection

[jiangzhx]

I think there are 2 issues here.

1. In the SQL case, the subquery plan inside the sub query expression does not get a chance to run all the optimizer rules,
   this is a known issue.
2. In the DataFrame case, after apply the push_down_projection rule, the subquery plan added unnecessary projection

Thank you for confirming my guess, I will try to fix the push_down_projection unexpected projection.

the second one not actually a bug, because when only count(*) is queried, push_down_projection will default to querying the first column.

https://github.com/apache/arrow-datafusion/blob/c8a3d589889dd1e67047de89db8b4ff56f90f04c/datafusion/optimizer/src/push_down_projection.rs#L506-L510

[mingmwang]

Well, you are right and this is not a bug, but I think it is an improvement point, if the backend are parquet files, we can just read the parquet files footers instead of reading any real fields.