A problem about the projection_push_down optimizer gathers valid columns

[ic4y]

Describe the bug
why gather all columns needed for expressions in this Aggregate by using aggr_expr, rather than schema.
An error occurs when the schema is generated by a custom optimizer and does not match the column name in aggr_expr.

https://github.com/apache/arrow-datafusion/blob/0b6db52435c7c955b51e44de8a3eae58e53716d0/datafusion/src/optimizer/projection_push_down.rs#L292-L305

Nothing will go wrong like this

            schema
                .fields()[group_expr.len()..].to_vec()
                .iter()
                .enumerate()
                .try_for_each(|(i, field)| {
                    if required_columns.contains(&field.qualified_column()) {
                        new_aggr_expr.push(aggr_expr[i].clone());
                        utils::expr_to_columns(&aggr_expr[i], &mut new_required_columns)
                    } else {
                        Ok(())
                    }
                })?;

[houqp]

I think this is probably written with the assumption that schema is always in sync with aggr_expr. @ic4y could you give a more concrete example on how a custom optimizer would change the schema without updating aggr_expr?

[ic4y]

@houqp
Like the following, I added group by for distinct Aggregate, and then deleted distinct in count. At this time, aggr_expr has changed, but I don’t want to rewrite the schema, because subsequent Plan Nodes will use this schema like ProjectionNode

Projection: #COUNT(DISTINCT lineorder_flat.lo_orderkey) AS a [a:UInt64;N]
  Aggregate: groupBy=[[]], aggr=[[COUNT(DISTINCT #lineorder_flat.lo_orderkey)]] [COUNT(DISTINCT lineorder_flat.lo_orderkey):UInt64;N]
    TableScan: lineorder_flat projection=Some([0]) [lo_orderkey:Int64;N]

-------------------custom optimizer------------------------------------------
Projection: #COUNT(DISTINCT lineorder_flat.lo_orderkey) AS a [a:UInt64;N]
  Aggregate: groupBy=[[]], aggr=[[COUNT(#lineorder_flat.lo_orderkey)]] [COUNT(DISTINCT lineorder_flat.lo_orderkey):UInt64;N]
    Aggregate: groupBy=[[#lineorder_flat.lo_orderkey]], aggr=[[]] [lo_orderkey:Int64;N]
      TableScan: lineorder_flat projection=Some([0]) [lo_orderkey:Int64;N]

[alamb]

This sounds very similar to #1316, which @viirya is working on in #1319 I believe

[ic4y]

I used projection output relation alias to solve this problem