[SPARK-46677][CONNECT][FOLLOWUP] Convert count(df["*"]) to count(1) on client side

[zhengruifeng]

What changes were proposed in this pull request?

before #44689, df["*"] and sf.col("*") are both convert to UnresolvedStar, and then Count(UnresolvedStar) is converted to Count(1) in Analyzer:

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

Lines 1893 to 1897 in 381f369

	case f0: UnresolvedFunction if !f0.isDistinct &&
	f0.nameParts.map(_.toLowerCase(Locale.ROOT)) == Seq("count") &&
	f0.arguments == Seq(UnresolvedStar(None)) =>
	// Transform COUNT(*) into COUNT(1).
	f0.copy(nameParts = Seq("count"), arguments = Seq(Literal(1)))

in that fix, we introduced a new node UnresolvedDataFrameStar for df["*"] which will be replaced to ResolvedStar later. Unfortunately, it doesn't match Count(UnresolvedStar) any more.
So it causes:

In [1]: from pyspark.sql import functions as sf

In [2]: df1 = spark.createDataFrame([{"id": 1, "val": "v"}])

In [3]: df1.select(sf.count(df1["*"]))
Out[3]: DataFrame[count(id, val): bigint]

which should be

In [3]: df1.select(sf.count(df1["*"]))
Out[3]: DataFrame[count(1): bigint]

In vanilla Spark, it is up to the count function to make such conversion sf.count(df1["*"]) -> sf.count(sf.lit(1)), see

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

Lines 422 to 436 in e8dfcd3

	/**
	* Aggregate function: returns the number of items in a group.
	*
	* @group agg_funcs
	* @since 1.3.0
	*/
	def count(e: Column): Column = {
	val withoutStar = e.expr match {
	// Turn count(*) into count(1)
	case _: Star => Column(Literal(1))
	case _ => e
	}
	Column.fn("count", withoutStar)
	}

So it is a natural way to fix this behavior on the client side.

Why are the changes needed?

to keep the behavior

Does this PR introduce any user-facing change?

it fix a behavior change introduced in #44689

How was this patch tested?

added ut

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

no

[zhengruifeng]

thanks, merged to master