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

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

@@ -105,6 +105,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 \"<left>\" and \"<right>\""
+        ]
+      },
+      "IMPLICIT" : {
+        "message" : [
+          "Error occurred due to the mismatch between multiple implicit 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 "`<left>`" and "`<right>`"
+
+## IMPLICIT
+
+Error occurred due to the mismatch between multiple implicit collations. Use COLLATE function to set the collation explicitly.
+
+

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,6 +41,12 @@ 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.

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

@@ -90,6 +90,7 @@ object AnsiTypeCoercion extends TypeCoercionBase {
       Division ::
       IntegralDivision ::
       ImplicitTypeCasts ::
+      CollationTypeCasts ::
       DateTimeOperations ::
       WindowFrameCoercion ::
       GetDateFieldOperations:: Nil) :: Nil
@@ -138,15 +139,16 @@ 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
     }
   }
@@ -186,23 +188,26 @@ object AnsiTypeCoercion extends TypeCoercionBase {
       case (NullType, target) if !target.isInstanceOf[TypeCollection] =>
         Some(target.defaultConcreteType)
 
+      case (_: StringType, st: StringType) =>
+        Some(st)
+
       // 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) =>