[SPARK-47296][SQL][COLLATION] Fail unsupported functions for non-binary collations

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

@@ -40,6 +40,7 @@ class StringType private(val collationId: Int) extends AtomicType with Serializa
    * equality and hashing).
    */
   def isBinaryCollation: Boolean = CollationFactory.fetchCollation(collationId).isBinaryCollation
+  def isLowercaseCollation: Boolean = collationId == CollationFactory.LOWERCASE_COLLATION_ID
 
   /**
    * Type name that is shown to the customer.
@@ -54,8 +55,6 @@ class StringType private(val collationId: Int) extends AtomicType with Serializa
 
   override def hashCode(): Int = collationId.hashCode()
 
-  override private[sql] def acceptsType(other: DataType): Boolean = other.isInstanceOf[StringType]
-
   /**
    * The default size of a value of the StringType is 20 bytes.
    */
@@ -65,6 +64,8 @@ class StringType private(val collationId: Int) extends AtomicType with Serializa
 }
 
 /**
+ * Use StringType for expressions supporting only binary collation.
+ *
  * @since 1.3.0
  */
 @Stable

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

@@ -205,6 +205,10 @@ object AnsiTypeCoercion extends TypeCoercionBase {
       case (StringType, AnyTimestampType) =>
         Some(AnyTimestampType.defaultConcreteType)
 
+      // If a function expects StringType, no StringType instance should be implicitly cast to
+      // StringType with default collation.
+      case (st: StringType, StringType) => Some(st)
+
       case (DateType, AnyTimestampType) =>
         Some(AnyTimestampType.defaultConcreteType)
 

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

@@ -994,8 +994,10 @@ object TypeCoercion extends TypeCoercionBase {
       case (StringType, datetime: DatetimeType) => datetime
       case (StringType, AnyTimestampType) => AnyTimestampType.defaultConcreteType
       case (StringType, BinaryType) => BinaryType
+      case (st: StringType, StringType) => st
       // Cast any atomic type to string.
       case (any: AtomicType, StringType) if any != StringType => StringType
+      case (any: AtomicType, _: StringTypeCollated) if any != StringType => StringType
 
       // When we reach here, input type is not acceptable for any types in this type collection,
       // try to find the first one we can implicitly cast.

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

@@ -0,0 +1,77 @@
+/*
+ * 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.expressions
+
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.DataTypeMismatch
+import org.apache.spark.sql.catalyst.util.CollationFactory
+import org.apache.spark.sql.types.{AbstractDataType, DataType, StringType}
+
+object CollationTypeConstraints {
+
+  def checkCollationCompatibility(collationId: Int, dataTypes: Seq[DataType]): TypeCheckResult = {
+    val collationName = CollationFactory.fetchCollation(collationId).collationName
+    // Additional check needed for collation compatibility
+    dataTypes.collectFirst {
+      case stringType: StringType if stringType.collationId != collationId =>
+        val collation = CollationFactory.fetchCollation(stringType.collationId)
+        DataTypeMismatch(
+          errorSubClass = "COLLATION_MISMATCH",
+          messageParameters = Map(
+            "collationNameLeft" -> collationName,
+            "collationNameRight" -> collation.collationName
+          )
+        )
+    } getOrElse TypeCheckResult.TypeCheckSuccess
+  }
+
+}
+
+/**
+ * StringTypeCollated is an abstract class for StringType with collation support.
+ */
+abstract class StringTypeCollated extends AbstractDataType {
+  override private[sql] def defaultConcreteType: DataType = StringType
+}
+
+/**
+ * Use StringTypeBinary for expressions supporting only binary collation.
+ */
+case object StringTypeBinary extends StringTypeCollated {
+  override private[sql] def simpleString: String = "string_binary"
+  override private[sql] def acceptsType(other: DataType): Boolean =
+    other.isInstanceOf[StringType] && other.asInstanceOf[StringType].isBinaryCollation
+}
+
+/**
+ * Use StringTypeBinaryLcase for expressions supporting only binary and lowercase collation.
+ */
+case object StringTypeBinaryLcase extends StringTypeCollated {
+  override private[sql] def simpleString: String = "string_binary_lcase"
+  override private[sql] def acceptsType(other: DataType): Boolean =
+    other.isInstanceOf[StringType] && (other.asInstanceOf[StringType].isBinaryCollation ||
+      other.asInstanceOf[StringType].isLowercaseCollation)
+}
+
+/**
+ * Use StringTypeAnyCollation for expressions supporting all possible collation types.
+ */
+case object StringTypeAnyCollation extends StringTypeCollated {
+  override private[sql] def simpleString: String = "string_any_collation"
+  override private[sql] def acceptsType(other: DataType): Boolean = other.isInstanceOf[StringType]
+}

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

@@ -82,7 +82,7 @@ case class Collate(child: Expression, collationName: String)
   extends UnaryExpression with ExpectsInputTypes {
   private val collationId = CollationFactory.collationNameToId(collationName)
   override def dataType: DataType = StringType(collationId)
-  override def inputTypes: Seq[AbstractDataType] = Seq(StringType)
+  override def inputTypes: Seq[AbstractDataType] = Seq(StringTypeAnyCollation)
 
   override protected def withNewChildInternal(
     newChild: Expression): Expression = copy(newChild)

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

@@ -501,26 +501,15 @@ abstract class StringPredicate extends BinaryExpression
 
   def compare(l: UTF8String, r: UTF8String): Boolean
 
-  override def inputTypes: Seq[DataType] = Seq(StringType, StringType)
+  override def inputTypes: Seq[AbstractDataType] =
+    Seq(StringTypeAnyCollation, StringTypeAnyCollation)
 
   override def checkInputDataTypes(): TypeCheckResult = {
-    val checkResult = super.checkInputDataTypes()
-    if (checkResult.isFailure) {
-      return checkResult
-    }
-    // Additional check needed for collation compatibility
-    val rightCollationId: Int = right.dataType.asInstanceOf[StringType].collationId
-    if (collationId != rightCollationId) {
-      DataTypeMismatch(
-        errorSubClass = "COLLATION_MISMATCH",
-        messageParameters = Map(
-          "collationNameLeft" -> CollationFactory.fetchCollation(collationId).collationName,
-          "collationNameRight" -> CollationFactory.fetchCollation(rightCollationId).collationName
-        )
-      )
-    } else {
-      TypeCheckResult.TypeCheckSuccess
+    val defaultCheck = super.checkInputDataTypes()
+    if (defaultCheck.isFailure) {
+      return defaultCheck
     }
+    CollationTypeConstraints.checkCollationCompatibility(collationId, children.map(_.dataType))
   }
 
   protected override def nullSafeEval(input1: Any, input2: Any): Any =