[SPARK-48831][CONNECT] Make default column name of cast compatible with Spark Classic

[zhengruifeng]

What changes were proposed in this pull request?

I think there are two issues regarding the default column name of cast:
1, It seems unclear that when the name is the input column or CAST(...), e.g. in Spark Classic,

scala> spark.range(1).select(col("id").cast("string"), lit(1).cast("string"), col("id").cast("long"), lit(1).cast("long")).printSchema
warning: 1 deprecation (since 2.13.3); for details, enable `:setting -deprecation` or `:replay -deprecation`
root
 |-- id: string (nullable = false)
 |-- CAST(1 AS STRING): string (nullable = false)
 |-- id: long (nullable = false)
 |-- CAST(1 AS BIGINT): long (nullable = false)

2, the column name is not consistent between Spark Connect and Spark Classic.

This PR aims to resolve the second issue, that is, making default column name of cast compatible with Spark Classic, by comparing with classic implementation

spark/sql/core/src/main/scala/org/apache/spark/sql/Column.scala

Lines 1208 to 1212 in 9cf6dc8

	def cast(to: DataType): Column = withExpr {
	val cast = Cast(expr, CharVarcharUtils.replaceCharVarcharWithStringForCast(to))
	cast.setTagValue(Cast.USER_SPECIFIED_CAST, ())
	cast
	}

Why are the changes needed?

the default column name is not consistent with the spark classic

Does this PR introduce any user-facing change?

yes,

spark classic:

In [2]: spark.range(1).select(sf.lit(b'123').cast("STRING"), sf.lit(123).cast("STRING"), sf.lit(123).cast("LONG"), sf.lit(123).cast("DOUBLE")).show()
+-------------------------+-------------------+-------------------+-------------------+
|CAST(X'313233' AS STRING)|CAST(123 AS STRING)|CAST(123 AS BIGINT)|CAST(123 AS DOUBLE)|
+-------------------------+-------------------+-------------------+-------------------+
|                      123|                123|                123|              123.0|
+-------------------------+-------------------+-------------------+-------------------+

spark connect (before):

In [3]: spark.range(1).select(sf.lit(b'123').cast("STRING"), sf.lit(123).cast("STRING"), sf.lit(123).cast("LONG"), sf.lit(123).cast("DOUBLE")).show()
+---------+---+---+-----+
|X'313233'|123|123|  123|
+---------+---+---+-----+
|      123|123|123|123.0|
+---------+---+---+-----+

spark connect (after):

In [2]: spark.range(1).select(sf.lit(b'123').cast("STRING"), sf.lit(123).cast("STRING"), sf.lit(123).cast("LONG"), sf.lit(123).cast("DOUBLE")).show()
+-------------------------+-------------------+-------------------+-------------------+
|CAST(X'313233' AS STRING)|CAST(123 AS STRING)|CAST(123 AS BIGINT)|CAST(123 AS DOUBLE)|
+-------------------------+-------------------+-------------------+-------------------+
|                      123|                123|                123|              123.0|
+-------------------------+-------------------+-------------------+-------------------+

How was this patch tested?

added test

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

no

[allisonwang-db]

Good catch!

[allisonwang-db]

I am wondering why changing it this way can fix the schema difference? It looks like the cast expression is the same as before?

[zhengruifeng]

I guess it is affected by castExpr.setTagValue(Cast.USER_SPECIFIED_CAST, ())

spark/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/package.scala

Lines 93 to 110 in 0487507

	// Replaces attributes, string literals, complex type extractors with their pretty form so that
	// generated column names don't contain back-ticks or double-quotes.
	def usePrettyExpression(e: Expression): Expression = e transform {
	case a: Attribute => new PrettyAttribute(a)
	case Literal(s: UTF8String, StringType) => PrettyAttribute(s.toString, StringType)
	case Literal(v, t: NumericType) if v != null => PrettyAttribute(v.toString, t)
	case Literal(null, dataType) => PrettyAttribute("NULL", dataType)
	case e: GetStructField =>
	val name = e.name.getOrElse(e.childSchema(e.ordinal).name)
	PrettyAttribute(usePrettyExpression(e.child).sql + "." + name, e.dataType)
	case e: GetArrayStructFields =>
	PrettyAttribute(s"${usePrettyExpression(e.child)}.${e.field.name}", e.dataType)
	case r: InheritAnalysisRules =>
	PrettyAttribute(r.makeSQLString(r.parameters.map(toPrettySQL)), r.dataType)
	case c: Cast if c.getTagValue(Cast.USER_SPECIFIED_CAST).isEmpty =>
	PrettyAttribute(usePrettyExpression(c.child).sql, c.dataType)
	case p: PythonFuncExpression => PrettyPythonUDF(p.name, p.dataType, p.children)
	}

[HyukjinKwon]

Merged to master.