fix

Author: mihailotim-db

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

@@ -19,7 +19,9 @@ package org.apache.spark.sql.catalyst.analysis
 
 import scala.jdk.CollectionConverters._
 
+import org.apache.spark.SparkException
 import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.ExprUtils.toSQLExpr
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.rules.Rule
 
@@ -39,10 +41,20 @@ object PullOutNondeterministic extends Rule[LogicalPlan] {
       val nondeterToAttr =
         NondeterministicExpressionCollection.getNondeterministicToAttributes(a.groupingExpressions)
       val newChild = Project(a.child.output ++ nondeterToAttr.values.asScala.toSeq, a.child)
-      a.transformExpressions { case e =>
+      val deterministicAggregate = a.transformExpressions { case e =>
         Option(nondeterToAttr.get(e)).map(_.toAttribute).getOrElse(e)
       }.copy(child = newChild)
 
+      deterministicAggregate.groupingExpressions.foreach(expr => if (!expr.deterministic) {
+        throw SparkException.internalError(
+          msg = s"Non-deterministic expression '${toSQLExpr(expr)}' should not appear in " +
+            "grouping expression.",
+          context = expr.origin.getQueryContext,
+          summary = expr.origin.context.summary)
+      })
+
+      deterministicAggregate
+
     // Don't touch collect metrics. Top-level metrics are not supported (check analysis will fail)
     // and we want to retain them inside the aggregate functions.
     case m: CollectMetrics => m

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExprUtils.scala

@@ -20,7 +20,6 @@ package org.apache.spark.sql.catalyst.expressions
 import java.text.{DecimalFormat, DecimalFormatSymbols, ParsePosition}
 import java.util.Locale
 
-import org.apache.spark.SparkException
 import org.apache.spark.sql.catalyst.analysis._
 import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
 import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{DataTypeMismatch, TypeCheckSuccess}
@@ -209,17 +208,6 @@ object ExprUtils extends EvalHelper with QueryErrorsBase {
             "sqlExpr" -> toSQLExpr(expr),
             "dataType" -> toSQLType(expr.dataType)))
       }
-
-      if (!expr.deterministic) {
-        // This is just a sanity check, our analysis rule PullOutNondeterministic should
-        // already pull out those nondeterministic expressions and evaluate them in
-        // a Project node.
-        throw SparkException.internalError(
-          msg = s"Non-deterministic expression '${toSQLExpr(expr)}' should not appear in " +
-            "grouping expression.",
-          context = expr.origin.getQueryContext,
-          summary = expr.origin.context.summary)
-      }
     }
 
     a.groupingExpressions.foreach(checkValidGroupingExprs)