[SPARK-47430][SQL] Rework group by map type to fix bind reference exception

[ulysses-you]

What changes were proposed in this pull request?

This pr reworks the group by map type to fix issues:

• Can not bind reference excpetion at runtume since the attribute was wrapped by MapSort and we didi not transform the plan with new output
• The add MapSort rule should be put before PullOutGroupingExpressions to avoid complex expr existing in grouping keys

Why are the changes needed?

To fix issues.

for example:

select map(1, id) from range(10) group by map(1, id);


[INTERNAL_ERROR] Couldn't find _groupingexpression#18 in [mapsort(_groupingexpression#18)#19] SQLSTATE: XX000
org.apache.spark.SparkException: [INTERNAL_ERROR] Couldn't find _groupingexpression#18 in [mapsort(_groupingexpression#18)#19] SQLSTATE: XX000
	at org.apache.spark.SparkException$.internalError(SparkException.scala:92)
	at org.apache.spark.SparkException$.internalError(SparkException.scala:96)
	at org.apache.spark.sql.catalyst.expressions.BindReferences$$anonfun$bindReference$1.applyOrElse(BoundAttribute.scala:81)
	at org.apache.spark.sql.catalyst.expressions.BindReferences$$anonfun$bindReference$1.applyOrElse(BoundAttribute.scala:74)
	at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDownWithPruning$1(TreeNode.scala:470)

Does this PR introduce any user-facing change?

no, not released

How was this patch tested?

improve the tests to add more cases

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

no

[ulysses-you]

cc @cloud-fan @stevomitric thank you

[stevomitric]

cc @nebojsa-db

[HyukjinKwon]

To fix issues.

To fix which issue?

[ulysses-you]

@HyukjinKwon to fix the issue memtioned in pr description..

for example:

select map(1, id) from range(10) group by map(1, id);


[INTERNAL_ERROR] Couldn't find _groupingexpression#18 in [mapsort(_groupingexpression#18)#19] SQLSTATE: XX000
org.apache.spark.SparkException: [INTERNAL_ERROR] Couldn't find _groupingexpression#18 in [mapsort(_groupingexpression#18)#19] SQLSTATE: XX000
	at org.apache.spark.SparkException$.internalError(SparkException.scala:92)
	at org.apache.spark.SparkException$.internalError(SparkException.scala:96)
	at org.apache.spark.sql.catalyst.expressions.BindReferences$$anonfun$bindReference$1.applyOrElse(BoundAttribute.scala:81)
	at org.apache.spark.sql.catalyst.expressions.BindReferences$$anonfun$bindReference$1.applyOrElse(BoundAttribute.scala:74)
	at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDownWithPruning$1(TreeNode.scala:470)

I add this case in description to make it clear.

[yaooqinn]

Could you please ensure that the PR title does not sound like it's for refactoring if it's a bugfix?

[cloud-fan]

is it a simple rename? not sure why git diff doesn't detect it

[cloud-fan]

why rename? It's indeed inserting MapSort

[cloud-fan]

Suggested change

	exprToMapSort.getOrElseUpdate(
	expr.canonicalized, Alias(inserted, "_groupingmapsort")())
	.toAttribute
	exprToMapSort.getOrElseUpdate(
	expr.canonicalized,
	Alias(inserted, "_groupingmapsort")()
	).toAttribute

[cloud-fan]

let's add some comments to explain the rule order reasoning.

[ulysses-you]

thank you all, merged to master