[SPARK-52705][SQL] Refactor deterministic check for grouping expressions

[mihailotim-db]

What changes were proposed in this pull request?

Move check for non-deterministic expressions in grouping expressions from ExprUtils to CheckAnalysis.

Why are the changes needed?

This is necessary in order to be able to utilize PullOutNonDeterminstic rule as a post-processing rewrite rule in single-pass analyzer. Because ExprUtils.assertValidAggregate is called during the bottom-up traversal, we can't check for non-determinstic expressions there

Does this PR introduce any user-facing change?

How was this patch tested?

Existing tests.

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

No

[cloud-fan]

the spark connect test failure is unrelated, thanks, merging to master!

[dongjoon-hyun]

@mihailotim-db and @cloud-fan , is this safe to remove?

This PR only added this logic at CheckAnalysis.scala and PullOutNondeterministic.scala. However, assertValidAggregation seems to be used by more places?

$ git grep assertValidAggregation | grep -v CheckAnalysis.scala | grep -v PullOutNondeterministic.scala
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/AggregateResolver.scala:   * [[Aggregate]] using the [[ExprUtils.assertValidAggregation]], update the `scopes` with the
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/AggregateResolver.scala:        ExprUtils.assertValidAggregation(resolvedAggregate)
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/AggregateResolver.scala:   * unintentional exceptions from being thrown by [[ExprUtils.assertValidAggregation]], so the
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/LateralColumnAliasResolver.scala:        ExprUtils.assertValidAggregation(aggregate)
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ProjectResolver.scala:   * validate it using the [[ExprUtils.assertValidAggregation]].
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ProjectResolver.scala:          ExprUtils.assertValidAggregation(aggregate)
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExprUtils.scala:  def assertValidAggregation(a: Aggregate): Unit = {
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/LogicalPlan.scala:          ExprUtils.assertValidAggregation(a)

Can we add back this to be safe? I don't see any negative effect to have this here. Do have a chance for this to cause any failure at single pass?

[mihailotim-db]

We are moving this to CheckAnalysis. It going to be ran just before assertValidAggregation. For single-pass, we are going to run PullOutNondeterministic after we finish resolving the plan. Because we are calling ExprUtils.assertValidAggregation during the bottom-up pass, we can't have this check there because it would cause false positive failures (non-deterministic expressions would get pushed down later)

[dongjoon-hyun]

In other words, can I understand like you are saying that this PR is technically reducing test coverage of this code path (by reducing the invocation paths) due to the single-pass requirement, @mihailotim-db ?

[mihailotim-db]

In the sense that we are not invoking this between rules but only in CheckAnalysis, yes. But from what I can see, we only call assertValidAggregation from single-pass analyzer, CheckAnalysis and Optimizer, so we don't have an issue there

[peter-toth]

Thanks @mihailotim-db for the explanation. Late LGTM.

[dongjoon-hyun]

cc @peter-toth