[SPARK-47210][SQL] Addition of implicit casting without indeterminate support

common/unsafe/src/main/java/org/apache/spark/sql/catalyst/util/CollationFactory.java

@@ -109,6 +109,7 @@ public Collation(
   private static final Collation[] collationTable = new Collation[4];
   private static final HashMap<String, Integer> collationNameToIdMap = new HashMap<>();
 
+  public static final int INDETERMINATE_COLLATION_ID = -1;
   public static final int DEFAULT_COLLATION_ID = 0;
   public static final int LOWERCASE_COLLATION_ID = 1;
 

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

@@ -475,6 +475,24 @@
     ],
     "sqlState" : "42704"
   },
+  "COLLATION_MISMATCH" : {
+    "message" : [
+      "Could not determine which collation to use for string comparison."
+    ],
+    "subClass" : {
+      "EXPLICIT" : {
+        "message" : [
+          "Error occurred due to the mismatch between explicit collations: <explicitTypes>. Decide on a single explicit collation and remove others."
+        ]
+      },
+      "IMPLICIT" : {
+        "message" : [
+          "Error occurred due to the mismatch between multiple implicit non-default collations. Use COLLATE function to set the collation explicitly."
+        ]
+      }
+    },
+    "sqlState" : "42P21"
+  },
   "COLLECTION_SIZE_LIMIT_EXCEEDED" : {
     "message" : [
       "Can't create array with <numberOfElements> elements which exceeding the array size limit <maxRoundedArrayLength>,"
@@ -696,11 +714,6 @@
           "To convert values from <srcType> to <targetType>, you can use the functions <functionNames> instead."
         ]
       },
-      "COLLATION_MISMATCH" : {
-        "message" : [
-          "Collations <collationNameLeft> and <collationNameRight> are not compatible. Please use the same collation for both strings."
-        ]
-      },
       "CREATE_MAP_KEY_DIFF_TYPES" : {
         "message" : [
           "The given keys of function <functionName> should all be the same type, but they are <dataType>."
@@ -1574,6 +1587,12 @@
     ],
     "sqlState" : "22003"
   },
+  "INDETERMINATE_COLLATION" : {
+    "message" : [
+      "Function called requires knowledge of the collation it should apply, but indeterminate collation was found. Use COLLATE function to set the collation explicitly."
+    ],
+    "sqlState" : "42P22"
+  },
   "INDEX_ALREADY_EXISTS" : {
     "message" : [
       "Cannot create the index <indexName> on table <tableName> because it already exists."

docs/sql-error-conditions-collation-mismatch-error-class.md

@@ -0,0 +1,41 @@
+---
+layout: global
+title: COLLATION_MISMATCH error class
+displayTitle: COLLATION_MISMATCH error class
+license: |
+  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.
+---
+
+<!--
+  DO NOT EDIT THIS FILE.
+  It was generated automatically by `org.apache.spark.SparkThrowableSuite`.
+-->
+
+[SQLSTATE: 42P21](sql-error-conditions-sqlstates.html#class-42-syntax-error-or-access-rule-violation)
+
+Could not determine which collation to use for string comparison.
+
+This error class has the following derived error classes:
+
+## EXPLICIT
+
+Error occurred due to the mismatch between explicit collations: `<explicitTypes>`. Decide on a single explicit collation and remove others.
+
+## IMPLICIT
+
+Error occurred due to the mismatch between multiple implicit non-default collations. Use COLLATE function to set the collation explicitly.
+
+

docs/sql-error-conditions-datatype-mismatch-error-class.md

@@ -76,10 +76,6 @@ If you have to cast `<srcType>` to `<targetType>`, you can set `<config>` as `<c
 cannot cast `<srcType>` to `<targetType>`.
 To convert values from `<srcType>` to `<targetType>`, you can use the functions `<functionNames>` instead.
 
-## COLLATION_MISMATCH
-
-Collations `<collationNameLeft>` and `<collationNameRight>` are not compatible. Please use the same collation for both strings.
-
 ## CREATE_MAP_KEY_DIFF_TYPES
 
 The given keys of function `<functionName>` should all be the same type, but they are `<dataType>`.

docs/sql-error-conditions.md

@@ -398,6 +398,14 @@ Cannot find a short name for the codec `<codecName>`.
 
 The value `<collationName>` does not represent a correct collation name. Suggested valid collation name: [`<proposal>`].
 
+### [COLLATION_MISMATCH](sql-error-conditions-collation-mismatch-error-class.html)
+
+[SQLSTATE: 42P21](sql-error-conditions-sqlstates.html#class-42-syntax-error-or-access-rule-violation)
+
+Could not determine which collation to use for string comparison.
+
+For more details see [COLLATION_MISMATCH](sql-error-conditions-collation-mismatch-error-class.html)
+
 ### [COLLECTION_SIZE_LIMIT_EXCEEDED](sql-error-conditions-collection-size-limit-exceeded-error-class.html)
 
 [SQLSTATE: 54000](sql-error-conditions-sqlstates.html#class-54-program-limit-exceeded)
@@ -945,6 +953,12 @@ For more details see [INCONSISTENT_BEHAVIOR_CROSS_VERSION](sql-error-conditions-
 
 Max offset with `<rowsPerSecond>` rowsPerSecond is `<maxSeconds>`, but 'rampUpTimeSeconds' is `<rampUpTimeSeconds>`.
 
+### INDETERMINATE_COLLATION
+
+[SQLSTATE: 42P22](sql-error-conditions-sqlstates.html#class-42-syntax-error-or-access-rule-violation)
+
+Function called requires knowledge of the collation it should apply, but indeterminate collation was found. Use COLLATE function to set the collation explicitly.
+
 ### INDEX_ALREADY_EXISTS
 
 [SQLSTATE: 42710](sql-error-conditions-sqlstates.html#class-42-syntax-error-or-access-rule-violation)

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

@@ -41,14 +41,22 @@ class StringType private(val collationId: Int) extends AtomicType with Serializa
    */
   def isBinaryCollation: Boolean = CollationFactory.fetchCollation(collationId).isBinaryCollation
 
+  /**
+   * Returns whether the collation is indeterminate. An indeterminate collation is
+   * a result of combination of conflicting non-default implicit collations.
+   */
+  def isIndeterminateCollation: Boolean = collationId == CollationFactory.INDETERMINATE_COLLATION_ID
+
   /**
    * Type name that is shown to the customer.
    * If this is an UCS_BASIC collation output is `string` due to backwards compatibility.
    */
   override def typeName: String =
     if (isDefaultCollation) "string"
+    else if (isIndeterminateCollation) s"string COLLATE INDETERMINATE_COLLATION"
     else s"string COLLATE ${CollationFactory.fetchCollation(collationId).collationName}"
 
+
   override def equals(obj: Any): Boolean =
     obj.isInstanceOf[StringType] && obj.asInstanceOf[StringType].collationId == collationId
 

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

@@ -17,6 +17,7 @@
 
 package org.apache.spark.sql.catalyst.analysis
 
+import org.apache.spark.sql.catalyst.analysis.TypeCoercion.{castStringType, hasStringType}
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.catalyst.rules.Rule
@@ -90,6 +91,7 @@ object AnsiTypeCoercion extends TypeCoercionBase {
       Division ::
       IntegralDivision ::
       ImplicitTypeCasts ::
+      CollationTypeCasts ::
       DateTimeOperations ::
       WindowFrameCoercion ::
       GetDateFieldOperations:: Nil) :: Nil
@@ -138,21 +140,31 @@ object AnsiTypeCoercion extends TypeCoercionBase {
   @scala.annotation.tailrec
   private def findWiderTypeForString(dt1: DataType, dt2: DataType): Option[DataType] = {
     (dt1, dt2) match {
-      case (StringType, _: IntegralType) => Some(LongType)
-      case (StringType, _: FractionalType) => Some(DoubleType)
-      case (StringType, NullType) => Some(StringType)
+      case (_: StringType, _: IntegralType) => Some(LongType)
+      case (_: StringType, _: FractionalType) => Some(DoubleType)
+      case (st: StringType, NullType) => Some(st)
       // If a binary operation contains interval type and string, we can't decide which
       // interval type the string should be promoted as. There are many possible interval
       // types, such as year interval, month interval, day interval, hour interval, etc.
-      case (StringType, _: AnsiIntervalType) => None
-      case (StringType, a: AtomicType) => Some(a)
-      case (other, StringType) if other != StringType => findWiderTypeForString(StringType, other)
+      case (_: StringType, _: AnsiIntervalType) => None
+      case (_: StringType, a: AtomicType) => Some(a)
+      case (other, st: StringType) if !other.isInstanceOf[StringType] =>
+        findWiderTypeForString(st, other)
       case _ => None
     }
   }
 
-  override def findWiderCommonType(types: Seq[DataType]): Option[DataType] = {
-    types.foldLeft[Option[DataType]](Some(NullType))((r, c) =>
+  override def findWiderCommonType(exprs: Seq[Expression],
+                                   failOnIndeterminate: Boolean = false): Option[DataType] = {
+    (if (exprs.map(_.dataType).partition(hasStringType)._1.distinct.size > 1) {
+      val collationId = CollationTypeCasts.getOutputCollation(exprs, failOnIndeterminate)
+      exprs.map(e =>
+        if (hasStringType(e.dataType)) {
+          castStringType(e.dataType, collationId)
+          e
+        }
+        else e)
+    } else exprs).map(_.dataType).foldLeft[Option[DataType]](Some(NullType))((r, c) =>
       r match {
         case Some(d) => findWiderTypeForTwo(d, c)
         case _ => None
@@ -173,6 +185,8 @@ object AnsiTypeCoercion extends TypeCoercionBase {
       inType: DataType,
       expectedType: AbstractDataType): Option[DataType] = {
     (inType, expectedType) match {
+      case (_: StringType, st: StringType) =>
+        Some(st)
       // If the expected type equals the input type, no need to cast.
       case _ if expectedType.acceptsType(inType) => Some(inType)
 
@@ -188,21 +202,21 @@ object AnsiTypeCoercion extends TypeCoercionBase {
 
       // This type coercion system will allow implicit converting String type as other
       // primitive types, in case of breaking too many existing Spark SQL queries.
-      case (StringType, a: AtomicType) =>
+      case (_: StringType, a: AtomicType) =>
         Some(a)
 
       // If the target type is any Numeric type, convert the String type as Double type.
-      case (StringType, NumericType) =>
+      case (_: StringType, NumericType) =>
         Some(DoubleType)
 
       // If the target type is any Decimal type, convert the String type as the default
       // Decimal type.
-      case (StringType, DecimalType) =>
+      case (_: StringType, DecimalType) =>
         Some(DecimalType.SYSTEM_DEFAULT)
 
       // If the target type is any timestamp type, convert the String type as the default
       // Timestamp type.
-      case (StringType, AnyTimestampType) =>
+      case (_: StringType, AnyTimestampType) =>
         Some(AnyTimestampType.defaultConcreteType)
 
       case (DateType, AnyTimestampType) =>