cleanup

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/package.scala

@@ -20,7 +20,6 @@ package org.apache.spark.sql.catalyst
 import java.io._
 import java.nio.charset.StandardCharsets
 
-import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.types.{NumericType, StringType}
 import org.apache.spark.unsafe.types.UTF8String
@@ -138,10 +137,6 @@ package object util {
   // Replaces attributes, string literals, complex type extractors with their pretty form so that
   // generated column names don't contain back-ticks or double-quotes.
   def usePrettyExpression(e: Expression): Expression = e transform {
-    // For unresolved attributes that generated by `DataFrame.col`, we should ignore the generated
-    // qualifiers to not annoy users.
-    case u: UnresolvedAttribute if u.nameParts(0).startsWith("dataframe_") =>
-      new PrettyAttribute(u.copy(nameParts = u.nameParts.drop(1)))
     case a: Attribute => new PrettyAttribute(a)
     case Literal(s: UTF8String, StringType) => PrettyAttribute(s.toString, StringType)
     case Literal(v, t: NumericType) if v != null => PrettyAttribute(v.toString, t)

sql/core/src/main/scala/org/apache/spark/sql/Column.scala

@@ -118,11 +118,6 @@ class Column(protected[sql] val expr: Expression) extends Logging {
    * Returns the expression for this column either with an existing or auto assigned name.
    */
   private[sql] def named: NamedExpression = expr match {
-    // Wrap UnresolvedAttribute with UnresolvedAlias, as when we resolve UnresolvedAttribute, we
-    // will remove intermediate Alias for ExtractValue chain, and we need to alias it again to
-    // make it a NamedExpression.
-    case u: UnresolvedAttribute => UnresolvedAlias(u)
-
     case u: UnresolvedExtractValue => UnresolvedAlias(u)
 
     case expr: NamedExpression => expr
@@ -133,21 +128,28 @@ class Column(protected[sql] val expr: Expression) extends Logging {
 
     case jt: JsonTuple => MultiAlias(jt, Nil)
 
-    case func: UnresolvedFunction => UnresolvedAlias(func, Some(usePrettyExpression(func).sql))
+    case func: UnresolvedFunction => UnresolvedAlias(func, Some(toPresentableString(func)))
 
     // If we have a top level Cast, there is a chance to give it a better alias, if there is a
     // NamedExpression under this Cast.
     case c: Cast => c.transformUp {
       case Cast(ne: NamedExpression, to) => UnresolvedAlias(Cast(ne, to))
     } match {
       case ne: NamedExpression => ne
-      case other => Alias(expr, usePrettyExpression(expr).sql)()
+      case other => Alias(expr, toPresentableString(expr))()
     }
 
-    case expr: Expression => Alias(expr, usePrettyExpression(expr).sql)()
+    case expr: Expression => Alias(expr, toPresentableString(expr))()
   }
 
-  override def toString: String = usePrettyExpression(expr).sql
+  override def toString: String = toPresentableString(expr)
+
+  private def toPresentableString(expr: Expression): String = usePrettyExpression(expr transform {
+    // For unresolved attributes that generated by `DataFrame.col`, we should ignore the generated
+    // qualifiers to not annoy users.
+    case u: UnresolvedAttribute if u.nameParts(0).startsWith("dataframe_") =>
+      u.copy(nameParts = u.nameParts.drop(1))
+  }).sql
 
   override def equals(that: Any): Boolean = that match {
     case that: Column => that.expr.equals(this.expr)

sql/core/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala

@@ -67,14 +67,7 @@ class RelationalGroupedDataset protected[sql](
     }
   }
 
-  // Wrap UnresolvedAttribute with UnresolvedAlias, as when we resolve UnresolvedAttribute, we
-  // will remove intermediate Alias for ExtractValue chain, and we need to alias it again to
-  // make it a NamedExpression.
-  private[this] def alias(expr: Expression): NamedExpression = expr match {
-    case u: UnresolvedAttribute => UnresolvedAlias(u)
-    case expr: NamedExpression => expr
-    case expr: Expression => Alias(expr, usePrettyExpression(expr).sql)()
-  }
+  private[this] def alias(expr: Expression): NamedExpression = Column(expr).named
 
   private[this] def aggregateNumericColumns(colNames: String*)(f: Expression => AggregateFunction)
     : DataFrame = {