fix toString

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

@@ -20,6 +20,7 @@ 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
@@ -137,6 +138,10 @@ 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

@@ -133,31 +133,21 @@ class Column(protected[sql] val expr: Expression) extends Logging {
 
     case jt: JsonTuple => MultiAlias(jt, Nil)
 
-    case func: UnresolvedFunction => UnresolvedAlias(func, Some(presentableExpression(func).sql))
+    case func: UnresolvedFunction => UnresolvedAlias(func, Some(usePrettyExpression(func).sql))
 
     // 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, presentableExpression(expr).sql)()
+      case other => Alias(expr, usePrettyExpression(expr).sql)()
     }
 
-    case expr: Expression => Alias(expr, presentableExpression(expr).sql)()
+    case expr: Expression => Alias(expr, usePrettyExpression(expr).sql)()
   }
 
-  override def toString: String = presentableExpression(expr).sql
-
-  private def presentableExpression(expr: Expression): Expression = {
-    usePrettyExpression(expr transform {
-      case u: UnresolvedAttribute if u.nameParts.head.startsWith(Dataset.namePrefix) =>
-        u.copy(nameParts = u.nameParts.drop(1))
-
-      case a: AttributeReference if a.qualifiers.length == 1 &&
-        a.qualifiers.head.startsWith(Dataset.namePrefix) => a.withQualifiers(Nil)
-    })
-  }
+  override def toString: String = usePrettyExpression(expr).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/Dataset.scala

@@ -62,9 +62,7 @@ private[sql] object Dataset {
 
   private[this] val nextDataFrameId = new AtomicLong(0)
 
-  val namePrefix: String = "dataframe_"
-
-  def newDataFrameName: String = namePrefix + nextDataFrameId.getAndIncrement()
+  def newDataFrameName: String = s"dataframe_${nextDataFrameId.getAndIncrement()}"
 }
 
 /**