[SPARK-57054][SQL] Make view collation sticky across ALTER SCHEMA DEFAULT COLLATION

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

@@ -26,7 +26,7 @@ import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.catalyst.trees.CurrentOrigin
 import org.apache.spark.sql.catalyst.types.DataTypeUtils.{areSameBaseType, isDefaultStringCharOrVarcharType, replaceDefaultStringCharAndVarcharTypes}
 import org.apache.spark.sql.catalyst.util.CharVarcharUtils
-import org.apache.spark.sql.connector.catalog.{CatalogV2Util, SupportsNamespaces, TableCatalog, V1ViewInfo, ViewInfo}
+import org.apache.spark.sql.connector.catalog.{SupportsNamespaces, TableCatalog}
 import org.apache.spark.sql.types.{DataType, StringHelper, StringType}
 
 /**
@@ -206,46 +206,6 @@ object ApplyDefaultCollation extends Rule[LogicalPlan] {
           newCreateView.copyTagsFrom(createView)
           newCreateView
 
-        // We match against ResolvedPersistentView because temporary views don't have a
-        // schema/catalog. The rewrite covers both v1 (session-catalog, [[V1ViewInfo]]) and
-        // non-session v2 views: when the existing view has no `PROP_COLLATION` and the
-        // namespace supplies a default, fold that default into the resolved `ViewInfo`. For
-        // v1, `V1ViewInfo` is rebuilt around a `CatalogTable` whose typed `collation` field
-        // holds the new value; `V1ViewInfo.builderFrom` bridges that into the v2
-        // `properties()` bag, so downstream consumers (`fetchDefaultCollation`,
-        // `AlterV2ViewExec`'s `existingProp(PROP_COLLATION)`) see it under either surface.
-        // For v2, we rebuild the existing `ViewInfo` with `PROP_COLLATION` set so the same
-        // downstream consumers see it on the regular `info.properties` path.
-        case alterViewAs @ AlterViewAs(rpv @ ResolvedPersistentView(
-        catalog: SupportsNamespaces, identifier, info), _, _, _, _)
-            if Option(info.properties.get(TableCatalog.PROP_COLLATION)).isEmpty =>
-          // Only rewrite when the namespace actually supplies a default. [[ViewInfo]] /
-          // [[V1ViewInfo]] are non-case classes, so a copy with structurally-identical fields
-          // still reads as a different reference -- if we rewrote unconditionally, the
-          // resolution batch would see the plan change every iteration and never reach
-          // fixed point. Looking up the namespace default takes one `loadNamespaceMetadata`
-          // round trip, so do it once here and bail out before rewriting if the namespace has
-          // no default.
-          getCollationFromSchemaMetadata(catalog, identifier.namespace()) match {
-            case Some(newCollation) =>
-              val newInfo: ViewInfo = info match {
-                case v1Info: V1ViewInfo =>
-                  new V1ViewInfo(v1Info.v1Table.copy(collation = Some(newCollation)))
-                case _ =>
-                  CatalogV2Util.viewInfoBuilderFrom(info)
-                    .withCollation(newCollation)
-                    .build()
-              }
-              val newRpv = rpv.copy(info = newInfo)
-              val newAlterViewAs = CurrentOrigin.withOrigin(alterViewAs.origin) {
-                alterViewAs.copy(child = newRpv)
-              }
-              newAlterViewAs.copyTagsFrom(alterViewAs)
-              newAlterViewAs
-            case None =>
-              alterViewAs
-          }
-
         case createUserDefinedFunction@CreateUserDefinedFunction(
         ResolvedIdentifier(catalog: SupportsNamespaces, identifier),
         _, _, _, _, _, collation, _, _, _, _, _, _) if collation.isEmpty =>

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

@@ -1556,14 +1556,8 @@ abstract class DefaultCollationTestSuiteV1 extends DefaultCollationTestSuite {
     }
   }
 
-  testString("ALTER VIEW AS picks up the namespace's default collation when the existing " +
-      "view has none") {
+  testString("ALTER SCHEMA DEFAULT COLLATION does not retroactively change a view's collation") {
     _ =>
-    // The view is created in a schema with no default collation, so the stored
-    // `CatalogTable.collation` is `None`. After the schema gains a default, the next
-    // `ALTER VIEW AS` must fold that default into both the analyzed plan's literal types
-    // (so `assertTableColumnCollation` sees it on read) and the persisted CatalogTable so
-    // the AnalysisContext fallback fires on every subsequent read.
     withDatabase(testSchema) {
       sql(s"CREATE SCHEMA $testSchema")
       sql(s"USE $testSchema")
@@ -1573,8 +1567,8 @@ abstract class DefaultCollationTestSuiteV1 extends DefaultCollationTestSuite {
 
         sql(s"ALTER SCHEMA $testSchema DEFAULT COLLATION UTF8_LCASE")
         sql(s"ALTER VIEW $testView AS SELECT 'x' AS c1, 'y' AS c2")
-        assertTableColumnCollation(testView, "c1", "UTF8_LCASE")
-        assertTableColumnCollation(testView, "c2", "UTF8_LCASE")
+        assertTableColumnCollation(testView, "c1", "UTF8_BINARY")
+        assertTableColumnCollation(testView, "c2", "UTF8_BINARY")
       }
     }
   }

sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AlterViewAsSuite.scala

@@ -25,12 +25,11 @@ import org.apache.spark.sql.types.StringType
 
 class AlterViewAsSuite extends command.AlterViewAsSuiteBase with ViewCommandSuiteBase {
 
-  test("V2: ALTER VIEW AS picks up the namespace's default collation when the existing view " +
-      "has none") {
-    // Create the namespace with no default collation; create a view in it (PROP_COLLATION
-    // unset). Then set the namespace default and ALTER VIEW AS -- the new ViewInfo must end
-    // up with PROP_COLLATION = UTF8_LCASE (so v1Table.toCatalogTable's `collation` field is
-    // set, and view-read time picks up UTF8_LCASE via AnalysisContext.collation).
+  test("V2: ALTER NAMESPACE DEFAULT COLLATION does not retroactively change a view's collation") {
+    // A view created in a namespace without a default collation keeps its creation-time
+    // collation behavior. A later `ALTER NAMESPACE ... DEFAULT COLLATION` followed by
+    // `ALTER VIEW AS` does not fold the new namespace default into the view -- `PROP_COLLATION`
+    // stays at its create-time value (empty here) and the body literals stay UTF8_BINARY.
     withSQLConf(SQLConf.SCHEMA_LEVEL_COLLATIONS_ENABLED.key -> "true") {
       val viewName = "v2_alter_collation_inherit"
       val view = s"$catalog.$namespace.$viewName"
@@ -45,11 +44,10 @@ class AlterViewAsSuite extends command.AlterViewAsSuiteBase with ViewCommandSuit
       sql(s"ALTER VIEW $view AS SELECT 'x' AS c1, 'y' AS c2")
 
       val stored = viewCatalog.getStoredView(Array(namespace), viewName)
-      assert(stored.properties().get(TableCatalog.PROP_COLLATION) == "UTF8_LCASE")
-      // Read-time the view body's literal types reflect the inherited collation.
+      assert(Option(stored.properties().get(TableCatalog.PROP_COLLATION)).isEmpty)
       val df = spark.table(view)
-      assert(df.schema("c1").dataType === StringType("UTF8_LCASE"))
-      assert(df.schema("c2").dataType === StringType("UTF8_LCASE"))
+      assert(df.schema("c1").dataType === StringType)
+      assert(df.schema("c2").dataType === StringType)
     }
   }