Do not double alias Exprs

datafusion/expr/src/expr.rs

@@ -412,6 +412,19 @@ impl Alias {
             name: name.into(),
         }
     }
+
+    /// Return true if this alias represents the unaliased `name`
+    pub fn eq_name(&self, name: &str) -> bool {
+        self.eq_qualified_name(None, name)
+    }
+
+    pub fn eq_qualified_name(
+        &self,
+        relation: Option<&TableReference>,
+        name: &str,
+    ) -> bool {
+        self.relation.as_ref() == relation && self.name == name
+    }
 }
 
 /// Binary expression
@@ -1270,6 +1283,8 @@ impl Expr {
 
     /// Ensure `expr` has the name as `original_name` by adding an
     /// alias if necessary.
+    ///
+    // TODO: remove? and deprecate??
     pub fn alias_if_changed(self, original_name: String) -> Result<Expr> {
         let new_name = self.name_for_alias()?;
         if new_name == original_name {
@@ -1281,7 +1296,7 @@ impl Expr {
 
     /// Return `self AS name` alias expression
     pub fn alias(self, name: impl Into<String>) -> Expr {
-        Expr::Alias(Alias::new(self, None::<&str>, name.into()))
+        self.alias_qualified(None as Option<&str>, name)
     }
 
     /// Return `self AS name` alias expression with a specific qualifier
@@ -1290,7 +1305,14 @@ impl Expr {
         relation: Option<impl Into<TableReference>>,
         name: impl Into<String>,
     ) -> Expr {
-        Expr::Alias(Alias::new(self, relation, name.into()))
+        // Don't nest aliases
+        let Expr::Alias(mut alias) = self else {
+            return Expr::Alias(Alias::new(self, relation, name));
+        };
+
+        alias.relation = relation.map(|r| r.into());
+        alias.name = name.into();
+        Expr::Alias(alias)
     }
 
     /// Remove an alias from an expression if one exists.
@@ -2831,7 +2853,6 @@ pub fn physical_name(expr: &Expr) -> Result<String> {
         _ => Ok(expr.schema_name().to_string()),
     }
 }
-
 #[cfg(test)]
 mod test {
     use crate::expr_fn::col;

datafusion/optimizer/tests/optimizer_integration.rs

@@ -312,7 +312,7 @@ fn eliminate_redundant_null_check_on_count() {
         GROUP BY col_int32
         HAVING c IS NOT NULL";
     let plan = test_sql(sql).unwrap();
-    let expected = "Projection: test.col_int32, count(Int64(1)) AS count(*) AS c\
+    let expected = "Projection: test.col_int32, count(Int64(1)) AS c\
     \n  Aggregate: groupBy=[[test.col_int32]], aggr=[[count(Int64(1))]]\
     \n    TableScan: test projection=[col_int32]";
     assert_eq!(expected, format!("{plan}"));

datafusion/sqllogictest/test_files/aggregate.slt

@@ -6640,3 +6640,16 @@ SELECT a, median(b), arrow_typeof(median(b)) FROM group_median_all_nulls GROUP B
 ----
 group0 NULL Int32
 group1 NULL Int32
+
+# should not be a double alias
+query TT
+explain SELECT count(*) order by count(*);
+----
+logical_plan
+01)Sort: count(*) ASC NULLS LAST
+02)--Projection: count(Int64(1)) AS count(*)
+03)----Aggregate: groupBy=[[]], aggr=[[count(Int64(1))]]
+04)------EmptyRelation
+physical_plan
+01)ProjectionExec: expr=[1 as count(*)]
+02)--PlaceholderRowExec