[SPARK-51673][SQL] Apply default collation to alter view query

### What changes were proposed in this pull request?
Fixed the application of default collation in `ALTER VIEW` queries. For example, if a view has a default collation and we execute `ALTER VIEW v AS SELECT 'a' AS c1`, the default collation was not being applied to the literal `a`.

### Why are the changes needed?
Bug fix.

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

### How was this patch tested?
Tests added to `DefaultCollationTestSuite.scala`

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

Closes #50468 from ilicmarkodb/fix_alter_view.

Authored-by: ilicmarkodb <[email protected]>
Signed-off-by: Max Gekk <[email protected]>

Author: ilicmarkodb

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

@@ -66,6 +66,19 @@ object ResolveDDLCommandStringTypes extends Rule[LogicalPlan] {
             StringType(defaultCollation)
         }
 
+      case alterViewAs: AlterViewAs =>
+        alterViewAs.child match {
+          case resolvedPersistentView: ResolvedPersistentView =>
+            val collation = resolvedPersistentView.metadata.collation.getOrElse(defaultCollation)
+            StringType(collation)
+          case resolvedTempView: ResolvedTempView =>
+            val collation = resolvedTempView.metadata.collation.getOrElse(defaultCollation)
+            StringType(collation)
+          case _ =>
+            // As a safeguard, use the default collation for unknown cases.
+            StringType(defaultCollation)
+        }
+
       // Check if view has default collation
       case _ if AnalysisContext.get.collation.isDefined =>
         StringType(AnalysisContext.get.collation.get)

sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala

@@ -137,7 +137,8 @@ case class CreateViewCommand(
         originalText,
         analyzedPlan,
         aliasedPlan,
-        referredTempFunctions)
+        referredTempFunctions,
+        collation)
       catalog.createTempView(name.table, tableDefinition, overrideIfExists = replace)
     } else if (viewType == GlobalTempView) {
       val db = sparkSession.sessionState.conf.getConf(StaticSQLConf.GLOBAL_TEMP_DATABASE)
@@ -739,8 +740,10 @@ object ViewHelper extends SQLConfHelper with Logging {
       originalText: Option[String],
       analyzedPlan: LogicalPlan,
       aliasedPlan: LogicalPlan,
-      referredTempFunctions: Seq[String]): TemporaryViewRelation = {
-    val uncache = getRawTempView(name.table).map { r =>
+      referredTempFunctions: Seq[String],
+      collation: Option[String] = None): TemporaryViewRelation = {
+    val rawTempView = getRawTempView(name.table)
+    val uncache = rawTempView.map { r =>
       needsToUncache(r, aliasedPlan)
     }.getOrElse(false)
     val storeAnalyzedPlanForView = session.sessionState.conf.storeAnalyzedPlanForView ||
@@ -754,6 +757,16 @@ object ViewHelper extends SQLConfHelper with Logging {
       }
       CommandUtils.uncacheTableOrView(session, name)
     }
+    // When called from CreateViewCommand, this function determines the collation from the
+    // DEFAULT COLLATION clause in the query or assigns None if unspecified.
+    // When called from AlterViewAsCommand, it retrieves the collation from the view's metadata.
+    val defaultCollation = if (collation.isDefined) {
+      collation
+    } else if (rawTempView.isDefined) {
+      rawTempView.get.tableMeta.collation
+    } else {
+      None
+    }
     if (!storeAnalyzedPlanForView) {
       TemporaryViewRelation(
         prepareTemporaryView(
@@ -762,10 +775,11 @@ object ViewHelper extends SQLConfHelper with Logging {
           analyzedPlan,
           aliasedPlan.schema,
           originalText.get,
-          referredTempFunctions))
+          referredTempFunctions,
+          defaultCollation))
     } else {
       TemporaryViewRelation(
-        prepareTemporaryViewStoringAnalyzedPlan(name, aliasedPlan),
+        prepareTemporaryViewStoringAnalyzedPlan(name, aliasedPlan, defaultCollation),
         Some(aliasedPlan))
     }
   }
@@ -795,7 +809,8 @@ object ViewHelper extends SQLConfHelper with Logging {
       analyzedPlan: LogicalPlan,
       viewSchema: StructType,
       originalText: String,
-      tempFunctions: Seq[String]): CatalogTable = {
+      tempFunctions: Seq[String],
+      collation: Option[String]): CatalogTable = {
 
     val tempViews = collectTemporaryViews(analyzedPlan)
     val tempVariables = collectTemporaryVariables(analyzedPlan)
@@ -812,7 +827,8 @@ object ViewHelper extends SQLConfHelper with Logging {
       schema = viewSchema,
       viewText = Some(originalText),
       createVersion = org.apache.spark.SPARK_VERSION,
-      properties = newProperties)
+      properties = newProperties,
+      collation = collation)
   }
 
   /**
@@ -821,12 +837,14 @@ object ViewHelper extends SQLConfHelper with Logging {
    */
   private def prepareTemporaryViewStoringAnalyzedPlan(
       viewName: TableIdentifier,
-      analyzedPlan: LogicalPlan): CatalogTable = {
+      analyzedPlan: LogicalPlan,
+      collation: Option[String]): CatalogTable = {
     CatalogTable(
       identifier = viewName,
       tableType = CatalogTableType.VIEW,
       storage = CatalogStorageFormat.empty,
       schema = analyzedPlan.schema,
-      properties = Map((VIEW_STORING_ANALYZED_PLAN, "true")))
+      properties = Map((VIEW_STORING_ANALYZED_PLAN, "true")),
+      collation = collation)
   }
 }

sql/core/src/test/scala/org/apache/spark/sql/collation/DefaultCollationTestSuite.scala

@@ -475,6 +475,44 @@ class DefaultCollationTestSuiteV1 extends DefaultCollationTestSuite {
       checkAnswer(sql(s"SELECT COUNT(*) FROM $testView WHERE c2 = 'b'"), Seq(Row(0)))
     }
   }
+
+  test("ALTER VIEW check default collation") {
+    Seq("", "TEMPORARY").foreach { temporary =>
+      withView(testView) {
+        sql(s"CREATE $temporary VIEW $testView DEFAULT COLLATION UTF8_LCASE AS SELECT 1")
+        sql(s"ALTER VIEW $testView AS SELECT 'a' AS c1, 'b' AS c2")
+        val prefix = "SYSTEM.BUILTIN"
+        checkAnswer(sql(s"SELECT COLLATION(c1) FROM $testView"),
+          Row(s"$prefix.UTF8_LCASE"))
+        checkAnswer(sql(s"SELECT COLLATION(c2) FROM $testView"),
+          Row(s"$prefix.UTF8_LCASE"))
+        sql(s"ALTER VIEW $testView AS SELECT 'c' AS c3 WHERE 'a' = 'A'")
+        checkAnswer(sql(s"SELECT COLLATION(c3) FROM $testView"),
+          Row(s"$prefix.UTF8_LCASE"))
+      }
+      withTable(testTable) {
+        sql(s"CREATE TABLE $testTable (c1 STRING COLLATE UTF8_LCASE, c2 STRING, c3 INT)")
+        sql(s"INSERT INTO $testTable VALUES ('a', 'b', 1)")
+        withView(testView) {
+          sql(s"CREATE $temporary VIEW $testView DEFAULT COLLATION sr_AI_CI AS SELECT 'a' AS c1")
+          // scalastyle:off
+          sql(
+            s"""ALTER VIEW $testView AS
+              | SELECT *, 'c' AS c4,
+              | (SELECT (SELECT CASE 'š' = 'S' WHEN TRUE THEN 'd' ELSE 'b' END)) AS c5
+              | FROM $testTable
+              | WHERE c1 = 'A' AND 'ć' = 'Č'""".stripMargin)
+          // scalastyle:on
+          val prefix = "SYSTEM.BUILTIN"
+          checkAnswer(sql(s"SELECT COLLATION(c4) FROM $testView"),
+            Row(s"$prefix.sr_CI_AI"))
+          checkAnswer(sql(s"SELECT COLLATION(c5) FROM $testView"),
+            Row(s"$prefix.sr_CI_AI"))
+          checkAnswer(sql(s"SELECT c5 FROM $testView"), Row("d"))
+        }
+      }
+    }
+  }
 }
 
 class DefaultCollationTestSuiteV2 extends DefaultCollationTestSuite with DatasourceV2SQLBase {