[SPARK-51067][SQL] Revert session level collation for DML queries and apply object level collation for DDL queries

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

This PR is a partial revert of the original PR #48962 that introduced the resolution of default session level collation for DDL and DML queries.
The part that is reverted is the default collation resolution for DML queries, whereas the part that is kept is the default collation resolution for DDL queries, which is required to apply the object level collation that was introduced as part of PR #49084. As part of this logic, object level collation is now applied to DDL queries accordingly, with the main logic implemented in ResolveDefaultStringTypes.stringTypeForDDLCommand() method.

### Why are the changes needed?

As there were some unresolved technical issues when attempting to merge the functionality from PR #48962 on Delta side, due to its effect on DML queries, it was decided to pause this functionality for now, thus partially reverting unused parts for maintaining a cleaner code moving forward.
Also, this is inline with customer feedback where object level collation is much more requested functionality, so the focus is to introduce the resolution of object level collation for DDL queries instead, allowing the collation to be specified per table or view on their creation or modification, with propagating the default collation specified to subsequent queries on top of those entities.

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

No

### How was this patch tested?

Existing tests that cover the collations functionality, as well adding new dedicated tests for applying object level collation to the underlying columns.

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

No

Closes #49772 from dejankrak-db/revert-session-collations.

Lead-authored-by: Dejan Krakovic <dejan.krakovic@databricks.com>
Co-authored-by: Stefan Kandic <stefan.kandic@databricks.com>
Signed-off-by: Wenchen Fan <wenchen@databricks.com>
(cherry picked from commit e92e12a)
Signed-off-by: Wenchen Fan <wenchen@databricks.com>

Author: 2 people

common/utils/src/main/resources/error/error-conditions.json

@@ -2308,16 +2308,6 @@
       "The value '<confValue>' in the config \"<confName>\" is invalid."
     ],
     "subClass" : {
-      "DEFAULT_COLLATION" : {
-        "message" : [
-          "Cannot resolve the given default collation. Suggested valid collation names: ['<proposals>']?"
-        ]
-      },
-      "DEFAULT_COLLATION_NOT_SUPPORTED" : {
-        "message" : [
-          "Setting default session collation other than UTF8_BINARY is currently not supported."
-        ]
-      },
       "TIME_ZONE" : {
         "message" : [
           "Cannot resolve the given timezone."

connector/avro/src/main/scala/org/apache/spark/sql/avro/SchemaOfAvro.scala

@@ -23,15 +23,14 @@ import org.apache.spark.sql.catalyst.expressions.{Expression, LeafExpression, Li
 import org.apache.spark.sql.catalyst.expressions.objects.StaticInvoke
 import org.apache.spark.sql.catalyst.util.{FailFastMode, ParseMode, PermissiveMode}
 import org.apache.spark.sql.errors.QueryCompilationErrors
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.{DataType, ObjectType}
+import org.apache.spark.sql.types.{DataType, ObjectType, StringType}
 
 private[sql] case class SchemaOfAvro(
     jsonFormatSchema: String,
     options: Map[String, String])
   extends LeafExpression with RuntimeReplaceable {
 
-  override def dataType: DataType = SQLConf.get.defaultStringType
+  override def dataType: DataType = StringType
 
   override def nullable: Boolean = false
 

sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4

@@ -342,8 +342,7 @@ statement
     ;
 
 setResetStatement
-    : SET COLLATION collationName=identifier                           #setCollation
-    | SET ROLE .*?                                                     #failSetRole
+    : SET ROLE .*?                                                     #failSetRole
     | SET TIME ZONE interval                                           #setTimeZone
     | SET TIME ZONE timezone                                           #setTimeZone
     | SET TIME ZONE .*?                                                #setTimeZone

sql/api/src/main/scala/org/apache/spark/sql/internal/SqlApiConf.scala

@@ -20,7 +20,7 @@ import java.util.TimeZone
 
 import scala.util.Try
 
-import org.apache.spark.sql.types.{AtomicType, StringType, TimestampType}
+import org.apache.spark.sql.types.{AtomicType, TimestampType}
 import org.apache.spark.util.SparkClassUtils
 
 /**
@@ -44,7 +44,6 @@ private[sql] trait SqlApiConf {
   def datetimeJava8ApiEnabled: Boolean
   def sessionLocalTimeZone: String
   def legacyTimeParserPolicy: LegacyBehaviorPolicy.Value
-  def defaultStringType: StringType
   def stackTracesInDataFrameContext: Int
   def dataFrameQueryContextEnabled: Boolean
   def legacyAllowUntypedScalaUDFs: Boolean
@@ -61,7 +60,6 @@ private[sql] object SqlApiConf {
   val LOCAL_RELATION_CACHE_THRESHOLD_KEY: String = {
     SqlApiConfHelper.LOCAL_RELATION_CACHE_THRESHOLD_KEY
   }
-  val DEFAULT_COLLATION: String = SqlApiConfHelper.DEFAULT_COLLATION
 
   def get: SqlApiConf = SqlApiConfHelper.getConfGetter.get()()
 
@@ -87,7 +85,6 @@ private[sql] object DefaultSqlApiConf extends SqlApiConf {
   override def datetimeJava8ApiEnabled: Boolean = false
   override def sessionLocalTimeZone: String = TimeZone.getDefault.getID
   override def legacyTimeParserPolicy: LegacyBehaviorPolicy.Value = LegacyBehaviorPolicy.CORRECTED
-  override def defaultStringType: StringType = StringType
   override def stackTracesInDataFrameContext: Int = 1
   override def dataFrameQueryContextEnabled: Boolean = true
   override def legacyAllowUntypedScalaUDFs: Boolean = false

sql/api/src/main/scala/org/apache/spark/sql/internal/SqlApiConfHelper.scala

@@ -32,7 +32,6 @@ private[sql] object SqlApiConfHelper {
   val CASE_SENSITIVE_KEY: String = "spark.sql.caseSensitive"
   val SESSION_LOCAL_TIMEZONE_KEY: String = "spark.sql.session.timeZone"
   val LOCAL_RELATION_CACHE_THRESHOLD_KEY: String = "spark.sql.session.localRelationCacheThreshold"
-  val DEFAULT_COLLATION: String = "spark.sql.session.collation.default"
   val ARROW_EXECUTION_USE_LARGE_VAR_TYPES = "spark.sql.execution.arrow.useLargeVarTypes"
 
   val confGetter: AtomicReference[() => SqlApiConf] = {

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

@@ -378,7 +378,7 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
       ResolveAliases ::
       ResolveSubquery ::
       ResolveSubqueryColumnAliases ::
-      ResolveDefaultStringTypes ::
+      ResolveDDLCommandStringTypes ::
       ResolveWindowOrder ::
       ResolveWindowFrame ::
       ResolveNaturalAndUsingJoin ::

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

@@ -246,10 +246,6 @@ object CollationTypeCoercion {
     if (!expressions.exists(e => SchemaUtils.hasNonUTF8BinaryCollation(e.dataType))) {
       // if there are no collated types we don't need to do anything
       return None
-    } else if (ResolveDefaultStringTypes.needsResolution(expressions)) {
-      // if any of the strings types are still not resolved
-      // we need to wait for them to be resolved first
-      return None
     }
 
     val collationContextWinner = expressions.foldLeft(findCollationContext(expressions.head)) {

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

@@ -18,87 +18,57 @@
 package org.apache.spark.sql.catalyst.analysis
 
 import org.apache.spark.sql.catalyst.expressions.{Cast, Expression, Literal}
-import org.apache.spark.sql.catalyst.plans.logical.{AddColumns, AlterColumns, AlterColumnSpec, AlterViewAs, ColumnDefinition, CreateView, LogicalPlan, QualifiedColType, ReplaceColumns, V1CreateTablePlan, V2CreateTablePlan}
-import org.apache.spark.sql.catalyst.rules.{Rule, RuleExecutor}
+import org.apache.spark.sql.catalyst.plans.logical.{AddColumns, AlterColumns, AlterColumnSpec, AlterTableCommand, AlterViewAs, ColumnDefinition, CreateTable, CreateView, LogicalPlan, QualifiedColType, ReplaceColumns, V2CreateTablePlan}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.connector.catalog.TableCatalog
 import org.apache.spark.sql.types.{DataType, StringType}
 
 /**
- * Resolves default string types in queries and commands. For queries, the default string type is
- * determined by the session's default string type. For DDL, the default string type is the
- * default type of the object (table -> schema -> catalog). However, this is not implemented yet.
- * So, we will just use UTF8_BINARY for now.
+ * Resolves string types in DDL commands, where the string type inherits the
+ * collation from the corresponding object (table/view -> schema -> catalog).
  */
-object ResolveDefaultStringTypes extends Rule[LogicalPlan] {
+object ResolveDDLCommandStringTypes extends Rule[LogicalPlan] {
   def apply(plan: LogicalPlan): LogicalPlan = {
-    val newPlan = apply0(plan)
-    if (plan.ne(newPlan)) {
-      // Due to how tree transformations work and StringType object being equal to
-      // StringType("UTF8_BINARY"), we need to transform the plan twice
-      // to ensure the correct results for occurrences of default string type.
-      val finalPlan = apply0(newPlan)
-      RuleExecutor.forceAdditionalIteration(finalPlan)
-      finalPlan
-    } else {
-      newPlan
-    }
-  }
-
-  private def apply0(plan: LogicalPlan): LogicalPlan = {
     if (isDDLCommand(plan)) {
       transformDDL(plan)
     } else {
-      transformPlan(plan, sessionDefaultStringType)
+      // For non-DDL commands no need to do any further resolution of string types
+      plan
     }
   }
 
-  /**
-   * Returns whether any of the given `plan` needs to have its
-   * default string type resolved.
-   */
-  def needsResolution(plan: LogicalPlan): Boolean = {
-    if (!isDDLCommand(plan) && isDefaultSessionCollationUsed) {
-      return false
+  /** Default collation used, if object level collation is not provided */
+  private def defaultCollation: String = "UTF8_BINARY"
+
+  /** Returns the string type that should be used in a given DDL command */
+  private def stringTypeForDDLCommand(table: LogicalPlan): StringType = {
+    table match {
+      case createTable: CreateTable if createTable.tableSpec.collation.isDefined =>
+        StringType(createTable.tableSpec.collation.get)
+      case createView: CreateView if createView.collation.isDefined =>
+        StringType(createView.collation.get)
+      case alterTable: AlterTableCommand if alterTable.table.resolved =>
+        val collation = Option(alterTable
+          .table.asInstanceOf[ResolvedTable]
+          .table.properties.get(TableCatalog.PROP_COLLATION))
+        if (collation.isDefined) {
+          StringType(collation.get)
+        } else {
+          StringType(defaultCollation)
+        }
+      case _ => StringType(defaultCollation)
     }
-
-    plan.exists(node => needsResolution(node.expressions))
-  }
-
-  /**
-   * Returns whether any of the given `expressions` needs to have its
-   * default string type resolved.
-   */
-  def needsResolution(expressions: Seq[Expression]): Boolean = {
-    expressions.exists(needsResolution)
   }
 
-  /**
-   * Returns whether the given `expression` needs to have its
-   * default string type resolved.
-   */
-  def needsResolution(expression: Expression): Boolean = {
-    expression.exists(e => transformExpression.isDefinedAt(e))
-  }
-
-  private def isDefaultSessionCollationUsed: Boolean = conf.defaultStringType == StringType
-
-  /**
-   * Returns the default string type that should be used in a given DDL command (for now always
-   * UTF8_BINARY).
-   */
-  private def stringTypeForDDLCommand(table: LogicalPlan): StringType =
-    StringType("UTF8_BINARY")
-
-  /** Returns the session default string type */
-  private def sessionDefaultStringType: StringType =
-    StringType(conf.defaultStringType.collationId)
-
   private def isDDLCommand(plan: LogicalPlan): Boolean = plan exists {
     case _: AddColumns | _: ReplaceColumns | _: AlterColumns => true
     case _ => isCreateOrAlterPlan(plan)
   }
 
   private def isCreateOrAlterPlan(plan: LogicalPlan): Boolean = plan match {
-    case _: V1CreateTablePlan | _: V2CreateTablePlan | _: CreateView | _: AlterViewAs => true
+    // 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 | _: CreateView | _: AlterViewAs => true
     case _ => false
   }
 
@@ -155,22 +125,22 @@ object ResolveDefaultStringTypes extends Rule[LogicalPlan] {
     dataType.existsRecursively(isDefaultStringType)
 
   private def isDefaultStringType(dataType: DataType): Boolean = {
+    // STRING (without explicit collation) is considered default string type.
+    // STRING COLLATE <collation_name> (with explicit collation) is not considered
+    // default string type even when explicit collation is UTF8_BINARY (default collation).
     dataType match {
-      case st: StringType =>
-        // should only return true for StringType object and not StringType("UTF8_BINARY")
-        st.eq(StringType) || st.isInstanceOf[TemporaryStringType]
+      // should only return true for StringType object and not for StringType("UTF8_BINARY")
+      case st: StringType => st.eq(StringType)
       case _ => false
     }
   }
 
   private def replaceDefaultStringType(dataType: DataType, newType: StringType): DataType = {
+    // Should replace STRING with the new type.
+    // Should not replace STRING COLLATE UTF8_BINARY, as that is explicit collation.
     dataType.transformRecursively {
       case currentType: StringType if isDefaultStringType(currentType) =>
-        if (currentType == newType) {
-          TemporaryStringType()
-        } else {
-          newType
-        }
+        newType
     }
   }
 
@@ -186,7 +156,3 @@ object ResolveDefaultStringTypes extends Rule[LogicalPlan] {
     }
   }
 }
-
-case class TemporaryStringType() extends StringType(1) {
-  override def toString: String = s"TemporaryStringType($collationId)"
-}

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

@@ -29,12 +29,8 @@ import org.apache.spark.sql.catalyst.trees.AlwaysProcess
 object ResolveInlineTables extends Rule[LogicalPlan] with EvalHelper {
   override def apply(plan: LogicalPlan): LogicalPlan = {
     plan.resolveOperatorsWithPruning(AlwaysProcess.fn, ruleId) {
-      case table: UnresolvedInlineTable if canResolveTable(table) =>
+      case table: UnresolvedInlineTable if table.expressionsResolved =>
         EvaluateUnresolvedInlineTable.evaluateUnresolvedInlineTable(table)
     }
   }
-
-  private def canResolveTable(table: UnresolvedInlineTable): Boolean = {
-    table.expressionsResolved && !ResolveDefaultStringTypes.needsResolution(table)
-  }
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Expression.scala

@@ -1467,5 +1467,5 @@ case class MultiCommutativeOp(
  * Trait for expressions whose data type should be a default string type.
  */
 trait DefaultStringProducingExpression extends Expression {
-  override def dataType: DataType = SQLConf.get.defaultStringType
+  override def dataType: DataType = StringType
 }