[SPARK-49022][CONNECT][SQL][FOLLOW-UP] Parse unresolved identifier to…

… keep the behavior same

### What changes were proposed in this pull request?

This PR is a followup of apache#47688 that keeps `Column.toString` as the same before.

### Why are the changes needed?

To keep the same behaviour with Spark Classic and Connect.

### Does this PR introduce _any_ user-facing change?

No, the main change has not been released out yet.

### How was this patch tested?

Will be added separately. I manually tested:

```scala
import org.apache.spark.sql.functions.col
val name = "with`!#$%dot".replace("`", "``")
col(s"`${name}`").toString.equals("with`!#$%dot")
```

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes apache#48376 from HyukjinKwon/SPARK-49022-followup.

Authored-by: Hyukjin Kwon <gurwls223@apache.org>
Signed-off-by: Hyukjin Kwon <gurwls223@apache.org>

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

@@ -52,9 +52,10 @@ object ColumnNodeToProtoConverter extends (ColumnNode => proto.Expression) {
       case Literal(value, Some(dataType), _) =>
         builder.setLiteral(toLiteralProtoBuilder(value, dataType))
 
-      case UnresolvedAttribute(unparsedIdentifier, planId, isMetadataColumn, _) =>
+      case u @ UnresolvedAttribute(unparsedIdentifier, planId, isMetadataColumn, _) =>
+        val escapedName = u.sql
         val b = builder.getUnresolvedAttributeBuilder
-          .setUnparsedIdentifier(unparsedIdentifier)
+          .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.

connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/ColumnTestSuite.scala

@@ -173,8 +173,8 @@ class ColumnTestSuite extends ConnectFunSuite {
     assert(explain1 != explain2)
     assert(explain1.strip() == "+(a, b)")
     assert(explain2.contains("UnresolvedFunction(+"))
-    assert(explain2.contains("UnresolvedAttribute(a"))
-    assert(explain2.contains("UnresolvedAttribute(b"))
+    assert(explain2.contains("UnresolvedAttribute(List(a"))
+    assert(explain2.contains("UnresolvedAttribute(List(b"))
   }
 
   private def testColName(dataType: DataType, f: ColumnName => StructField): Unit = {

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
@@ -122,22 +123,48 @@ private[sql] case class Literal(
 /**
  * Reference to an attribute produced by one of the underlying DataFrames.
  *
- * @param unparsedIdentifier
+ * @param nameParts
  *   name of the attribute.
  * @param planId
  *   id of the plan (Dataframe) that produces the attribute.
  * @param isMetadataColumn
  *   whether this is a metadata column.
  */
 private[sql] case class UnresolvedAttribute(
-    unparsedIdentifier: String,
+    nameParts: Seq[String],
     planId: Option[Long] = None,
     isMetadataColumn: Boolean = false,
     override val origin: Origin = CurrentOrigin.get)
     extends ColumnNode {
+
   override private[internal] def normalize(): UnresolvedAttribute =
     copy(planId = None, origin = NO_ORIGIN)
-  override def sql: String = unparsedIdentifier
+
+  override def sql: String = nameParts.map(n => if (n.contains(".")) s"`$n`" else n).mkString(".")
+}
+
+private[sql] object UnresolvedAttribute {
+  def apply(
+      unparsedIdentifier: String,
+      planId: Option[Long],
+      isMetadataColumn: Boolean,
+      origin: Origin): UnresolvedAttribute = UnresolvedAttribute(
+    AttributeNameParser.parseAttributeName(unparsedIdentifier),
+    planId = planId,
+    isMetadataColumn = isMetadataColumn,
+    origin = origin)
+
+  def apply(
+      unparsedIdentifier: String,
+      planId: Option[Long],
+      isMetadataColumn: Boolean): UnresolvedAttribute =
+    apply(unparsedIdentifier, planId, isMetadataColumn, CurrentOrigin.get)
+
+  def apply(unparsedIdentifier: String, planId: Option[Long]): UnresolvedAttribute =
+    apply(unparsedIdentifier, planId, false, CurrentOrigin.get)
+
+  def apply(unparsedIdentifier: String): UnresolvedAttribute =
+    apply(unparsedIdentifier, None, false, CurrentOrigin.get)
 }
 
 /**

sql/core/src/main/scala/org/apache/spark/sql/internal/columnNodeSupport.scala

@@ -54,8 +54,8 @@ private[sql] trait ColumnNodeToExpressionConverter extends (ColumnNode => Expres
         case Literal(value, None, _) =>
           expressions.Literal(value)
 
-        case UnresolvedAttribute(unparsedIdentifier, planId, isMetadataColumn, _) =>
-          convertUnresolvedAttribute(unparsedIdentifier, planId, isMetadataColumn)
+        case UnresolvedAttribute(nameParts, planId, isMetadataColumn, _) =>
+          convertUnresolvedAttribute(nameParts, planId, isMetadataColumn)
 
         case UnresolvedStar(unparsedTarget, None, _) =>
           val target = unparsedTarget.map { t =>
@@ -74,7 +74,7 @@ private[sql] trait ColumnNodeToExpressionConverter extends (ColumnNode => Expres
           analysis.UnresolvedRegex(columnNameRegex, Some(nameParts), conf.caseSensitiveAnalysis)
 
         case UnresolvedRegex(unparsedIdentifier, planId, _) =>
-          convertUnresolvedAttribute(unparsedIdentifier, planId, isMetadataColumn = false)
+          convertUnresolvedRegex(unparsedIdentifier, planId)
 
         case UnresolvedFunction(functionName, arguments, isDistinct, isUDF, isInternal, _) =>
           val nameParts = if (isUDF) {
@@ -223,10 +223,10 @@ private[sql] trait ColumnNodeToExpressionConverter extends (ColumnNode => Expres
   }
 
   private def convertUnresolvedAttribute(
-      unparsedIdentifier: String,
+      nameParts: Seq[String],
       planId: Option[Long],
       isMetadataColumn: Boolean): analysis.UnresolvedAttribute = {
-    val attribute = analysis.UnresolvedAttribute.quotedString(unparsedIdentifier)
+    val attribute = analysis.UnresolvedAttribute(nameParts)
     if (planId.isDefined) {
       attribute.setTagValue(LogicalPlan.PLAN_ID_TAG, planId.get)
     }
@@ -235,6 +235,16 @@ private[sql] trait ColumnNodeToExpressionConverter extends (ColumnNode => Expres
     }
     attribute
   }
+
+  private def convertUnresolvedRegex(
+      unparsedIdentifier: String,
+      planId: Option[Long]): analysis.UnresolvedAttribute = {
+    val attribute = analysis.UnresolvedAttribute.quotedString(unparsedIdentifier)
+    if (planId.isDefined) {
+      attribute.setTagValue(LogicalPlan.PLAN_ID_TAG, planId.get)
+    }
+    attribute
+  }
 }
 
 private[sql] object ColumnNodeToExpressionConverter extends ColumnNodeToExpressionConverter {