(spark) branch master updated: [SPARK-51114][SQL] Refactor PullOutNondeterministic rule

[we...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

wenchen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/master by this push:
     new f52a6790f787 [SPARK-51114][SQL] Refactor PullOutNondeterministic rule
f52a6790f787 is described below

commit f52a6790f787c28194838bca7d3119610da7c97a
Author: mihailoale-db <mi...@databricks.com>
AuthorDate: Wed Feb 12 10:18:32 2025 +0800

    [SPARK-51114][SQL] Refactor PullOutNondeterministic rule
    
    ### What changes were proposed in this pull request?
    Refactor `PullOutNondeterministic` rule body so it can be reused in the single-pass analyzer.
    
    ### Why are the changes needed?
    Better reusability of the `PullOutNondeterministic` rule.
    
    ### Does this PR introduce _any_ user-facing change?
    No.
    
    ### How was this patch tested?
    Existing tests (just refactoring).
    
    ### Was this patch authored or co-authored using generative AI tooling?
    No.
    
    Closes #49837 from mihailoale-db/nondeterministicrefactor.
    
    Authored-by: mihailoale-db <mi...@databricks.com>
    Signed-off-by: Wenchen Fan <we...@databricks.com>
---
 .../NondeterministicExpressionCollection.scala     | 42 ++++++++++++++++++++++
 .../analysis/PullOutNondeterministic.scala         | 22 +++---------
 2 files changed, 46 insertions(+), 18 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/NondeterministicExpressionCollection.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/NondeterministicExpressionCollection.scala
new file mode 100644
index 000000000000..206bba77d1e1
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/NondeterministicExpressionCollection.scala
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.catalyst.expressions._
+
+object NondeterministicExpressionCollection {
+  def getNondeterministicToAttributes(
+      expressions: Seq[Expression]): Map[Expression, NamedExpression] = {
+    expressions
+      .filterNot(_.deterministic)
+      .flatMap { expr =>
+        val leafNondeterministic = expr.collect {
+          case n: Nondeterministic => n
+          case udf: UserDefinedExpression if !udf.deterministic => udf
+        }
+        leafNondeterministic.distinct.map { e =>
+          val ne = e match {
+            case n: NamedExpression => n
+            case _ => Alias(e, "_nondeterministic")()
+          }
+          e -> ne
+        }
+      }
+      .toMap
+  }
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/PullOutNondeterministic.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/PullOutNondeterministic.scala
index 395514216683..c10cc2faf35d 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/PullOutNondeterministic.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/PullOutNondeterministic.scala
@@ -34,7 +34,8 @@ object PullOutNondeterministic extends Rule[LogicalPlan] {
     case f: Filter => f
 
     case a: Aggregate if a.groupingExpressions.exists(!_.deterministic) =>
-      val nondeterToAttr = getNondeterToAttr(a.groupingExpressions)
+      val nondeterToAttr =
+        NondeterministicExpressionCollection.getNondeterministicToAttributes(a.groupingExpressions)
       val newChild = Project(a.child.output ++ nondeterToAttr.values, a.child)
       a.transformExpressions { case e =>
         nondeterToAttr.get(e).map(_.toAttribute).getOrElse(e)
@@ -51,27 +52,12 @@ object PullOutNondeterministic extends Rule[LogicalPlan] {
     // from LogicalPlan, currently we only do it for UnaryNode which has same output
     // schema with its child.
     case p: UnaryNode if p.output == p.child.output && p.expressions.exists(!_.deterministic) =>
-      val nondeterToAttr = getNondeterToAttr(p.expressions)
+      val nondeterToAttr =
+        NondeterministicExpressionCollection.getNondeterministicToAttributes(p.expressions)
       val newPlan = p.transformExpressions { case e =>
         nondeterToAttr.get(e).map(_.toAttribute).getOrElse(e)
       }
       val newChild = Project(p.child.output ++ nondeterToAttr.values, p.child)
       Project(p.output, newPlan.withNewChildren(newChild :: Nil))
   }
-
-  private def getNondeterToAttr(exprs: Seq[Expression]): Map[Expression, NamedExpression] = {
-    exprs.filterNot(_.deterministic).flatMap { expr =>
-      val leafNondeterministic = expr.collect {
-        case n: Nondeterministic => n
-        case udf: UserDefinedExpression if !udf.deterministic => udf
-      }
-      leafNondeterministic.distinct.map { e =>
-        val ne = e match {
-          case n: NamedExpression => n
-          case _ => Alias(e, "_nondeterministic")()
-        }
-        e -> ne
-      }
-    }.toMap
-  }
 }


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org