fixup

Author: HyukjinKwon

connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/Dataset.scala

@@ -414,11 +414,7 @@ class Dataset[T] private[sql] (
 
   /** @inheritdoc */
   def metadataColumn(colName: String): Column = {
-    Column(
-      UnresolvedAttribute(
-        AttributeNameParser.parseAttributeName(colName),
-        getPlanId,
-        isMetadataColumn = true))
+    Column(UnresolvedAttribute(colName, getPlanId, isMetadataColumn = true))
   }
 
   /** @inheritdoc */

connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/internal/columnNodeSupport.scala

@@ -53,8 +53,9 @@ object ColumnNodeToProtoConverter extends (ColumnNode => proto.Expression) {
         builder.setLiteral(toLiteralProtoBuilder(value, dataType))
 
       case u @ UnresolvedAttribute(unparsedIdentifier, planId, isMetadataColumn, _) =>
+        val escapedName = u.sql
         val b = builder.getUnresolvedAttributeBuilder
-          .setUnparsedIdentifier(u.sql)
+          .setUnparsedIdentifier(escapedName)
         if (isMetadataColumn) {
           // We only set this field when it is needed. If we would always set it,
           // too many of the verbatims we use for testing would have to be regenerated.

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

@@ -143,7 +143,7 @@ class Column(val node: ColumnNode) extends Logging {
     name match {
       case "*" => internal.UnresolvedStar(None, planId)
       case _ if name.endsWith(".*") => internal.UnresolvedStar(Option(name), planId)
-      case _ => internal.UnresolvedAttribute(AttributeNameParser.parseAttributeName(name), planId)
+      case _ => internal.UnresolvedAttribute(name, planId)
     }
   })
 

sql/api/src/main/scala/org/apache/spark/sql/internal/columnNodes.scala

@@ -21,6 +21,7 @@ import java.util.concurrent.atomic.AtomicLong
 import ColumnNode._
 
 import org.apache.spark.sql.catalyst.trees.{CurrentOrigin, Origin}
+import org.apache.spark.sql.catalyst.util.AttributeNameParser
 import org.apache.spark.sql.errors.DataTypeErrorsBase
 import org.apache.spark.sql.types.{DataType, IntegerType, LongType, Metadata}
 import org.apache.spark.util.SparkClassUtils
@@ -143,8 +144,15 @@ private[sql] case class UnresolvedAttribute(
 }
 
 private[sql] object UnresolvedAttribute {
-  // For testing
-  def apply(singlePart: String): UnresolvedAttribute = UnresolvedAttribute(singlePart :: Nil)
+  def apply(
+      unparsedIdentifier: String,
+      planId: Option[Long] = None,
+      isMetadataColumn: Boolean = false,
+      origin: Origin = CurrentOrigin.get): UnresolvedAttribute = UnresolvedAttribute(
+    AttributeNameParser.parseAttributeName(unparsedIdentifier),
+    planId = planId,
+    isMetadataColumn = isMetadataColumn,
+    origin = origin)
 }
 
 /**