[SPARK-51646][SQL] Fix propagating collation in views with default collation

### What changes were proposed in this pull request?
Fixed propagating default collation to literals, subqueries, etc., in `CREATE VIEW ... DEFAULT COLLATION ...` query.
The issue was that the saved string used to construct the view did not include the `DEFAULT COLLATION` ... clause, resulting in the view being created without collation information.

### 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`.

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

Closes apache#50436 from ilicmarkodb/fix_subquery_literals_in_views_with_default_collation.

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/Analyzer.scala

@@ -158,6 +158,7 @@ case class AnalysisContext(
     referredTempVariableNames: Seq[Seq[String]] = Seq.empty,
     outerPlan: Option[LogicalPlan] = None,
     isExecuteImmediate: Boolean = false,
+    collation: Option[String] = None,
 
     /**
      * This is a bridge state between this fixed-point [[Analyzer]] and a single-pass [[Resolver]].
@@ -213,7 +214,8 @@ object AnalysisContext {
       viewDesc.viewReferredTempViewNames,
       mutable.Set(viewDesc.viewReferredTempFunctionNames: _*),
       viewDesc.viewReferredTempVariableNames,
-      isExecuteImmediate = originContext.isExecuteImmediate)
+      isExecuteImmediate = originContext.isExecuteImmediate,
+      collation = viewDesc.collation)
     set(context)
     try f finally { set(originContext) }
   }

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

@@ -17,7 +17,7 @@
 
 package org.apache.spark.sql.catalyst.analysis
 
-import org.apache.spark.sql.catalyst.expressions.{Cast, DefaultStringProducingExpression, Expression, Literal}
+import org.apache.spark.sql.catalyst.expressions.{Cast, DefaultStringProducingExpression, Expression, Literal, SubqueryExpression}
 import org.apache.spark.sql.catalyst.plans.logical.{AddColumns, AlterColumns, AlterColumnSpec, AlterTableCommand, AlterViewAs, ColumnDefinition, CreateTable, CreateView, LogicalPlan, QualifiedColType, ReplaceColumns, ReplaceTable, V2CreateTablePlan}
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.connector.catalog.TableCatalog
@@ -66,6 +66,10 @@ object ResolveDDLCommandStringTypes extends Rule[LogicalPlan] {
             StringType(defaultCollation)
         }
 
+      // Check if view has default collation
+      case _ if AnalysisContext.get.collation.isDefined =>
+        StringType(AnalysisContext.get.collation.get)
+
       case _ => StringType(defaultCollation)
     }
   }
@@ -79,6 +83,8 @@ object ResolveDDLCommandStringTypes extends Rule[LogicalPlan] {
     // For CREATE TABLE, only v2 CREATE TABLE command is supported.
     // Also, table DEFAULT COLLATION cannot be specified through CREATE TABLE AS SELECT command.
     case _: V2CreateTablePlan | _: ReplaceTable | _: CreateView | _: AlterViewAs => true
+    // Check if view has default collation
+    case _ if AnalysisContext.get.collation.isDefined => true
     case _ => false
   }
 
@@ -126,11 +132,22 @@ object ResolveDDLCommandStringTypes extends Rule[LogicalPlan] {
     case columnDef: ColumnDefinition if hasDefaultStringType(columnDef.dataType) =>
       newType => columnDef.copy(dataType = replaceDefaultStringType(columnDef.dataType, newType))
 
-    case cast: Cast if hasDefaultStringType(cast.dataType) =>
+    case cast: Cast if hasDefaultStringType(cast.dataType) &&
+      cast.getTagValue(Cast.USER_SPECIFIED_CAST).isDefined =>
       newType => cast.copy(dataType = replaceDefaultStringType(cast.dataType, newType))
 
     case Literal(value, dt) if hasDefaultStringType(dt) =>
       newType => Literal(value, replaceDefaultStringType(dt, newType))
+
+    case subquery: SubqueryExpression =>
+      val plan = subquery.plan
+      newType =>
+        val newPlan = plan resolveExpressionsUp { expression =>
+          transformExpression
+            .andThen(_.apply(newType))
+            .applyOrElse(expression, identity[Expression])
+        }
+        subquery.withNewPlan(newPlan)
   }
 
   /**

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

@@ -410,6 +410,70 @@ class DefaultCollationTestSuiteV1 extends DefaultCollationTestSuite {
         // scalastyle:on
       }
     }
+    withView(testView) {
+      // scalastyle:off
+      sql(
+        s"""CREATE OR REPLACE VIEW $testView
+          | (c1)
+          | DEFAULT COLLATION sr_ai
+          | AS SELECT 'Ć' as c1 WHERE 'Ć' = 'C'
+          |""".stripMargin)
+      checkAnswer(sql(s"SELECT COUNT(*) FROM $testView WHERE c1 = 'Č'"), Row(1))
+      // scalastyle:on
+    }
+  }
+
+  test("CREATE VIEW with DEFAULT COLLATION") {
+    withView(testView) {
+      sql(
+        s"""CREATE VIEW $testView DEFAULT COLLATION UTF8_LCASE
+          | as SELECT 'a' as c1
+          |""".stripMargin)
+      checkAnswer(sql(s"SELECT COUNT(*) FROM $testView WHERE c1 = 'A'"), Seq(Row(1)))
+    }
+    withTable(testTable) {
+      sql(s"CREATE TABLE $testTable (c1 STRING COLLATE UTF8_LCASE)")
+      sql(s"INSERT INTO $testTable VALUES ('a'), ('A')")
+      withView(testView) {
+        withSQLConf() {
+          // scalastyle:off
+          sql(
+            s"""CREATE VIEW $testView DEFAULT COLLATION SR_AI_CI
+              | AS SELECT c1 FROM $testTable
+              | WHERE 'ć' = 'č'
+              |""".stripMargin)
+          // scalastyle:on
+          checkAnswer(sql(s"SELECT COUNT(*) FROM $testView"), Seq(Row(2)))
+          checkAnswer(sql(s"SELECT COUNT(*) FROM $testView WHERE c1 = 'A'"), Seq(Row(2)))
+        }
+      }
+    }
+    withTable(testTable) {
+      sql(s"CREATE TABLE $testTable (c1 STRING COLLATE UTF8_LCASE)")
+      // scalastyle:off
+      sql(s"INSERT INTO $testTable VALUES ('ć'), ('č')")
+      // scalastyle:on
+      withView(testView) {
+        sql(
+          s"""CREATE VIEW $testView DEFAULT COLLATION UNICODE
+            | AS SELECT CAST(c1 AS STRING COLLATE SR_AI) FROM $testTable
+            |""".stripMargin)
+        val prefix = "SYSTEM.BUILTIN"
+        checkAnswer(sql(s"SELECT DISTINCT COLLATION(c1) FROM $testView"), Row(s"$prefix.sr_AI"))
+        checkAnswer(sql(s"SELECT COUNT(*) FROM $testView WHERE c1 = 'c'"), Row(2))
+      }
+    }
+    withView(testView) {
+      sql(
+        s"""CREATE VIEW $testView DEFAULT COLLATION UTF8_LCASE
+          | AS SELECT 'a' AS c1,
+          | (SELECT (SELECT CASE 'a' = 'A' WHEN TRUE THEN 'a' ELSE 'b' END)
+          |  WHERE (SELECT 'b' WHERE 'c' = 'C') = 'B') AS c2
+          |""".stripMargin)
+      checkAnswer(sql(s"SELECT COUNT(*) FROM $testView WHERE c1 = 'A'"), Seq(Row(1)))
+      checkAnswer(sql(s"SELECT COUNT(*) FROM $testView WHERE c2 = 'a'"), Seq(Row(1)))
+      checkAnswer(sql(s"SELECT COUNT(*) FROM $testView WHERE c2 = 'b'"), Seq(Row(0)))
+    }
   }
 }