[SPARK-54817][SQL] Refactor Unpivot resolution logic to UnpivotTransformer

### What changes were proposed in this pull request?
In this PR I propose to refactor `Unpivot` resolution logic to `UnpivotTransformer`.

### Why are the changes needed?
In order to reuse it in the single-pass implementation.

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

### How was this patch tested?
Existing tests.

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

Closes #53577 from mihailoale-db/resolveunpivottransformer.

Authored-by: mihailoale-db <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>

Author: mihailoale-db

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -952,32 +952,7 @@ class Analyzer(
       // TypeCoercionBase.UnpivotCoercion determines valueType
       // and casts values once values are set and resolved
       case Unpivot(Some(ids), Some(values), aliases, variableColumnName, valueColumnNames, child) =>
-
-        def toString(values: Seq[NamedExpression]): String =
-          values.map(v => v.name).mkString("_")
-
-        // construct unpivot expressions for Expand
-        val exprs: Seq[Seq[Expression]] =
-          values.zip(aliases.getOrElse(values.map(_ => None))).map {
-            case (vals, Some(alias)) => (ids :+ Literal(alias)) ++ vals
-            case (Seq(value), None) => (ids :+ Literal(value.name)) :+ value
-            // there are more than one value in vals
-            case (vals, None) => (ids :+ Literal(toString(vals))) ++ vals
-          }
-
-        // construct output attributes
-        val variableAttr = AttributeReference(variableColumnName, StringType, nullable = false)()
-        val valueAttrs = valueColumnNames.zipWithIndex.map {
-          case (valueColumnName, idx) =>
-            AttributeReference(
-              valueColumnName,
-              values.head(idx).dataType,
-              values.map(_(idx)).exists(_.nullable))()
-        }
-        val output = (ids.map(_.toAttribute) :+ variableAttr) ++ valueAttrs
-
-        // expand the unpivot expressions
-        Expand(exprs, output, child)
+        UnpivotTransformer(ids, values, aliases, variableColumnName, valueColumnNames, child)
     }
   }
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnpivotTransformer.scala

@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.catalyst.expressions.{
+  AttributeReference,
+  Expression,
+  Literal,
+  NamedExpression
+}
+import org.apache.spark.sql.catalyst.plans.logical.{Expand, LogicalPlan}
+import org.apache.spark.sql.types.StringType
+
+/**
+ * Object used to transform the given [[Unpivot]] node to an [[Expand]] node.
+ */
+object UnpivotTransformer {
+
+  /**
+   * Construct an [[Expand]] node from the given [[Unpivot]] node. Do that by:
+   *  1. Constructing expressions for [[Expand]] out of [[aliases]] and [[values]].
+   *  2. Constructing output attributes.
+   *  3. Creating the [[Expand]] node using the expressions, outputs and the [[Unpivot.child]].
+   */
+  def apply(
+      ids: Seq[NamedExpression],
+      values: Seq[Seq[NamedExpression]],
+      aliases: Option[Seq[Option[String]]],
+      variableColumnName: String,
+      valueColumnNames: Seq[String],
+      child: LogicalPlan): Expand = {
+
+    val expressions: Seq[Seq[Expression]] =
+      values.zip(aliases.getOrElse(values.map(_ => None))).map {
+        case (values, Some(alias)) => (ids :+ Literal(alias)) ++ values
+        case (Seq(value), None) => (ids :+ Literal(value.name)) :+ value
+        case (values, None) =>
+          val stringOfValues = values
+            .map { value =>
+              value.name
+            }
+            .mkString("_")
+          (ids :+ Literal(stringOfValues)) ++ values
+      }
+
+    val variableAttribute =
+      AttributeReference(variableColumnName, StringType, nullable = false)()
+    val valueAttributes = valueColumnNames.zipWithIndex.map {
+      case (valueColumnName, index) =>
+        AttributeReference(
+          valueColumnName,
+          values.head(index).dataType,
+          values.map(_(index)).exists(_.nullable)
+        )()
+    }
+
+    val output = (ids.map(_.toAttribute) :+ variableAttribute) ++ valueAttributes
+
+    Expand(expressions, output, child)
+  }
+}