[DRAFT] Resolve default string producing expressions in analyzer

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

@@ -17,9 +17,10 @@
 
 package org.apache.spark.sql.catalyst.analysis
 
-import org.apache.spark.sql.catalyst.expressions.{Cast, Expression, Literal}
+import org.apache.spark.sql.catalyst.expressions.{Cast, DefaultStringProducingExpression, Expression, Literal}
 import org.apache.spark.sql.catalyst.plans.logical.{AddColumns, AlterColumns, AlterColumnSpec, AlterViewAs, ColumnDefinition, CreateView, LogicalPlan, QualifiedColType, ReplaceColumns, V1CreateTablePlan, V2CreateTablePlan}
 import org.apache.spark.sql.catalyst.rules.{Rule, RuleExecutor}
+import org.apache.spark.sql.catalyst.trees.TreeNodeTag
 import org.apache.spark.sql.types.{DataType, StringType}
 
 /**
@@ -29,6 +30,8 @@ import org.apache.spark.sql.types.{DataType, StringType}
  * So, we will just use UTF8_BINARY for now.
  */
 object ResolveDefaultStringTypes extends Rule[LogicalPlan] {
+  private val CAST_ADDED_TAG = new TreeNodeTag[Unit]("castAddedTag")
+
   def apply(plan: LogicalPlan): LogicalPlan = {
     val newPlan = apply0(plan)
     if (plan.ne(newPlan)) {
@@ -149,6 +152,21 @@ object ResolveDefaultStringTypes extends Rule[LogicalPlan] {
 
     case Literal(value, dt) if hasDefaultStringType(dt) =>
       newType => Literal(value, replaceDefaultStringType(dt, newType))
+
+    case expression if needsCast(expression) =>
+      expression.setTagValue(CAST_ADDED_TAG, ())
+      newType => {
+        if (newType == StringType) {
+          expression
+        } else {
+          Cast(expression, replaceDefaultStringType(expression.dataType, newType))
+        }
+      }
+  }
+
+  private def needsCast(expression: Expression): Boolean = expression match {
+    case ex: DefaultStringProducingExpression => ex.getTagValue(CAST_ADDED_TAG).isEmpty
+    case _ => false
   }
 
   private def hasDefaultStringType(dataType: DataType): Boolean =

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

@@ -1467,5 +1467,5 @@ case class MultiCommutativeOp(
  * Trait for expressions whose data type should be a default string type.
  */
 trait DefaultStringProducingExpression extends Expression {
-  override def dataType: DataType = SQLConf.get.defaultStringType
+  override def dataType: DataType = StringType
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/variant/variantExpressions.scala

@@ -858,12 +858,13 @@ case class SchemaOfVariantAgg(
     extends TypedImperativeAggregate[DataType]
     with ExpectsInputTypes
     with QueryErrorsBase
-    with DefaultStringProducingExpression
     with UnaryLike[Expression] {
   def this(child: Expression) = this(child, 0, 0)
 
   override def inputTypes: Seq[AbstractDataType] = Seq(VariantType)
 
+  override def dataType: DataType = SQLConf.get.defaultStringType
+
   override def nullable: Boolean = false
 
   override def createAggregationBuffer(): DataType = NullType

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/finishAnalysis.scala

@@ -33,7 +33,6 @@ import org.apache.spark.sql.catalyst.util.DateTimeUtils
 import org.apache.spark.sql.catalyst.util.DateTimeUtils.{convertSpecialDate, convertSpecialTimestamp, convertSpecialTimestampNTZ, instantToMicros, localDateTimeToMicros}
 import org.apache.spark.sql.catalyst.util.TypeUtils.toSQLExpr
 import org.apache.spark.sql.connector.catalog.CatalogManager
-import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
 
 
@@ -153,14 +152,15 @@ case class ReplaceCurrentLike(catalogManager: CatalogManager) extends Rule[Logic
     val currentCatalog = catalogManager.currentCatalog.name()
     val currentUser = CurrentUserContext.getCurrentUser
 
-    plan.transformAllExpressionsWithPruning(_.containsPattern(CURRENT_LIKE)) {
+    val res = plan.transformAllExpressionsWithPruning(_.containsPattern(CURRENT_LIKE)) {
       case CurrentDatabase() =>
-        Literal.create(currentNamespace, SQLConf.get.defaultStringType)
+        Literal.create(currentNamespace)
       case CurrentCatalog() =>
-        Literal.create(currentCatalog, SQLConf.get.defaultStringType)
+        Literal.create(currentCatalog)
       case CurrentUser() =>
-        Literal.create(currentUser, SQLConf.get.defaultStringType)
+        Literal.create(currentUser)
     }
+    res
   }
 }
 

sql/core/src/test/scala/org/apache/spark/sql/collation/DefaultCollationTestSuite.scala

@@ -31,6 +31,13 @@ abstract class DefaultCollationTestSuite extends QueryTest with SharedSparkSessi
   def testView: String = "test_view"
   protected val fullyQualifiedPrefix = s"${CollationFactory.CATALOG}.${CollationFactory.SCHEMA}."
 
+  val defaultStringProducingExpressions: Seq[String] = Seq(
+    "current_timezone()", "current_database()", "md5('Spark' collate unicode)",
+    "soundex('Spark' collate unicode)", "url_encode('https://spark.apache.org' collate unicode)",
+    "url_decode('https%3A%2F%2Fspark.apache.org')", "uuid()", "chr(65)", "collation('UNICODE')",
+    "version()", "space(5)", "randstr(5, 123)"
+  )
+
   def withSessionCollationAndTable(collation: String, testTables: String*)(f: => Unit): Unit = {
     withTable(testTables: _*) {
       withSessionCollation(collation) {
@@ -233,6 +240,25 @@ abstract class DefaultCollationTestSuite extends QueryTest with SharedSparkSessi
     }
   }
 
+  test("expressions that have default string result in DDL should have object collation") {
+    withSessionCollationAndTable("UTF8_LCASE", testTable) {
+
+      val columns = defaultStringProducingExpressions.zipWithIndex.map {
+        case (expr, index) => s"$expr AS c${index + 1}"
+      }.mkString(", ")
+
+      sql(s"""
+        |CREATE TABLE $testTable
+        |USING $dataSource AS
+        |SELECT $columns
+        |""".stripMargin)
+
+      (1 to defaultStringProducingExpressions.length).foreach { index =>
+        assertTableColumnCollation(testTable, s"c$index", "UTF8_BINARY")
+      }
+    }
+  }
+
   // endregion
 
   // region DML tests
@@ -389,6 +415,17 @@ abstract class DefaultCollationTestSuite extends QueryTest with SharedSparkSessi
     }
   }
 
+  test("expressions that have default string result in DML should have session collation") {
+    withSessionCollation("UTF8_LCASE") {
+      defaultStringProducingExpressions.foreach { expr =>
+        checkAnswer(
+          sql(s"SELECT COLLATION($expr)"),
+          Seq(Row(fullyQualifiedPrefix + "UTF8_LCASE"))
+        )
+      }
+    }
+  }
+
   // endregion
 }