[SPARK-50175][SQL] Change collation precedence calculation

### What changes were proposed in this pull request?
Changing the way how the collation strength of string expressions are calculated. Currently, there are three different collation strengths:
 - explicit - result of the `collate` expression
 - implicit - column references and output of string functions
 - default - literals and cast expression

However, unlike in other database systems (pg, sql server) collation strengths were not propagated up the expression tree, meaning that `substring('a' collate unicode), 0, 1)` would have implicit strength because it is the result of a string expression.

My proposal is to change the behavior to be more in line with other systems mentioned above; and to do it by traversing the expression tree and propagating the highest precedence strengths up (explicit being the highest and default the lowest) while also finding the conflicts between them (conflicting explicit or implicit strenghts).

### Why are the changes needed?
To be more consistent with other systems that have collations (postgre, sql server etc.)

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

### How was this patch tested?
New unit tests.

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

Closes apache#48663 from stefankandic/newCollationPrec-separate.

Authored-by: Stefan Kandic <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>

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

@@ -565,11 +565,14 @@ case class StringToMap(text: Expression, pairDelim: Expression, keyValueDelim: E
   extends TernaryExpression with ExpectsInputTypes {
   override def nullIntolerant: Boolean = true
   def this(child: Expression, pairDelim: Expression) = {
-    this(child, pairDelim, Literal(":"))
+    this(child, pairDelim, Literal.create(":", SQLConf.get.defaultStringType))
   }
 
   def this(child: Expression) = {
-    this(child, Literal(","), Literal(":"))
+    this(
+      child,
+      Literal.create(",", SQLConf.get.defaultStringType),
+      Literal.create(":", SQLConf.get.defaultStringType))
   }
 
   override def stateful: Boolean = true

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

@@ -93,6 +93,7 @@ object ExtractValue {
 trait ExtractValue extends Expression {
   override def nullIntolerant: Boolean = true
   final override val nodePatterns: Seq[TreePattern] = Seq(EXTRACT_VALUE)
+  val child: Expression
 }
 
 /**

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

@@ -1862,7 +1862,8 @@ trait PadExpressionBuilderBase extends ExpressionBuilder {
       if (expressions(0).dataType == BinaryType && behaviorChangeEnabled) {
         BinaryPad(funcName, expressions(0), expressions(1), Literal(Array[Byte](0)))
       } else {
-        createStringPad(expressions(0), expressions(1), Literal(" "))
+        createStringPad(expressions(0),
+          expressions(1), Literal.create(" ", SQLConf.get.defaultStringType))
       }
     } else if (numArgs == 3) {
       if (expressions(0).dataType == BinaryType && expressions(2).dataType == BinaryType
@@ -1992,7 +1993,10 @@ object RPadExpressionBuilder extends PadExpressionBuilderBase {
   }
 }
 
-case class StringRPad(str: Expression, len: Expression, pad: Expression = Literal(" "))
+case class StringRPad(
+    str: Expression,
+    len: Expression,
+    pad: Expression = Literal.create(" ", SQLConf.get.defaultStringType))
   extends TernaryExpression with ImplicitCastInputTypes {
   override def nullIntolerant: Boolean = true
   override def first: Expression = str

sql/core/src/test/scala/org/apache/spark/sql/CollationExpressionWalkerSuite.scala

@@ -742,7 +742,7 @@ class CollationExpressionWalkerSuite extends SparkFunSuite with SharedSparkSessi
           }
         } catch {
           case e: SparkRuntimeException => assert(e.getCondition == "USER_RAISED_EXCEPTION")
-          case other: Throwable => throw other
+          case other: Throwable => throw new Exception(s"Query $query failed", other)
         }
       }
     }

sql/core/src/test/scala/org/apache/spark/sql/CollationSQLExpressionsSuite.scala

@@ -2416,8 +2416,15 @@ class CollationSQLExpressionsSuite
            |collate('${testCase.left}', '${testCase.leftCollation}'))=
            |collate('${testCase.right}', '${testCase.rightCollation}');
            |""".stripMargin
-      val testQuery = sql(query)
-      checkAnswer(testQuery, Row(testCase.result))
+
+      if (testCase.leftCollation == testCase.rightCollation) {
+        checkAnswer(sql(query), Row(testCase.result))
+      } else {
+        val exception = intercept[AnalysisException] {
+          sql(query)
+        }
+        assert(exception.getCondition === "COLLATION_MISMATCH.EXPLICIT")
+      }
     })
 
     val queryPass =

sql/core/src/test/scala/org/apache/spark/sql/CollationStringExpressionsSuite.scala

@@ -180,11 +180,12 @@ class CollationStringExpressionsSuite
 
     // Because `StringSplitSQL` is an internal expression,
     // E2E SQL test cannot be performed in `collations.sql`.
+
     checkError(
       exception = intercept[AnalysisException] {
         val expr = StringSplitSQL(
-          Cast(Literal.create("1a2"), StringType("UTF8_BINARY")),
-          Cast(Literal.create("a"), StringType("UTF8_LCASE")))
+          Literal.create("1a2", StringType("UTF8_BINARY")),
+          Literal.create("a", StringType("UTF8_LCASE")))
         CollationTypeCasts.transform(expr)
       },
       condition = "COLLATION_MISMATCH.IMPLICIT",
@@ -193,6 +194,7 @@ class CollationStringExpressionsSuite
         "implicitTypes" -> """"STRING", "STRING COLLATE UTF8_LCASE""""
       )
     )
+
     checkError(
       exception = intercept[AnalysisException] {
         val expr = StringSplitSQL(

sql/core/src/test/scala/org/apache/spark/sql/CollationSuite.scala

@@ -624,17 +624,11 @@ class CollationSuite extends DatasourceV2SQLBase with AdaptiveSparkPlanHelper {
       checkAnswer(sql(s"SELECT c1 FROM $tableName WHERE c1 = COLLATE('a', 'UTF8_BINARY')"),
         Seq(Row("a")))
 
-      // fail with implicit mismatch, as function return should be considered implicit
-      checkError(
-        exception = intercept[AnalysisException] {
-          sql(s"SELECT c1 FROM $tableName " +
-            s"WHERE c1 = SUBSTR(COLLATE('a', 'UNICODE'), 0)")
-        },
-        condition = "COLLATION_MISMATCH.IMPLICIT",
-        parameters = Map(
-          "implicitTypes" -> """"STRING COLLATE UTF8_LCASE", "STRING COLLATE UNICODE""""
-        )
-      )
+      // explicit collation propagates up
+      checkAnswer(
+        sql(s"SELECT c1 FROM $tableName " +
+          s"WHERE c1 = SUBSTR(COLLATE('a', 'UNICODE'), 0)"),
+        Row("a"))
 
       // in operator
       checkAnswer(sql(s"SELECT c1 FROM $tableName WHERE c1 IN ('a')"),
@@ -742,9 +736,16 @@ class CollationSuite extends DatasourceV2SQLBase with AdaptiveSparkPlanHelper {
       )
 
       // concat + in
-      checkAnswer(sql(s"SELECT c1 FROM $tableName WHERE c1 || COLLATE('a', 'UTF8_BINARY') IN " +
-        s"(COLLATE('aa', 'UNICODE'))"),
-        Seq(Row("a")))
+      checkError(
+        exception = intercept[AnalysisException] {
+          sql(s"SELECT c1 FROM $tableName where c1 || COLLATE('a', 'UTF8_BINARY') IN " +
+            s"(COLLATE('aa', 'UNICODE'))")
+        },
+        condition = "COLLATION_MISMATCH.EXPLICIT",
+        parameters = Map(
+          "explicitTypes" -> """"STRING", "STRING COLLATE UNICODE""""
+        )
+      )
 
       // array creation supports implicit casting
       checkAnswer(sql(s"SELECT typeof(array('a' COLLATE UNICODE, 'b')[1])"),
@@ -765,14 +766,21 @@ class CollationSuite extends DatasourceV2SQLBase with AdaptiveSparkPlanHelper {
         exception = intercept[AnalysisException] {
           sql(s"SELECT array('A', 'a' COLLATE UNICODE) == array('b' COLLATE UNICODE_CI)")
         },
-        condition = "COLLATION_MISMATCH.IMPLICIT",
+        condition = "COLLATION_MISMATCH.EXPLICIT",
         parameters = Map(
-          "implicitTypes" -> """"STRING COLLATE UNICODE", "STRING COLLATE UNICODE_CI""""
+          "explicitTypes" -> """"STRING COLLATE UNICODE", "STRING COLLATE UNICODE_CI""""
         )
       )
 
-      checkAnswer(sql("SELECT array_join(array('a', 'b' collate UNICODE), 'c' collate UNICODE_CI)"),
-        Seq(Row("acb")))
+      checkError(
+        exception = intercept[AnalysisException] {
+          sql("SELECT array_join(array('a', 'b' collate UNICODE), 'c' collate UNICODE_CI)")
+        },
+        condition = "COLLATION_MISMATCH.EXPLICIT",
+        parameters = Map(
+          "explicitTypes" -> """"STRING COLLATE UNICODE", "STRING COLLATE UNICODE_CI""""
+        )
+      )
     }
   }
 
@@ -851,26 +859,6 @@ class CollationSuite extends DatasourceV2SQLBase with AdaptiveSparkPlanHelper {
     }
   }
 
-  test("SPARK-47692: Parameter markers with variable mapping") {
-    checkAnswer(
-      spark.sql(
-        "SELECT collation(:var1 || :var2)",
-        Map("var1" -> Literal.create('a', StringType("UTF8_BINARY")),
-            "var2" -> Literal.create('b', StringType("UNICODE")))),
-      Seq(Row("UTF8_BINARY"))
-    )
-
-    withSQLConf(SqlApiConf.DEFAULT_COLLATION -> "UNICODE") {
-      checkAnswer(
-        spark.sql(
-          "SELECT collation(:var1 || :var2)",
-          Map("var1" -> Literal.create('a', StringType("UTF8_BINARY")),
-              "var2" -> Literal.create('b', StringType("UNICODE")))),
-        Seq(Row("UNICODE"))
-      )
-    }
-  }
-
   test("SPARK-47210: Cast of default collated strings in IN expression") {
     val tableName = "t1"
     withTable(tableName) {