[SPARK-32307][SQL] ScalaUDF's canonicalized expression should exclude inputEncoders

[Ngone51]

What changes were proposed in this pull request?

Override canonicalized to empty the inputEncoders for the canonicalized ScalaUDF.

Why are the changes needed?

The following fails on the Master branch currently.

spark.udf.register("key", udf((m: Map[String, String]) => m.keys.head.toInt))
Seq(Map("1" -> "one", "2" -> "two")).toDF("a").createOrReplaceTempView("t")
checkAnswer(sql("SELECT key(a) AS k FROM t GROUP BY key(a)"), Row(1) :: Nil)

[info]   org.apache.spark.sql.AnalysisException: expression 't.`a`' is neither present in the group by, nor is it an aggregate function. Add to group by or wrap in first() (or first_value) if you don't care which value you get.;;
[info] Aggregate [UDF(a#6)], [UDF(a#6) AS k#8]
[info] +- SubqueryAlias t
[info]    +- Project [value#3 AS a#6]
[info]       +- LocalRelation [value#3]
[info]   at org.apache.spark.sql.catalyst.analysis.CheckAnalysis.failAnalysis(CheckAnalysis.scala:49)
[info]   at org.apache.spark.sql.catalyst.analysis.CheckAnalysis.failAnalysis$(CheckAnalysis.scala:48)
[info]   at org.apache.spark.sql.catalyst.analysis.Analyzer.failAnalysis(Analyzer.scala:130)
[info]   at org.apache.spark.sql.catalyst.analysis.CheckAnalysis.checkValidAggregateExpression$1(CheckAnalysis.scala:257)
[info]   at org.apache.spark.sql.catalyst.analysis.CheckAnalysis.$anonfun$checkAnalysis$10(CheckAnalysis.scala:259)
[info]   at org.apache.spark.sql.catalyst.analysis.CheckAnalysis.$anonfun$checkAnalysis$10$adapted(CheckAnalysis.scala:259)
[info]   at scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62)
[info]   at scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55)
[info]   at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49)
[info]   at org.apache.spark.sql.catalyst.analysis.CheckAnalysis.checkValidAggregateExpression$1(CheckAnalysis.scala:259)
[info]   at org.apache.spark.sql.catalyst.analysis.CheckAnalysis.$anonfun$checkAnalysis$10(CheckAnalysis.scala:259)
[info]   at org.apache.spark.sql.catalyst.analysis.CheckAnalysis.$anonfun$checkAnalysis$10$adapted(CheckAnalysis.scala:259)
[info]   at scala.collection.immutable.List.foreach(List.scala:392)
[info]   at org.apache.spark.sql.catalyst.analysis.CheckAnalysis.checkValidAggregateExpression$1(CheckAnalysis.scala:259)
...

We use the ruleResolveEncodersInUDF to resolve inputEncoders and the originalScalaUDF instance will be updated to a new ScalaUDF instance with the resolved encoders at the end. Note, during encoder resolving, types like map, array will be resolved to new expression(e.g. MapObjects, CatalystToExternalMap).

However, ExpressionEncoder can't be canonicalized. Thus, the canonicalized ScalaUDFs become different even if their original ScalaUDFs are the same. Finally, it fails the checkValidAggregateExpression when this ScalaUDF is used as a group expression.

Does this PR introduce any user-facing change?

Yes, users will not hit the exception after this fix.

How was this patch tested?

Added tests.

[dongjoon-hyun]

Hi, @Ngone51 .

The JIRA description and this PR description is misleading because it's not reproducible in the vanilla Apache Spark. Could you be more precise on the contribution of this PR?

scala> spark.version
res0: String = 3.0.0

scala> spark.udf.register("key", udf((m: Map[String, String]) => m.keys.head.toInt))
res1: org.apache.spark.sql.expressions.UserDefinedFunction = SparkUserDefinedFunction($Lambda$1954/1561881364@8937f62,IntegerType,List(Some(class[value[0]: map<string,string>])),None,false,true)

scala> Seq(Map("1" -> "one", "2" -> "two")).toDF("a").createOrReplaceTempView("t")

scala> sql("SELECT key(a) AS k FROM t GROUP BY key(a)").collect()
res3: Array[org.apache.spark.sql.Row] = Array([1])

[dongjoon-hyun]

If this exists on branch-3.0 and is not released yet, the affected version should be 3.0.1 instead of 3.0.0.

[dongjoon-hyun]

+1, LGTM. Thank you, @Ngone51 and @cloud-fan . I revised the JIRA and the PR description here.
Merged to master/3.0.

[dongjoon-hyun]

BTW, @cloud-fan and @HyukjinKwon . I merged this because GitHub Action passed.

[SparkQA]

Test build #125846 has finished for PR 29106 at commit 427d112.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[maropu]

late LGTM, tahnks, @Ngone51

[Ngone51]

thanks all!!

[dongjoon-hyun]

Hi, @Ngone51. Could you make a backporting PR for branch-3.0 again?
branch-3.0 seems to have a different reason for this fix.

[cloud-fan]

Merged to master/3.0.

has it been merged to 3.0 or not?

[Ngone51]

It has been reverted at 4ef535f.

branch-3.0 build fail at test SPARK-32307: Aggression that use array type input UDF as group expression since it does not support udf with type Array[Int]. Actually, yet, only the Master branch supports Array[Int] type after SPARK-31826.

This PR fixes the issue which introduced by SPARK-31826 and branch-3.0 actually doesn't have the issue but fail by the unsupported Array type.

So, we may not need to backport it to branch-3.0?

[cloud-fan]

OK, then let's not backport. @Ngone51 can you update the PR description to make it clear?

[Ngone51]

Sure

[dongjoon-hyun]

Thank you for clarification!

[cloud-fan]

@Ngone51 shall we do the same for outputEncoder?

[Ngone51]

Make sense. I'll do a follow-up.