[SQL] Fix session level collation

sql/api/src/main/scala/org/apache/spark/sql/catalyst/parser/DataTypeAstBuilder.scala

@@ -28,7 +28,7 @@ import org.apache.spark.sql.catalyst.util.CollationFactory
 import org.apache.spark.sql.catalyst.util.SparkParserUtils.{string, withOrigin}
 import org.apache.spark.sql.errors.QueryParsingErrors
 import org.apache.spark.sql.internal.SqlApiConf
-import org.apache.spark.sql.types.{ArrayType, BinaryType, BooleanType, ByteType, CalendarIntervalType, CharType, DataType, DateType, DayTimeIntervalType, DecimalType, DoubleType, FloatType, IntegerType, LongType, MapType, MetadataBuilder, NullType, ShortType, StringType, StructField, StructType, TimestampNTZType, TimestampType, VarcharType, VariantType, YearMonthIntervalType}
+import org.apache.spark.sql.types.{ArrayType, BinaryType, BooleanType, ByteType, CalendarIntervalType, CharType, DataType, DateType, DayTimeIntervalType, DecimalType, DoubleType, FloatType, ImplicitStringType, IntegerType, LongType, MapType, MetadataBuilder, NullType, ShortType, StringType, StructField, StructType, TimestampNTZType, TimestampType, VarcharType, VariantType, YearMonthIntervalType}
 
 class DataTypeAstBuilder extends SqlBaseParserBaseVisitor[AnyRef] {
   protected def typedVisit[T](ctx: ParseTree): T = {
@@ -74,7 +74,7 @@ class DataTypeAstBuilder extends SqlBaseParserBaseVisitor[AnyRef] {
       case (TIMESTAMP_LTZ, Nil) => TimestampType
       case (STRING, Nil) =>
         typeCtx.children.asScala.toSeq match {
-          case Seq(_) => SqlApiConf.get.defaultStringType
+          case Seq(_) => ImplicitStringType
           case Seq(_, ctx: CollateClauseContext) =>
             val collationName = visitCollateClause(ctx)
             val collationId = CollationFactory.collationNameToId(collationName)

sql/api/src/main/scala/org/apache/spark/sql/types/StringType.scala

@@ -19,7 +19,7 @@ package org.apache.spark.sql.types
 
 import org.json4s.JsonAST.{JString, JValue}
 
-import org.apache.spark.annotation.Stable
+import org.apache.spark.annotation.{Evolving, Stable}
 import org.apache.spark.sql.catalyst.util.CollationFactory
 
 /**
@@ -30,7 +30,7 @@ import org.apache.spark.sql.catalyst.util.CollationFactory
  *   The id of collation for this StringType.
  */
 @Stable
-class StringType private (val collationId: Int) extends AtomicType with Serializable {
+class StringType private[sql] (val collationId: Int) extends AtomicType with Serializable {
 
   /**
    * Support for Binary Equality implies that strings are considered equal only if they are byte
@@ -106,3 +106,9 @@ case object StringType extends StringType(0) {
     new StringType(collationId)
   }
 }
+
+/**
+ * A.
+ */
+@Evolving
+object ImplicitStringType extends StringType(-2) {}

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

@@ -314,12 +314,14 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
       ResolveProcedures ::
       BindProcedures ::
       ResolveTableSpec ::
+      ResolveImplicitStringTypes ::
       ResolveAliases ::
       ResolveSubquery ::
       ResolveSubqueryColumnAliases ::
       ResolveWindowOrder ::
       ResolveWindowFrame ::
       ResolveNaturalAndUsingJoin ::
+//      ResolveImplicitStringTypes ::
       ResolveOutputRelation ::
       new ResolveDataFrameDropColumns(catalogManager) ::
       new ResolveSetVariable(catalogManager) ::
@@ -339,6 +341,7 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
       ResolveRowLevelCommandAssignments ::
       MoveParameterizedQueriesDown ::
       BindParameters ::
+//      ResolveImplicitStringTypes ::
       typeCoercionRules() ++
       Seq(
         ResolveWithCTE,

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

@@ -457,6 +457,9 @@ trait CheckAnalysis extends PredicateHelper with LookupCatalog with QueryErrorsB
               errorClass = "UNBOUND_SQL_PARAMETER",
               messageParameters = Map("name" -> p.name))
 
+          case other if other.dataType == ImplicitStringType =>
+            throw new RuntimeException("Found implicit string type in: " + other.toJSON)
+
           case _ =>
         })
 

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

@@ -0,0 +1,94 @@
+/*
+ * 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.{Cast, Literal}
+import org.apache.spark.sql.catalyst.plans.logical.{AddColumns, AlterColumn, AlterViewAs, AlterViewSchemaBinding, ColumnDefinition, CreateFunction, CreateView, LogicalPlan, QualifiedColType, ReplaceColumns, V2CreateTablePlan}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{DataType, ImplicitStringType, StringType}
+
+object ResolveImplicitStringTypes extends Rule[LogicalPlan] {
+  def apply(plan: LogicalPlan): LogicalPlan = {
+    plan match {
+
+      // Implicit string type should be resolved to the collation of the object for DDL commands.
+      // However, this is not implemented yet. So, we will just use UTF8_BINARY for now.
+      case _: V2CreateTablePlan |
+           _: CreateView | _: AlterViewAs | _: AlterViewSchemaBinding | _: AlterViewSchemaBinding |
+           _: CreateFunction =>
+        val res = replaceWith(plan, StringType)
+        res
+
+      case addCols: AddColumns if hasImplicitStringType(addCols.columnsToAdd) =>
+        addCols.copy(columnsToAdd = replaceColTypes(addCols.columnsToAdd, StringType))
+
+      case replaceCols: ReplaceColumns if hasImplicitStringType(replaceCols.columnsToAdd) =>
+        replaceCols.copy(columnsToAdd = replaceColTypes(replaceCols.columnsToAdd, StringType))
+
+      case a: AlterColumn if a.dataType.isDefined && isImplicitStringType(a.dataType.get) =>
+        a.copy(dataType = Some(StringType))
+
+      // Implicit string type should be resolved to the session collation for DML commands.
+      case _ if SQLConf.get.defaultStringType != StringType =>
+        val res = replaceWith(plan, SQLConf.get.defaultStringType)
+        res
+
+      case _ =>
+        replaceWith(plan, StringType)
+    }
+  }
+
+  private def replaceWith(plan: LogicalPlan, newType: StringType): LogicalPlan = {
+    plan resolveOperators {
+      case operator =>
+        operator transformExpressions {
+          case columnDef: ColumnDefinition if isImplicitStringType(columnDef.dataType) =>
+            columnDef.copy(dataType = newType)
+
+          case cast: Cast if isImplicitStringType(cast.dataType) =>
+            cast.copy(dataType = newType)
+
+          case Literal(value, ImplicitStringType) =>
+            Literal(value, newType)
+        }
+    }
+  }
+
+  private def replaceColTypes(
+      colTypes: Seq[QualifiedColType],
+      newType: StringType): Seq[QualifiedColType] = {
+    colTypes.map {
+      case col if isImplicitStringType(col.dataType) =>
+        col.copy(dataType = newType)
+      case col => col
+    }
+  }
+
+  private def hasImplicitStringType(colTypes: Seq[QualifiedColType]): Boolean = {
+    colTypes.exists(col => isImplicitStringType(col.dataType))
+  }
+
+  private def isImplicitStringType(dataType: DataType): Boolean = {
+    dataType match {
+      case ImplicitStringType => true
+      case _ => false
+    }
+  }
+}

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

@@ -564,6 +564,11 @@ object TableOutputResolver extends SQLConfHelper with Logging {
       conf: SQLConf,
       colName: String): Expression = {
 
+    // this can avoid unneeded casts
+//    if (expr.dataType == expectedType) {
+//      return expr
+//    }
+
     conf.storeAssignmentPolicy match {
       case StoreAssignmentPolicy.ANSI =>
         val cast = Cast(expr, expectedType, Option(conf.sessionLocalTimeZone), ansiEnabled = true)

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala

@@ -2048,7 +2048,7 @@ class AstBuilder extends DataTypeAstBuilder
     }
 
     val unresolvedTable = UnresolvedInlineTable(aliases, rows.toSeq)
-    val table = if (conf.getConf(SQLConf.EAGER_EVAL_OF_UNRESOLVED_INLINE_TABLE_ENABLED)) {
+    val table = if (false && conf.getConf(SQLConf.EAGER_EVAL_OF_UNRESOLVED_INLINE_TABLE_ENABLED)) {
       EvaluateUnresolvedInlineTable.evaluate(unresolvedTable)
     } else {
       unresolvedTable
@@ -3266,7 +3266,7 @@ class AstBuilder extends DataTypeAstBuilder
    * Create a String literal expression.
    */
   override def visitStringLiteral(ctx: StringLiteralContext): Literal = withOrigin(ctx) {
-    Literal.create(createString(ctx), conf.defaultStringType)
+    Literal.create(createString(ctx), ImplicitStringType)
   }
 
   /**

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleExecutor.scala

@@ -233,16 +233,21 @@ abstract class RuleExecutor[TreeType <: TreeNode[_]] extends Logging {
               planChangeLogger.logRule(rule.ruleName, plan, result)
               // Run the plan changes validation after each rule.
               if (Utils.isTesting || enableValidation) {
-                validatePlanChanges(plan, result) match {
-                  case Some(msg) =>
-                    throw new SparkException(
-                      errorClass = "PLAN_VALIDATION_FAILED_RULE_IN_BATCH",
-                      messageParameters = Map(
-                        "rule" -> rule.ruleName,
-                        "batch" -> batch.name,
-                        "reason" -> msg),
-                      cause = null)
-                  case _ =>
+                try {
+                  validatePlanChanges(plan, result) match {
+                    case Some(msg) =>
+                      throw new SparkException(
+                        errorClass = "PLAN_VALIDATION_FAILED_RULE_IN_BATCH",
+                        messageParameters = Map(
+                          "rule" -> rule.ruleName,
+                          "batch" -> batch.name,
+                          "reason" -> msg),
+                        cause = null)
+                    case _ =>
+                  }
+                } catch {
+                  case  a: Throwable =>
+                    throw new RuntimeException("asdf")
                 }
               }
             }