panic when GROUP BY column order doesn't match USING column order #4873

#[cfg(test)]
mod test_cross_join_bug2 {
    use std::sync::Arc;
    use datafusion::arrow::array::{ArrayRef, StringArray, UInt64Array};
    use datafusion::arrow::datatypes::{DataType, Field, Schema, SchemaRef};
    use datafusion::arrow::record_batch::RecordBatch;
    use datafusion::arrow::util::pretty::pretty_format_batches;
    use datafusion::datasource::MemTable;
    use datafusion::prelude::SessionContext;

    #[tokio::test]
    async fn count_distinct_error() {
        let col1 = Arc::new(StringArray::from(vec![
            "A", "A", "A", "A",
            "A", "A", "A", "A",
            "BB", "BB", "BB", "BB",
        ])) as ArrayRef;

        let col2 = Arc::new(UInt64Array::from(vec![
            1, 1, 2, 2, 3, 3, 4, 4, 5, 5, 6, 6
        ])) as ArrayRef;

        let col3 = Arc::new(UInt64Array::from(vec![
            1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1
        ])) as ArrayRef;

        let schema = Arc::new(Schema::new(vec![
            Field::new("col1", DataType::Utf8, true),
            Field::new("col2", DataType::UInt64, true),
            Field::new("col3", DataType::UInt64, true),
        ])) as SchemaRef;

        let batch =
            RecordBatch::try_new(schema.clone(), vec![
                col1, col2, col3,
            ]).unwrap();
        let mem_table = MemTable::try_new(schema, vec![vec![batch]]).unwrap();

        // Create context and register table
        let ctx = SessionContext::new();
        ctx.register_table("tbl", Arc::new(mem_table)).unwrap();

        // Pretty print the input table
        let res = ctx
            .sql("SELECT * from tbl")
            .await
            .unwrap()
            .collect()
            .await
            .unwrap();
        let formatted = pretty_format_batches(res.as_slice()).unwrap();
        println!("{}", formatted);

        // Perform query where GROUP BY and USING both specify "col1", "col2" (works)
        let sql1 = r#"
SELECT "col1", "col2", coalesce("sum_col3", 0) as sum_col3
FROM (SELECT DISTINCT "col2" FROM "tbl") AS "subq1"
CROSS JOIN (SELECT DISTINCT "col1" FROM "tbl") AS "subq2"
LEFT OUTER JOIN (SELECT "col1", "col2", sum("col3") as "sum_col3" FROM "tbl" GROUP BY "col1", "col2") AS "subq3"
USING("col1", "col2")
    "#;

        let res = ctx.sql(sql1).await.unwrap().collect().await.unwrap();
        let formatted = pretty_format_batches(res.as_slice()).unwrap();
        println!("{}", formatted);

        // Perform query where GROUP BY specifies "col1", "col2" and USING specifies "col2", "col1" (panic)
        let sql2 = r#"
SELECT "col1", "col2", coalesce("sum_col3", 0) as sum_col3
FROM (SELECT DISTINCT "col2" FROM "tbl") AS "subq1"
CROSS JOIN (SELECT DISTINCT "col1" FROM "tbl") AS "subq2"
LEFT OUTER JOIN (SELECT "col1", "col2", sum("col3") as "sum_col3" FROM "tbl" GROUP BY "col1", "col2") AS "subq3"
USING("col2", "col1")
    "#;

        let res = ctx.sql(sql2).await.unwrap().collect().await.unwrap();
        let formatted = pretty_format_batches(res.as_slice()).unwrap();
        println!("{}", formatted);
    }
}