You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "haiyangsun-db (via GitHub)" <gi...@apache.org> on 2026/07/31 09:48:17 UTC
[PR] [SPARK-55278][SQL] Add scalar external UDF planning support [spark]
haiyangsun-db opened a new pull request, #57670:
URL: https://github.com/apache/spark/pull/57670
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as new features, bug fixes, or other behavior changes. Documentation-only updates are not considered user-facing changes.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
### Was this patch authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55278][SQL] Add scalar external UDF planning support [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #57670:
URL: https://github.com/apache/spark/pull/57670#discussion_r3926334124
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkOptimizer.scala:
##########
@@ -147,7 +154,7 @@ class SparkOptimizer(
* batch executing the [[ExperimentalMethods]] optimizer rules. This hook can be used to add
* custom optimizer batches to the Spark optimizer.
*
- * Note that 'Extract Python UDFs' batch is an exception and ran after the batches defined here.
+ * Note that 'Extract UDFs' batch is an exception and ran after the batches defined here.
Review Comment:
**Nit (P3):** The sentence mixes present and past tense: `is an exception and ran after`. Since this documents the current batch ordering, please use `is an exception and runs after the batches defined here`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/ExtractExternalUDFFromWindow.scala:
##########
@@ -0,0 +1,71 @@
+/*
+ * 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.execution.externalUDF
+
+import scala.collection.mutable
+
+import org.apache.spark.sql.catalyst.expressions.{Alias, Expression, ExprId,
+ ExternalUserDefinedFunction, NamedExpression, WindowExpression}
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Project, Window}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern.{EXTERNAL_UDF, WINDOW}
+
+/**
+ * Extracts external UDFs that are parents of window expressions from a [[Window]] operator.
+ * The window expressions are evaluated by the [[Window]], and [[PlanExternalUDFs]] subsequently
+ * converts the external UDFs in the new [[Project]] into evaluation nodes above it.
+ */
+private[sql] object ExtractExternalUDFFromWindow extends Rule[LogicalPlan] {
+
+ private def containsExternalUDFOverWindowExpression(expression: Expression): Boolean = {
+ expression.exists {
+ case udf: ExternalUserDefinedFunction =>
+ udf.exists(_.isInstanceOf[WindowExpression])
+ case _ => false
+ }
+ }
+
+ override def apply(plan: LogicalPlan): LogicalPlan = {
+ plan.transformWithPruning(
+ _.containsAllPatterns(EXTERNAL_UDF, WINDOW)) {
+ case window: Window
+ if window.windowExpressions.exists(containsExternalUDFOverWindowExpression) =>
+ val windowProjectExprIds = mutable.Set.empty[ExprId]
+ val windowProjectList = mutable.ArrayBuffer.empty[NamedExpression]
+ val externalUdfProjectList = window.windowExpressions.map { expression =>
+ if (containsExternalUDFOverWindowExpression(expression)) {
+ expression.transformDown {
+ case windowExpression: WindowExpression =>
+ val alias = Alias(windowExpression, s"w_${windowProjectList.size}")()
+ windowProjectList += alias
+ alias.toAttribute
+ }.asInstanceOf[NamedExpression]
+ } else {
+ if (!windowProjectExprIds.contains(expression.exprId)) {
+ windowProjectList += expression
+ windowProjectExprIds += expression.exprId
+ }
+ expression.toAttribute
+ }
+ }
+ Project(
+ externalUdfProjectList,
Review Comment:
**Blocking (P1):** This Project drops every pass-through attribute from `window.child.output`: `externalUdfProjectList` is derived only from `window.windowExpressions`, while `Window.output` is `child.output ++ windowExpressions.map(_.toAttribute)`. For a parent that selects both the input column and `externalUDF(lag(input) over (...))`, the rewritten child no longer produces the input attribute. Please include `window.child.output` in this projection in the original output order, and add an assertion that the window rewrite preserves both the input and UDF result columns.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2452,6 +2456,7 @@ object PushPredicateThroughNonJoin extends Rule[LogicalPlan] with PredicateHelpe
case _: RebalancePartitions => true
case _: ScriptTransformation => true
case _: Sort => true
+ case _: ExecuteExternalUDF => true
Review Comment:
**Non-blocking (P2):** The added predicate-pushdown test asserts only that the child predicate moved below `ExecuteExternalUDF`; it never checks that the predicate referencing the UDF result stays above it. A regression that drops `resultPredicate` would therefore pass while returning unfiltered rows. Please also assert an upper `Filter` whose condition is semantically equal to `resultPredicate`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExternalUserDefinedFunction.scala:
##########
@@ -61,6 +67,24 @@ case class ExternalUserDefinedFunction(
override def nullable: Boolean = udfNullable
+ override def checkInputDataTypes(): TypeCheckResult = {
+ inputTypes match {
+ case Some(types) if types.length != children.length =>
+ throw QueryCompilationErrors.wrongNumArgsError(
+ name = name.getOrElse(prettyName),
+ validParametersCount = Seq(types.length),
+ actualNumber = children.length)
+ case Some(types) =>
+ ExpectsInputTypes.checkInputDataTypes(children, types)
+ case None => TypeCheckSuccess
+ }
+ }
+
+ // Worker specifications and payloads can contain sensitive execution details.
Review Comment:
**Non-blocking (P2):** The safe-rendering contract names payloads separately from worker specifications, but the test puts its unique secret only in a worker environment variable; the payload is the public UDF name. A renderer could start appending `udf.payload` and all current secret assertions would still pass. Please use a distinct payload-only sentinel and assert it is absent from `sql`, `toString`, logged conditions, and logical and physical tree strings.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PlanExternalUDFs.scala:
##########
@@ -0,0 +1,266 @@
+/*
+ * 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.execution.externalUDF
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.LogKeys.JOIN_CONDITION
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
+import org.apache.spark.sql.catalyst.plans.InnerLike
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern.{AGGREGATE, EXTERNAL_UDF, JOIN}
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Converts each scalar external UDF expression into a separate logical evaluation node.
+ * Join-condition handling mirrors `ExtractPythonUDFFromJoinCondition`.
+ *
+ * TODO(SPARK-55278): Add an external UDF equivalent of `ExtractPythonUDFFromLambda`.
+ * TODO(SPARK-55278): Revisit sharing placement logic with the Python UDF extractors after
+ * external UDF planning semantics stabilize.
+ */
+private[sql] object PlanExternalUDFs
+ extends Rule[LogicalPlan] with Logging with PredicateHelper {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan match {
+ // A correlated subquery is rewritten as a join and revisits this rule later.
+ case subquery: Subquery if subquery.correlated => plan
+ case _ if !conf.getConf(SQLConf.UNIFIED_UDF_EXECUTION_ENABLED) =>
+ if (plan.containsPattern(EXTERNAL_UDF)) {
+ throw QueryCompilationErrors.externalUDFsDisabledError(
+ SQLConf.UNIFIED_UDF_EXECUTION_ENABLED.key)
+ }
+ plan
+ case _ =>
+ var preparedPlan = extractExternalUDFFromJoinCondition(plan)
+ preparedPlan = extractExternalUDFFromAggregate(preparedPlan)
+ preparedPlan = extractGroupingExternalUDFFromAggregate(preparedPlan)
+ preparedPlan.transformUpWithPruning(_.containsPattern(EXTERNAL_UDF)) {
+ // These nodes already own their external UDF expressions.
+ case udfPlan: ExternalUDF => udfPlan
+ case other => extract(other)
+ }
+ }
+
+ private def hasUnevaluableExternalUDF(expression: Expression, join: Join): Boolean = {
+ expression.exists {
+ case udf: ExternalUserDefinedFunction =>
+ !canEvaluate(udf, join.left) && !canEvaluate(udf, join.right)
+ case _ => false
+ }
+ }
+
+ private def extractExternalUDFFromJoinCondition(plan: LogicalPlan): LogicalPlan = {
+ plan.transformUpWithPruning(_.containsAllPatterns(EXTERNAL_UDF, JOIN)) {
+ case join @ Join(_, _, joinType, Some(condition), _)
+ if hasUnevaluableExternalUDF(condition, join) =>
+ if (!joinType.isInstanceOf[InnerLike]) {
+ // Match `PYTHON_UDF_IN_ON_CLAUSE`: moving a cross-side UDF to a post-join filter
+ // changes the semantics of non-inner joins.
+ throw QueryCompilationErrors.useExternalUDFInJoinConditionUnsupportedError(joinType)
+ }
+
+ val (udfConditions, otherConditions) = splitConjunctivePredicates(condition)
+ .partition(hasUnevaluableExternalUDF(_, join))
+ val newCondition = if (otherConditions.isEmpty) {
+ logWarning(log"The join condition:${MDC(JOIN_CONDITION, condition)} " +
+ log"of the join plan contains external UDFs only, " +
+ log"so it will be moved out and the join plan will become a cross join.")
+ None
+ } else {
+ Some(otherConditions.reduceLeft(And))
+ }
+ Filter(udfConditions.reduceLeft(And), join.copy(condition = newCondition))
+ }
+ }
+
+ private def belongsToAggregate(
+ expression: Expression,
+ groupingExpressions: ExpressionSet): Boolean = {
+ expression.isInstanceOf[AggregateExpression] ||
+ groupingExpressions.contains(expression)
+ }
+
+ private def hasExternalUDFOverAggregate(
+ expression: Expression,
+ groupingExpressions: ExpressionSet): Boolean = {
+ expression.exists {
+ case udf: ExternalUserDefinedFunction =>
+ udf.references.isEmpty || udf.exists(belongsToAggregate(_, groupingExpressions))
+ case _ => false
+ }
+ }
+
+ private def extractExternalUDFFromAggregate(plan: LogicalPlan): LogicalPlan = {
+ plan.transformUpWithPruning(_.containsAllPatterns(EXTERNAL_UDF, AGGREGATE)) {
+ case aggregate: Aggregate =>
+ val groupingExpressions = ExpressionSet(aggregate.groupingExpressions)
+ if (!aggregate.aggregateExpressions.exists(
+ hasExternalUDFOverAggregate(_, groupingExpressions))) {
+ aggregate
+ } else {
+ val projectExpressions = ArrayBuffer.empty[NamedExpression]
+ val aggregateExpressions = ArrayBuffer.empty[NamedExpression]
+ aggregate.aggregateExpressions.foreach { expression =>
+ if (hasExternalUDFOverAggregate(expression, groupingExpressions)) {
+ val newExpression = expression.transformDown {
+ case child: Expression if belongsToAggregate(child, groupingExpressions) =>
+ val alias = child match {
+ case named: NamedExpression => named
+ case other => Alias(other, "agg")()
+ }
+ aggregateExpressions += alias
+ alias.toAttribute
+ }
+ projectExpressions += newExpression.asInstanceOf[NamedExpression]
+ } else {
+ aggregateExpressions += expression
+ projectExpressions += expression.toAttribute
+ }
+ }
+ Project(
+ projectExpressions.toSeq,
+ aggregate.copy(aggregateExpressions = aggregateExpressions.toSeq))
+ }
+ }
+ }
+
+ private def hasExternalUDF(expression: Expression): Boolean = {
+ expression.exists(_.isInstanceOf[ExternalUserDefinedFunction])
+ }
+
+ private def extractGroupingExternalUDFFromAggregate(plan: LogicalPlan): LogicalPlan = {
+ plan.transformUpWithPruning(_.containsAllPatterns(EXTERNAL_UDF, AGGREGATE)) {
+ case aggregate: Aggregate if aggregate.groupingExpressions.exists(hasExternalUDF) =>
+ val projectExpressions = ArrayBuffer.empty[NamedExpression]
+ val groupingExpressions = ArrayBuffer.empty[Expression]
+ val attributeMap = ArrayBuffer.empty[
+ (ExternalUserDefinedFunction, NamedExpression)]
+
+ def mappedAttribute(udf: ExternalUserDefinedFunction): Option[NamedExpression] = {
+ attributeMap.collectFirst {
+ case (candidate, attribute) if sameUDF(candidate, udf) => attribute
+ }
+ }
+
+ aggregate.groupingExpressions.foreach { expression =>
+ if (hasExternalUDF(expression)) {
+ val newExpression = expression.transformDown {
+ case udf: ExternalUserDefinedFunction =>
+ assert(udf.udfDeterministic,
+ "Non-deterministic external UDFs should not appear in grouping expressions")
+ mappedAttribute(udf).getOrElse {
+ val alias = Alias(udf, "groupingExternalUDF")()
+ projectExpressions += alias
+ attributeMap += ((udf, alias.toAttribute))
+ alias.toAttribute
+ }
+ }
+ groupingExpressions += newExpression
+ } else {
+ groupingExpressions += expression
+ }
+ }
+
+ val aggregateExpressions = aggregate.aggregateExpressions.map { expression =>
+ expression.transformUp {
+ case udf: ExternalUserDefinedFunction if udf.udfDeterministic =>
+ mappedAttribute(udf).getOrElse(udf)
+ }.asInstanceOf[NamedExpression]
+ }
+ aggregate.copy(
+ groupingExpressions = groupingExpressions.toSeq,
+ aggregateExpressions = aggregateExpressions,
+ child = Project((projectExpressions ++ aggregate.child.output).toSeq, aggregate.child))
+ }
+ }
+
+ private def containsExternalUDF(expression: Expression): Boolean = {
+ expression.exists(_.isInstanceOf[ExternalUserDefinedFunction])
+ }
+
+ private def isEvaluable(udf: ExternalUserDefinedFunction): Boolean = {
+ !udf.children.exists(containsExternalUDF)
+ }
+
+ private def sameUDF(
+ left: ExternalUserDefinedFunction,
+ right: ExternalUserDefinedFunction): Boolean = {
+ if (left.deterministic && right.deterministic) {
+ val normalizedPayload = Array.emptyByteArray
+ left.payload.sameElements(right.payload) &&
+ left.copy(payload = normalizedPayload).semanticEquals(
Review Comment:
**Non-blocking (P2):** This equality path is used specifically when mapping grouping UDFs, but the existing different-worker test nests dependent UDFs and never calls `sameUDF`. Please add a grouping case with otherwise equivalent deterministic UDFs that differ only by `workerSpec`, and assert that they produce two evaluation nodes with the correct worker/result mapping. That test should fail if canonicalization ever stops considering the worker specification.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PlanExternalUDFs.scala:
##########
@@ -0,0 +1,266 @@
+/*
+ * 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.execution.externalUDF
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.LogKeys.JOIN_CONDITION
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
+import org.apache.spark.sql.catalyst.plans.InnerLike
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern.{AGGREGATE, EXTERNAL_UDF, JOIN}
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Converts each scalar external UDF expression into a separate logical evaluation node.
+ * Join-condition handling mirrors `ExtractPythonUDFFromJoinCondition`.
+ *
+ * TODO(SPARK-55278): Add an external UDF equivalent of `ExtractPythonUDFFromLambda`.
+ * TODO(SPARK-55278): Revisit sharing placement logic with the Python UDF extractors after
+ * external UDF planning semantics stabilize.
+ */
+private[sql] object PlanExternalUDFs
+ extends Rule[LogicalPlan] with Logging with PredicateHelper {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan match {
+ // A correlated subquery is rewritten as a join and revisits this rule later.
+ case subquery: Subquery if subquery.correlated => plan
+ case _ if !conf.getConf(SQLConf.UNIFIED_UDF_EXECUTION_ENABLED) =>
Review Comment:
**Blocking (P1):** This live SQLConf read can disagree with the planner retained by SessionState. The config is non-static and can be changed with `spark.conf.set`, but BaseSessionStateBuilder selects ClassicExternalUDFPlanner or UnifiedExternalUDFPlanner only once. For example, constructing SessionState with the flag enabled and then disabling it leaves the unified planner producing `MapPartitionsExternalUDF`, which this branch now rejects. Please make planner selection and optimizer gating observe the value at the same lifecycle point and cover both post-construction transitions.
**Recommended change:** Dispatch both external-UDF plan construction and optimizer gating from the current session value for each operation.
**Why this works:** Move mode selection behind a per-operation planner dispatch so both the producer and PlanExternalUDFs consume one coherent SQLConf snapshot.
**Scope:** BaseSessionStateBuilder or ExternalUDFPlanner mode dispatch, PlanExternalUDFs, and focused session-config lifecycle tests.
**Compatibility:** Preserve the existing key and fixed-value behavior while making supported runtime updates take effect consistently.
**Risks:** Planner dispatch must not create mixed classic and unified nodes within one operation.
**Constraints:** Keep disabled-mode rejection and enabled planning semantics unchanged when the value is not mutated.
**Success:** Switching the config in either direction after SessionState construction selects one coherent mode and mapInPandas no longer fails because its producer and optimizer disagree.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PlanExternalUDFs.scala:
##########
@@ -0,0 +1,266 @@
+/*
+ * 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.execution.externalUDF
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.LogKeys.JOIN_CONDITION
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
+import org.apache.spark.sql.catalyst.plans.InnerLike
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern.{AGGREGATE, EXTERNAL_UDF, JOIN}
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Converts each scalar external UDF expression into a separate logical evaluation node.
+ * Join-condition handling mirrors `ExtractPythonUDFFromJoinCondition`.
+ *
+ * TODO(SPARK-55278): Add an external UDF equivalent of `ExtractPythonUDFFromLambda`.
+ * TODO(SPARK-55278): Revisit sharing placement logic with the Python UDF extractors after
+ * external UDF planning semantics stabilize.
+ */
+private[sql] object PlanExternalUDFs
+ extends Rule[LogicalPlan] with Logging with PredicateHelper {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan match {
+ // A correlated subquery is rewritten as a join and revisits this rule later.
+ case subquery: Subquery if subquery.correlated => plan
+ case _ if !conf.getConf(SQLConf.UNIFIED_UDF_EXECUTION_ENABLED) =>
+ if (plan.containsPattern(EXTERNAL_UDF)) {
+ throw QueryCompilationErrors.externalUDFsDisabledError(
+ SQLConf.UNIFIED_UDF_EXECUTION_ENABLED.key)
+ }
+ plan
+ case _ =>
+ var preparedPlan = extractExternalUDFFromJoinCondition(plan)
+ preparedPlan = extractExternalUDFFromAggregate(preparedPlan)
+ preparedPlan = extractGroupingExternalUDFFromAggregate(preparedPlan)
+ preparedPlan.transformUpWithPruning(_.containsPattern(EXTERNAL_UDF)) {
+ // These nodes already own their external UDF expressions.
+ case udfPlan: ExternalUDF => udfPlan
+ case other => extract(other)
+ }
+ }
+
+ private def hasUnevaluableExternalUDF(expression: Expression, join: Join): Boolean = {
+ expression.exists {
+ case udf: ExternalUserDefinedFunction =>
+ !canEvaluate(udf, join.left) && !canEvaluate(udf, join.right)
+ case _ => false
+ }
+ }
+
+ private def extractExternalUDFFromJoinCondition(plan: LogicalPlan): LogicalPlan = {
+ plan.transformUpWithPruning(_.containsAllPatterns(EXTERNAL_UDF, JOIN)) {
+ case join @ Join(_, _, joinType, Some(condition), _)
+ if hasUnevaluableExternalUDF(condition, join) =>
Review Comment:
**Non-blocking (P2):** Please add the adjacent allowed-boundary case: a left outer join whose external UDF references only the left child. The test should verify that planning succeeds, `ExecuteExternalUDF` is placed below the Join on the left, and the Join condition uses its result attribute. Current cross-side tests would not catch a regression that rejects every external UDF in a non-inner ON clause.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55278][SQL] Add scalar external UDF planning support [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #57670:
URL: https://github.com/apache/spark/pull/57670#discussion_r3920863766
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/ExtractExternalUDFFromWindow.scala:
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.execution.externalUDF
+
+import scala.collection.mutable
+
+import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, Expression, ExprId,
+ ExternalUserDefinedFunction, NamedExpression, WindowExpression}
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Project, Window}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern.{EXTERNAL_UDF, WINDOW}
+
+/**
+ * Extracts external UDFs that are parents of window expressions from a [[Window]] operator.
+ * The window expressions are evaluated by the [[Window]], and [[PlanExternalUDFs]] subsequently
+ * converts the external UDFs in the new [[Project]] into evaluation nodes above it.
+ */
+private[sql] object ExtractExternalUDFFromWindow extends Rule[LogicalPlan] {
+
+ private def containsExternalUDFOverWindowExpression(expression: Expression): Boolean = {
+ expression.exists {
+ case udf: ExternalUserDefinedFunction =>
+ udf.exists(_.isInstanceOf[WindowExpression])
+ case _ => false
+ }
+ }
+
+ override def apply(plan: LogicalPlan): LogicalPlan = {
+ plan.transformWithPruning(
+ _.containsAllPatterns(EXTERNAL_UDF, WINDOW)) {
+ case window: Window
+ if window.windowExpressions.exists(containsExternalUDFOverWindowExpression) =>
+ val windowProjectExprIds = mutable.Set.empty[ExprId]
+ val windowProjectList = mutable.ArrayBuffer.empty[NamedExpression]
+ val externalUdfProjectList = window.windowExpressions.map { expression =>
+ if (containsExternalUDFOverWindowExpression(expression)) {
+ expression.transformDown {
+ case windowExpression: WindowExpression =>
+ val alias = Alias(windowExpression, s"w_${windowProjectList.size}")()
+ windowProjectList += alias
+ alias.toAttribute
+ case attribute: Attribute if !windowProjectExprIds.contains(attribute.exprId) =>
+ windowProjectList += attribute
Review Comment:
**Nit (P3):** `Window.output` already prepends every attribute from `child.output`, so adding this ordinary child attribute to `windowProjectList` exposes the same ExprId a second time. This is reachable for `externalUDF(windowExpression, input)`. Please leave direct child attributes out of `Window.windowExpressions` and add a focused test that checks unique Window output ExprIds while preserving both UDF arguments.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/ExecuteExternalUDFExec.scala:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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.execution.externalUDF
+
+import org.apache.spark.annotation.Experimental
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{
+ Attribute,
+ AttributeSet,
+ ExternalUserDefinedFunction
+}
+import org.apache.spark.sql.errors.QueryExecutionErrors
+import org.apache.spark.sql.execution.SparkPlan
+import org.apache.spark.udf.worker.UDFWorkerSpecification
+
+/**
+ * :: Experimental ::
+ * Physical plan node that evaluates one scalar UDF in an external worker process.
+ *
+ * @param udf UDF expression evaluated by the worker session.
+ * @param resultAttr Output attribute for the UDF expression.
+ * @param child Child plan providing input rows.
+ */
+@Experimental
+case class ExecuteExternalUDFExec(
+ udf: ExternalUserDefinedFunction,
Review Comment:
**Nit (P3):** The class Scaladoc currently says this node evaluates the UDF in a worker, but `doExecute` always throws `methodNotImplementedError`, and the PR intentionally leaves scalar execution out of scope. Please describe this as a planning placeholder with unimplemented execution until the worker path exists.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PlanExternalUDFs.scala:
##########
@@ -0,0 +1,266 @@
+/*
+ * 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.execution.externalUDF
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.LogKeys.JOIN_CONDITION
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
+import org.apache.spark.sql.catalyst.plans.InnerLike
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern.{AGGREGATE, EXTERNAL_UDF, JOIN}
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Converts each scalar external UDF expression into a separate logical evaluation node.
+ * Join-condition handling mirrors `ExtractPythonUDFFromJoinCondition`.
+ *
+ * TODO(SPARK-55278): Add an external UDF equivalent of `ExtractPythonUDFFromLambda`.
+ * TODO(SPARK-55278): Revisit sharing placement logic with the Python UDF extractors after
+ * external UDF planning semantics stabilize.
+ */
+private[sql] object PlanExternalUDFs
+ extends Rule[LogicalPlan] with Logging with PredicateHelper {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan match {
+ // A correlated subquery is rewritten as a join and revisits this rule later.
+ case subquery: Subquery if subquery.correlated => plan
+ case _ if !conf.getConf(SQLConf.UNIFIED_UDF_EXECUTION_ENABLED) =>
+ if (plan.containsPattern(EXTERNAL_UDF)) {
+ throw QueryCompilationErrors.externalUDFsDisabledError(
+ SQLConf.UNIFIED_UDF_EXECUTION_ENABLED.key)
+ }
+ plan
+ case _ =>
+ var preparedPlan = extractExternalUDFFromJoinCondition(plan)
+ preparedPlan = extractExternalUDFFromAggregate(preparedPlan)
+ preparedPlan = extractGroupingExternalUDFFromAggregate(preparedPlan)
+ preparedPlan.transformUpWithPruning(_.containsPattern(EXTERNAL_UDF)) {
+ // These nodes already own their external UDF expressions.
+ case udfPlan: ExternalUDF => udfPlan
+ case other => extract(other)
+ }
+ }
+
+ private def hasUnevaluableExternalUDF(expression: Expression, join: Join): Boolean = {
+ expression.exists {
+ case udf: ExternalUserDefinedFunction =>
+ !canEvaluate(udf, join.left) && !canEvaluate(udf, join.right)
+ case _ => false
+ }
+ }
+
+ private def extractExternalUDFFromJoinCondition(plan: LogicalPlan): LogicalPlan = {
+ plan.transformUpWithPruning(_.containsAllPatterns(EXTERNAL_UDF, JOIN)) {
+ case join @ Join(_, _, joinType, Some(condition), _)
+ if hasUnevaluableExternalUDF(condition, join) =>
+ if (!joinType.isInstanceOf[InnerLike]) {
+ // Match `PYTHON_UDF_IN_ON_CLAUSE`: moving a cross-side UDF to a post-join filter
+ // changes the semantics of non-inner joins.
+ throw QueryCompilationErrors.useExternalUDFInJoinConditionUnsupportedError(joinType)
+ }
+
+ val (udfConditions, otherConditions) = splitConjunctivePredicates(condition)
+ .partition(hasUnevaluableExternalUDF(_, join))
+ val newCondition = if (otherConditions.isEmpty) {
+ logWarning(log"The join condition:${MDC(JOIN_CONDITION, condition)} " +
+ log"of the join plan contains external UDFs only, " +
+ log"so it will be moved out and the join plan will become a cross join.")
+ None
+ } else {
+ Some(otherConditions.reduceLeft(And))
Review Comment:
**Nit (P3):** The suite exercises only the `otherConditions.isEmpty` path. Please add a mixed condition such as `leftKey = rightKey AND externalUDF(leftValue, rightValue)` and assert that the equality remains on the `Join` while only the external-UDF predicate moves to the post-join `Filter`; otherwise a regression here can silently turn a keyed join into a Cartesian plan.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PlanExternalUDFs.scala:
##########
@@ -0,0 +1,266 @@
+/*
+ * 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.execution.externalUDF
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.LogKeys.JOIN_CONDITION
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
+import org.apache.spark.sql.catalyst.plans.InnerLike
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern.{AGGREGATE, EXTERNAL_UDF, JOIN}
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Converts each scalar external UDF expression into a separate logical evaluation node.
+ * Join-condition handling mirrors `ExtractPythonUDFFromJoinCondition`.
+ *
+ * TODO(SPARK-55278): Add an external UDF equivalent of `ExtractPythonUDFFromLambda`.
+ * TODO(SPARK-55278): Revisit sharing placement logic with the Python UDF extractors after
+ * external UDF planning semantics stabilize.
+ */
+private[sql] object PlanExternalUDFs
+ extends Rule[LogicalPlan] with Logging with PredicateHelper {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan match {
+ // A correlated subquery is rewritten as a join and revisits this rule later.
+ case subquery: Subquery if subquery.correlated => plan
+ case _ if !conf.getConf(SQLConf.UNIFIED_UDF_EXECUTION_ENABLED) =>
+ if (plan.containsPattern(EXTERNAL_UDF)) {
+ throw QueryCompilationErrors.externalUDFsDisabledError(
+ SQLConf.UNIFIED_UDF_EXECUTION_ENABLED.key)
+ }
+ plan
+ case _ =>
+ var preparedPlan = extractExternalUDFFromJoinCondition(plan)
+ preparedPlan = extractExternalUDFFromAggregate(preparedPlan)
+ preparedPlan = extractGroupingExternalUDFFromAggregate(preparedPlan)
+ preparedPlan.transformUpWithPruning(_.containsPattern(EXTERNAL_UDF)) {
+ // These nodes already own their external UDF expressions.
+ case udfPlan: ExternalUDF => udfPlan
+ case other => extract(other)
+ }
+ }
+
+ private def hasUnevaluableExternalUDF(expression: Expression, join: Join): Boolean = {
+ expression.exists {
+ case udf: ExternalUserDefinedFunction =>
+ !canEvaluate(udf, join.left) && !canEvaluate(udf, join.right)
+ case _ => false
+ }
+ }
+
+ private def extractExternalUDFFromJoinCondition(plan: LogicalPlan): LogicalPlan = {
+ plan.transformUpWithPruning(_.containsAllPatterns(EXTERNAL_UDF, JOIN)) {
+ case join @ Join(_, _, joinType, Some(condition), _)
+ if hasUnevaluableExternalUDF(condition, join) =>
+ if (!joinType.isInstanceOf[InnerLike]) {
+ // Match `PYTHON_UDF_IN_ON_CLAUSE`: moving a cross-side UDF to a post-join filter
+ // changes the semantics of non-inner joins.
+ throw QueryCompilationErrors.useExternalUDFInJoinConditionUnsupportedError(joinType)
+ }
+
+ val (udfConditions, otherConditions) = splitConjunctivePredicates(condition)
+ .partition(hasUnevaluableExternalUDF(_, join))
+ val newCondition = if (otherConditions.isEmpty) {
+ logWarning(log"The join condition:${MDC(JOIN_CONDITION, condition)} " +
+ log"of the join plan contains external UDFs only, " +
+ log"so it will be moved out and the join plan will become a cross join.")
+ None
+ } else {
+ Some(otherConditions.reduceLeft(And))
+ }
+ Filter(udfConditions.reduceLeft(And), join.copy(condition = newCondition))
+ }
+ }
+
+ private def belongsToAggregate(
+ expression: Expression,
+ groupingExpressions: ExpressionSet): Boolean = {
+ expression.isInstanceOf[AggregateExpression] ||
+ groupingExpressions.contains(expression)
+ }
+
+ private def hasExternalUDFOverAggregate(
+ expression: Expression,
+ groupingExpressions: ExpressionSet): Boolean = {
+ expression.exists {
+ case udf: ExternalUserDefinedFunction =>
+ udf.references.isEmpty || udf.exists(belongsToAggregate(_, groupingExpressions))
+ case _ => false
+ }
+ }
+
+ private def extractExternalUDFFromAggregate(plan: LogicalPlan): LogicalPlan = {
+ plan.transformUpWithPruning(_.containsAllPatterns(EXTERNAL_UDF, AGGREGATE)) {
+ case aggregate: Aggregate =>
+ val groupingExpressions = ExpressionSet(aggregate.groupingExpressions)
+ if (!aggregate.aggregateExpressions.exists(
+ hasExternalUDFOverAggregate(_, groupingExpressions))) {
+ aggregate
+ } else {
+ val projectExpressions = ArrayBuffer.empty[NamedExpression]
+ val aggregateExpressions = ArrayBuffer.empty[NamedExpression]
+ aggregate.aggregateExpressions.foreach { expression =>
+ if (hasExternalUDFOverAggregate(expression, groupingExpressions)) {
+ val newExpression = expression.transformDown {
+ case child: Expression if belongsToAggregate(child, groupingExpressions) =>
+ val alias = child match {
+ case named: NamedExpression => named
+ case other => Alias(other, "agg")()
+ }
+ aggregateExpressions += alias
+ alias.toAttribute
+ }
+ projectExpressions += newExpression.asInstanceOf[NamedExpression]
+ } else {
+ aggregateExpressions += expression
+ projectExpressions += expression.toAttribute
+ }
+ }
+ Project(
+ projectExpressions.toSeq,
+ aggregate.copy(aggregateExpressions = aggregateExpressions.toSeq))
+ }
+ }
+ }
+
+ private def hasExternalUDF(expression: Expression): Boolean = {
+ expression.exists(_.isInstanceOf[ExternalUserDefinedFunction])
+ }
+
+ private def extractGroupingExternalUDFFromAggregate(plan: LogicalPlan): LogicalPlan = {
+ plan.transformUpWithPruning(_.containsAllPatterns(EXTERNAL_UDF, AGGREGATE)) {
+ case aggregate: Aggregate if aggregate.groupingExpressions.exists(hasExternalUDF) =>
+ val projectExpressions = ArrayBuffer.empty[NamedExpression]
+ val groupingExpressions = ArrayBuffer.empty[Expression]
+ val attributeMap = ArrayBuffer.empty[
+ (ExternalUserDefinedFunction, NamedExpression)]
+
+ def mappedAttribute(udf: ExternalUserDefinedFunction): Option[NamedExpression] = {
+ attributeMap.collectFirst {
+ case (candidate, attribute) if sameUDF(candidate, udf) => attribute
+ }
+ }
+
+ aggregate.groupingExpressions.foreach { expression =>
+ if (hasExternalUDF(expression)) {
+ val newExpression = expression.transformDown {
+ case udf: ExternalUserDefinedFunction =>
+ assert(udf.udfDeterministic,
Review Comment:
**Non-blocking (P2):** `PullOutNondeterministic` processes this Aggregate before `PlanExternalUDFs`, but `NondeterministicExpressionCollection` only collects `Nondeterministic` and nondeterministic `UserDefinedExpression` leaves. `ExternalUserDefinedFunction` implements neither, so a grouping UDF with `udfDeterministic = false` is left in the Aggregate and triggers the internal "should not appear in grouping expression" error before this assertion is reached. Please teach the collector to hoist nondeterministic external UDFs and add a grouping-key regression that verifies the UDF is evaluated below the Aggregate.
**Recommended change:** Teach NondeterministicExpressionCollection to hoist nondeterministic ExternalUserDefinedFunction leaves and add a focused aggregate-grouping regression test.
**Why this works:** Add an ExternalUserDefinedFunction case guarded by nondeterminism alongside the existing user-defined-expression case, allowing PullOutNondeterministic to materialize an alias in the child Project and replace the Aggregate grouping key before PlanExternalUDFs runs.
**Scope:** Catalyst nondeterministic-expression collection plus the focused scalar external-UDF planning test suite.
**Compatibility:** This changes only internal analysis of a currently failing unified external-UDF query; deterministic grouping behavior and the disabled configuration path remain unchanged.
**Risks:** The new collection case must preserve canonicalized deduplication and must not hoist deterministic external UDFs.
**Constraints:** Keep the existing PullOutNondeterministic ordering and its requirement that Aggregate grouping expressions are deterministic after rewriting.
**Success:** A nondeterministic external UDF used as both grouping key and selected grouping expression analyzes without an internal error, is evaluated in the child plan, and leaves a deterministic attribute in the Aggregate.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58467][SQL] Add scalar external UDF planning support [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon closed pull request #57670: [SPARK-58467][SQL] Add scalar external UDF planning support
URL: https://github.com/apache/spark/pull/57670
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55278][SQL] Add scalar external UDF planning support [spark]
Posted by "haiyangsun-db (via GitHub)" <gi...@apache.org>.
haiyangsun-db commented on code in PR #57670:
URL: https://github.com/apache/spark/pull/57670#discussion_r3926986408
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PlanExternalUDFs.scala:
##########
@@ -0,0 +1,266 @@
+/*
+ * 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.execution.externalUDF
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.LogKeys.JOIN_CONDITION
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
+import org.apache.spark.sql.catalyst.plans.InnerLike
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern.{AGGREGATE, EXTERNAL_UDF, JOIN}
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Converts each scalar external UDF expression into a separate logical evaluation node.
+ * Join-condition handling mirrors `ExtractPythonUDFFromJoinCondition`.
+ *
+ * TODO(SPARK-55278): Add an external UDF equivalent of `ExtractPythonUDFFromLambda`.
+ * TODO(SPARK-55278): Revisit sharing placement logic with the Python UDF extractors after
+ * external UDF planning semantics stabilize.
+ */
+private[sql] object PlanExternalUDFs
+ extends Rule[LogicalPlan] with Logging with PredicateHelper {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan match {
+ // A correlated subquery is rewritten as a join and revisits this rule later.
+ case subquery: Subquery if subquery.correlated => plan
+ case _ if !conf.getConf(SQLConf.UNIFIED_UDF_EXECUTION_ENABLED) =>
+ if (plan.containsPattern(EXTERNAL_UDF)) {
+ throw QueryCompilationErrors.externalUDFsDisabledError(
+ SQLConf.UNIFIED_UDF_EXECUTION_ENABLED.key)
+ }
+ plan
+ case _ =>
+ var preparedPlan = extractExternalUDFFromJoinCondition(plan)
+ preparedPlan = extractExternalUDFFromAggregate(preparedPlan)
+ preparedPlan = extractGroupingExternalUDFFromAggregate(preparedPlan)
+ preparedPlan.transformUpWithPruning(_.containsPattern(EXTERNAL_UDF)) {
+ // These nodes already own their external UDF expressions.
+ case udfPlan: ExternalUDF => udfPlan
+ case other => extract(other)
+ }
+ }
+
+ private def hasUnevaluableExternalUDF(expression: Expression, join: Join): Boolean = {
+ expression.exists {
+ case udf: ExternalUserDefinedFunction =>
+ !canEvaluate(udf, join.left) && !canEvaluate(udf, join.right)
+ case _ => false
+ }
+ }
+
+ private def extractExternalUDFFromJoinCondition(plan: LogicalPlan): LogicalPlan = {
+ plan.transformUpWithPruning(_.containsAllPatterns(EXTERNAL_UDF, JOIN)) {
+ case join @ Join(_, _, joinType, Some(condition), _)
+ if hasUnevaluableExternalUDF(condition, join) =>
Review Comment:
Fixed. Added the left-local UDF/left-outer-join case and verified that evaluation is placed below the join on the left while the join condition uses the result attribute.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExternalUserDefinedFunction.scala:
##########
@@ -61,6 +67,24 @@ case class ExternalUserDefinedFunction(
override def nullable: Boolean = udfNullable
+ override def checkInputDataTypes(): TypeCheckResult = {
+ inputTypes match {
+ case Some(types) if types.length != children.length =>
+ throw QueryCompilationErrors.wrongNumArgsError(
+ name = name.getOrElse(prettyName),
+ validParametersCount = Seq(types.length),
+ actualNumber = children.length)
+ case Some(types) =>
+ ExpectsInputTypes.checkInputDataTypes(children, types)
+ case None => TypeCheckSuccess
+ }
+ }
+
+ // Worker specifications and payloads can contain sensitive execution details.
Review Comment:
Fixed. The rendering regression now uses separate worker-spec and payload-only sentinels and verifies that neither appears in SQL, expression strings, logged conditions, or logical and physical plan trees.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PlanExternalUDFs.scala:
##########
@@ -0,0 +1,266 @@
+/*
+ * 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.execution.externalUDF
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.LogKeys.JOIN_CONDITION
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
+import org.apache.spark.sql.catalyst.plans.InnerLike
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern.{AGGREGATE, EXTERNAL_UDF, JOIN}
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Converts each scalar external UDF expression into a separate logical evaluation node.
+ * Join-condition handling mirrors `ExtractPythonUDFFromJoinCondition`.
+ *
+ * TODO(SPARK-55278): Add an external UDF equivalent of `ExtractPythonUDFFromLambda`.
+ * TODO(SPARK-55278): Revisit sharing placement logic with the Python UDF extractors after
+ * external UDF planning semantics stabilize.
+ */
+private[sql] object PlanExternalUDFs
+ extends Rule[LogicalPlan] with Logging with PredicateHelper {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan match {
+ // A correlated subquery is rewritten as a join and revisits this rule later.
+ case subquery: Subquery if subquery.correlated => plan
+ case _ if !conf.getConf(SQLConf.UNIFIED_UDF_EXECUTION_ENABLED) =>
+ if (plan.containsPattern(EXTERNAL_UDF)) {
+ throw QueryCompilationErrors.externalUDFsDisabledError(
+ SQLConf.UNIFIED_UDF_EXECUTION_ENABLED.key)
+ }
+ plan
+ case _ =>
+ var preparedPlan = extractExternalUDFFromJoinCondition(plan)
+ preparedPlan = extractExternalUDFFromAggregate(preparedPlan)
+ preparedPlan = extractGroupingExternalUDFFromAggregate(preparedPlan)
+ preparedPlan.transformUpWithPruning(_.containsPattern(EXTERNAL_UDF)) {
+ // These nodes already own their external UDF expressions.
+ case udfPlan: ExternalUDF => udfPlan
+ case other => extract(other)
+ }
+ }
+
+ private def hasUnevaluableExternalUDF(expression: Expression, join: Join): Boolean = {
+ expression.exists {
+ case udf: ExternalUserDefinedFunction =>
+ !canEvaluate(udf, join.left) && !canEvaluate(udf, join.right)
+ case _ => false
+ }
+ }
+
+ private def extractExternalUDFFromJoinCondition(plan: LogicalPlan): LogicalPlan = {
+ plan.transformUpWithPruning(_.containsAllPatterns(EXTERNAL_UDF, JOIN)) {
+ case join @ Join(_, _, joinType, Some(condition), _)
+ if hasUnevaluableExternalUDF(condition, join) =>
+ if (!joinType.isInstanceOf[InnerLike]) {
+ // Match `PYTHON_UDF_IN_ON_CLAUSE`: moving a cross-side UDF to a post-join filter
+ // changes the semantics of non-inner joins.
+ throw QueryCompilationErrors.useExternalUDFInJoinConditionUnsupportedError(joinType)
+ }
+
+ val (udfConditions, otherConditions) = splitConjunctivePredicates(condition)
+ .partition(hasUnevaluableExternalUDF(_, join))
+ val newCondition = if (otherConditions.isEmpty) {
+ logWarning(log"The join condition:${MDC(JOIN_CONDITION, condition)} " +
+ log"of the join plan contains external UDFs only, " +
+ log"so it will be moved out and the join plan will become a cross join.")
+ None
+ } else {
+ Some(otherConditions.reduceLeft(And))
+ }
+ Filter(udfConditions.reduceLeft(And), join.copy(condition = newCondition))
+ }
+ }
+
+ private def belongsToAggregate(
+ expression: Expression,
+ groupingExpressions: ExpressionSet): Boolean = {
+ expression.isInstanceOf[AggregateExpression] ||
+ groupingExpressions.contains(expression)
+ }
+
+ private def hasExternalUDFOverAggregate(
+ expression: Expression,
+ groupingExpressions: ExpressionSet): Boolean = {
+ expression.exists {
+ case udf: ExternalUserDefinedFunction =>
+ udf.references.isEmpty || udf.exists(belongsToAggregate(_, groupingExpressions))
+ case _ => false
+ }
+ }
+
+ private def extractExternalUDFFromAggregate(plan: LogicalPlan): LogicalPlan = {
+ plan.transformUpWithPruning(_.containsAllPatterns(EXTERNAL_UDF, AGGREGATE)) {
+ case aggregate: Aggregate =>
+ val groupingExpressions = ExpressionSet(aggregate.groupingExpressions)
+ if (!aggregate.aggregateExpressions.exists(
+ hasExternalUDFOverAggregate(_, groupingExpressions))) {
+ aggregate
+ } else {
+ val projectExpressions = ArrayBuffer.empty[NamedExpression]
+ val aggregateExpressions = ArrayBuffer.empty[NamedExpression]
+ aggregate.aggregateExpressions.foreach { expression =>
+ if (hasExternalUDFOverAggregate(expression, groupingExpressions)) {
+ val newExpression = expression.transformDown {
+ case child: Expression if belongsToAggregate(child, groupingExpressions) =>
+ val alias = child match {
+ case named: NamedExpression => named
+ case other => Alias(other, "agg")()
+ }
+ aggregateExpressions += alias
+ alias.toAttribute
+ }
+ projectExpressions += newExpression.asInstanceOf[NamedExpression]
+ } else {
+ aggregateExpressions += expression
+ projectExpressions += expression.toAttribute
+ }
+ }
+ Project(
+ projectExpressions.toSeq,
+ aggregate.copy(aggregateExpressions = aggregateExpressions.toSeq))
+ }
+ }
+ }
+
+ private def hasExternalUDF(expression: Expression): Boolean = {
+ expression.exists(_.isInstanceOf[ExternalUserDefinedFunction])
+ }
+
+ private def extractGroupingExternalUDFFromAggregate(plan: LogicalPlan): LogicalPlan = {
+ plan.transformUpWithPruning(_.containsAllPatterns(EXTERNAL_UDF, AGGREGATE)) {
+ case aggregate: Aggregate if aggregate.groupingExpressions.exists(hasExternalUDF) =>
+ val projectExpressions = ArrayBuffer.empty[NamedExpression]
+ val groupingExpressions = ArrayBuffer.empty[Expression]
+ val attributeMap = ArrayBuffer.empty[
+ (ExternalUserDefinedFunction, NamedExpression)]
+
+ def mappedAttribute(udf: ExternalUserDefinedFunction): Option[NamedExpression] = {
+ attributeMap.collectFirst {
+ case (candidate, attribute) if sameUDF(candidate, udf) => attribute
+ }
+ }
+
+ aggregate.groupingExpressions.foreach { expression =>
+ if (hasExternalUDF(expression)) {
+ val newExpression = expression.transformDown {
+ case udf: ExternalUserDefinedFunction =>
+ assert(udf.udfDeterministic,
+ "Non-deterministic external UDFs should not appear in grouping expressions")
+ mappedAttribute(udf).getOrElse {
+ val alias = Alias(udf, "groupingExternalUDF")()
+ projectExpressions += alias
+ attributeMap += ((udf, alias.toAttribute))
+ alias.toAttribute
+ }
+ }
+ groupingExpressions += newExpression
+ } else {
+ groupingExpressions += expression
+ }
+ }
+
+ val aggregateExpressions = aggregate.aggregateExpressions.map { expression =>
+ expression.transformUp {
+ case udf: ExternalUserDefinedFunction if udf.udfDeterministic =>
+ mappedAttribute(udf).getOrElse(udf)
+ }.asInstanceOf[NamedExpression]
+ }
+ aggregate.copy(
+ groupingExpressions = groupingExpressions.toSeq,
+ aggregateExpressions = aggregateExpressions,
+ child = Project((projectExpressions ++ aggregate.child.output).toSeq, aggregate.child))
+ }
+ }
+
+ private def containsExternalUDF(expression: Expression): Boolean = {
+ expression.exists(_.isInstanceOf[ExternalUserDefinedFunction])
+ }
+
+ private def isEvaluable(udf: ExternalUserDefinedFunction): Boolean = {
+ !udf.children.exists(containsExternalUDF)
+ }
+
+ private def sameUDF(
+ left: ExternalUserDefinedFunction,
+ right: ExternalUserDefinedFunction): Boolean = {
+ if (left.deterministic && right.deterministic) {
+ val normalizedPayload = Array.emptyByteArray
+ left.payload.sameElements(right.payload) &&
+ left.copy(payload = normalizedPayload).semanticEquals(
Review Comment:
Fixed. Added a grouping regression with otherwise equivalent deterministic UDFs that differ only by `workerSpec`; it verifies two evaluation nodes and the correct worker/result mapping.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55278][SQL] Add scalar external UDF planning support [spark]
Posted by "haiyangsun-db (via GitHub)" <gi...@apache.org>.
haiyangsun-db commented on code in PR #57670:
URL: https://github.com/apache/spark/pull/57670#discussion_r3926984937
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/ExtractExternalUDFFromWindow.scala:
##########
@@ -0,0 +1,71 @@
+/*
+ * 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.execution.externalUDF
+
+import scala.collection.mutable
+
+import org.apache.spark.sql.catalyst.expressions.{Alias, Expression, ExprId,
+ ExternalUserDefinedFunction, NamedExpression, WindowExpression}
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Project, Window}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern.{EXTERNAL_UDF, WINDOW}
+
+/**
+ * Extracts external UDFs that are parents of window expressions from a [[Window]] operator.
+ * The window expressions are evaluated by the [[Window]], and [[PlanExternalUDFs]] subsequently
+ * converts the external UDFs in the new [[Project]] into evaluation nodes above it.
+ */
+private[sql] object ExtractExternalUDFFromWindow extends Rule[LogicalPlan] {
+
+ private def containsExternalUDFOverWindowExpression(expression: Expression): Boolean = {
+ expression.exists {
+ case udf: ExternalUserDefinedFunction =>
+ udf.exists(_.isInstanceOf[WindowExpression])
+ case _ => false
+ }
+ }
+
+ override def apply(plan: LogicalPlan): LogicalPlan = {
+ plan.transformWithPruning(
+ _.containsAllPatterns(EXTERNAL_UDF, WINDOW)) {
+ case window: Window
+ if window.windowExpressions.exists(containsExternalUDFOverWindowExpression) =>
+ val windowProjectExprIds = mutable.Set.empty[ExprId]
+ val windowProjectList = mutable.ArrayBuffer.empty[NamedExpression]
+ val externalUdfProjectList = window.windowExpressions.map { expression =>
+ if (containsExternalUDFOverWindowExpression(expression)) {
+ expression.transformDown {
+ case windowExpression: WindowExpression =>
+ val alias = Alias(windowExpression, s"w_${windowProjectList.size}")()
+ windowProjectList += alias
+ alias.toAttribute
+ }.asInstanceOf[NamedExpression]
+ } else {
+ if (!windowProjectExprIds.contains(expression.exprId)) {
+ windowProjectList += expression
+ windowProjectExprIds += expression.exprId
+ }
+ expression.toAttribute
+ }
+ }
+ Project(
+ externalUdfProjectList,
Review Comment:
Fixed. The rewritten `Project` now prepends `window.child.output` in its original order, and the regression verifies that both the input column and UDF result are preserved.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PlanExternalUDFs.scala:
##########
@@ -0,0 +1,266 @@
+/*
+ * 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.execution.externalUDF
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.LogKeys.JOIN_CONDITION
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
+import org.apache.spark.sql.catalyst.plans.InnerLike
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern.{AGGREGATE, EXTERNAL_UDF, JOIN}
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Converts each scalar external UDF expression into a separate logical evaluation node.
+ * Join-condition handling mirrors `ExtractPythonUDFFromJoinCondition`.
+ *
+ * TODO(SPARK-55278): Add an external UDF equivalent of `ExtractPythonUDFFromLambda`.
+ * TODO(SPARK-55278): Revisit sharing placement logic with the Python UDF extractors after
+ * external UDF planning semantics stabilize.
+ */
+private[sql] object PlanExternalUDFs
+ extends Rule[LogicalPlan] with Logging with PredicateHelper {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan match {
+ // A correlated subquery is rewritten as a join and revisits this rule later.
+ case subquery: Subquery if subquery.correlated => plan
+ case _ if !conf.getConf(SQLConf.UNIFIED_UDF_EXECUTION_ENABLED) =>
Review Comment:
Addressed with a different lifecycle choice. The planner is no longer retained by `SessionState`, and the unified-execution flag is now a static SQL config. Plan construction and optimizer gating therefore observe the same immutable value. Attempts to change the flag in either direction after session construction fail explicitly with `CANNOT_MODIFY_STATIC_CONFIG`, with focused coverage for both directions.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkOptimizer.scala:
##########
@@ -147,7 +154,7 @@ class SparkOptimizer(
* batch executing the [[ExperimentalMethods]] optimizer rules. This hook can be used to add
* custom optimizer batches to the Spark optimizer.
*
- * Note that 'Extract Python UDFs' batch is an exception and ran after the batches defined here.
+ * Note that 'Extract UDFs' batch is an exception and ran after the batches defined here.
Review Comment:
Fixed. Updated the sentence to use the present tense: “is an exception and runs after”.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2452,6 +2456,7 @@ object PushPredicateThroughNonJoin extends Rule[LogicalPlan] with PredicateHelpe
case _: RebalancePartitions => true
case _: ScriptTransformation => true
case _: Sort => true
+ case _: ExecuteExternalUDF => true
Review Comment:
Fixed. The predicate-pushdown regression now verifies both sides: the child-only predicate moves below `ExecuteExternalUDF`, and the UDF-result predicate remains in the upper `Filter`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55278][SQL] Add scalar external UDF planning support [spark]
Posted by "haiyangsun-db (via GitHub)" <gi...@apache.org>.
haiyangsun-db commented on code in PR #57670:
URL: https://github.com/apache/spark/pull/57670#discussion_r3926983275
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PlanExternalUDFs.scala:
##########
@@ -0,0 +1,266 @@
+/*
+ * 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.execution.externalUDF
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.LogKeys.JOIN_CONDITION
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
+import org.apache.spark.sql.catalyst.plans.InnerLike
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern.{AGGREGATE, EXTERNAL_UDF, JOIN}
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Converts each scalar external UDF expression into a separate logical evaluation node.
+ * Join-condition handling mirrors `ExtractPythonUDFFromJoinCondition`.
+ *
+ * TODO(SPARK-55278): Add an external UDF equivalent of `ExtractPythonUDFFromLambda`.
+ * TODO(SPARK-55278): Revisit sharing placement logic with the Python UDF extractors after
+ * external UDF planning semantics stabilize.
+ */
+private[sql] object PlanExternalUDFs
+ extends Rule[LogicalPlan] with Logging with PredicateHelper {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan match {
+ // A correlated subquery is rewritten as a join and revisits this rule later.
+ case subquery: Subquery if subquery.correlated => plan
+ case _ if !conf.getConf(SQLConf.UNIFIED_UDF_EXECUTION_ENABLED) =>
+ if (plan.containsPattern(EXTERNAL_UDF)) {
+ throw QueryCompilationErrors.externalUDFsDisabledError(
+ SQLConf.UNIFIED_UDF_EXECUTION_ENABLED.key)
+ }
+ plan
+ case _ =>
+ var preparedPlan = extractExternalUDFFromJoinCondition(plan)
+ preparedPlan = extractExternalUDFFromAggregate(preparedPlan)
+ preparedPlan = extractGroupingExternalUDFFromAggregate(preparedPlan)
+ preparedPlan.transformUpWithPruning(_.containsPattern(EXTERNAL_UDF)) {
+ // These nodes already own their external UDF expressions.
+ case udfPlan: ExternalUDF => udfPlan
+ case other => extract(other)
+ }
+ }
+
+ private def hasUnevaluableExternalUDF(expression: Expression, join: Join): Boolean = {
+ expression.exists {
+ case udf: ExternalUserDefinedFunction =>
+ !canEvaluate(udf, join.left) && !canEvaluate(udf, join.right)
+ case _ => false
+ }
+ }
+
+ private def extractExternalUDFFromJoinCondition(plan: LogicalPlan): LogicalPlan = {
+ plan.transformUpWithPruning(_.containsAllPatterns(EXTERNAL_UDF, JOIN)) {
+ case join @ Join(_, _, joinType, Some(condition), _)
+ if hasUnevaluableExternalUDF(condition, join) =>
+ if (!joinType.isInstanceOf[InnerLike]) {
+ // Match `PYTHON_UDF_IN_ON_CLAUSE`: moving a cross-side UDF to a post-join filter
+ // changes the semantics of non-inner joins.
+ throw QueryCompilationErrors.useExternalUDFInJoinConditionUnsupportedError(joinType)
+ }
+
+ val (udfConditions, otherConditions) = splitConjunctivePredicates(condition)
+ .partition(hasUnevaluableExternalUDF(_, join))
+ val newCondition = if (otherConditions.isEmpty) {
+ logWarning(log"The join condition:${MDC(JOIN_CONDITION, condition)} " +
+ log"of the join plan contains external UDFs only, " +
+ log"so it will be moved out and the join plan will become a cross join.")
+ None
+ } else {
+ Some(otherConditions.reduceLeft(And))
+ }
+ Filter(udfConditions.reduceLeft(And), join.copy(condition = newCondition))
+ }
+ }
+
+ private def belongsToAggregate(
+ expression: Expression,
+ groupingExpressions: ExpressionSet): Boolean = {
+ expression.isInstanceOf[AggregateExpression] ||
+ groupingExpressions.contains(expression)
+ }
+
+ private def hasExternalUDFOverAggregate(
+ expression: Expression,
+ groupingExpressions: ExpressionSet): Boolean = {
+ expression.exists {
+ case udf: ExternalUserDefinedFunction =>
+ udf.references.isEmpty || udf.exists(belongsToAggregate(_, groupingExpressions))
+ case _ => false
+ }
+ }
+
+ private def extractExternalUDFFromAggregate(plan: LogicalPlan): LogicalPlan = {
+ plan.transformUpWithPruning(_.containsAllPatterns(EXTERNAL_UDF, AGGREGATE)) {
+ case aggregate: Aggregate =>
+ val groupingExpressions = ExpressionSet(aggregate.groupingExpressions)
+ if (!aggregate.aggregateExpressions.exists(
+ hasExternalUDFOverAggregate(_, groupingExpressions))) {
+ aggregate
+ } else {
+ val projectExpressions = ArrayBuffer.empty[NamedExpression]
+ val aggregateExpressions = ArrayBuffer.empty[NamedExpression]
+ aggregate.aggregateExpressions.foreach { expression =>
+ if (hasExternalUDFOverAggregate(expression, groupingExpressions)) {
+ val newExpression = expression.transformDown {
+ case child: Expression if belongsToAggregate(child, groupingExpressions) =>
+ val alias = child match {
+ case named: NamedExpression => named
+ case other => Alias(other, "agg")()
+ }
+ aggregateExpressions += alias
+ alias.toAttribute
+ }
+ projectExpressions += newExpression.asInstanceOf[NamedExpression]
+ } else {
+ aggregateExpressions += expression
+ projectExpressions += expression.toAttribute
+ }
+ }
+ Project(
+ projectExpressions.toSeq,
+ aggregate.copy(aggregateExpressions = aggregateExpressions.toSeq))
+ }
+ }
+ }
+
+ private def hasExternalUDF(expression: Expression): Boolean = {
+ expression.exists(_.isInstanceOf[ExternalUserDefinedFunction])
+ }
+
+ private def extractGroupingExternalUDFFromAggregate(plan: LogicalPlan): LogicalPlan = {
+ plan.transformUpWithPruning(_.containsAllPatterns(EXTERNAL_UDF, AGGREGATE)) {
+ case aggregate: Aggregate if aggregate.groupingExpressions.exists(hasExternalUDF) =>
+ val projectExpressions = ArrayBuffer.empty[NamedExpression]
+ val groupingExpressions = ArrayBuffer.empty[Expression]
+ val attributeMap = ArrayBuffer.empty[
+ (ExternalUserDefinedFunction, NamedExpression)]
+
+ def mappedAttribute(udf: ExternalUserDefinedFunction): Option[NamedExpression] = {
+ attributeMap.collectFirst {
+ case (candidate, attribute) if sameUDF(candidate, udf) => attribute
+ }
+ }
+
+ aggregate.groupingExpressions.foreach { expression =>
+ if (hasExternalUDF(expression)) {
+ val newExpression = expression.transformDown {
+ case udf: ExternalUserDefinedFunction =>
+ assert(udf.udfDeterministic,
Review Comment:
Fixed. `NondeterministicExpressionCollection` now hoists nondeterministic `ExternalUserDefinedFunction` expressions, and the regression test verifies that the UDF is evaluated below `Aggregate` while the grouping expressions are deterministic.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/ExtractExternalUDFFromWindow.scala:
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.execution.externalUDF
+
+import scala.collection.mutable
+
+import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, Expression, ExprId,
+ ExternalUserDefinedFunction, NamedExpression, WindowExpression}
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Project, Window}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern.{EXTERNAL_UDF, WINDOW}
+
+/**
+ * Extracts external UDFs that are parents of window expressions from a [[Window]] operator.
+ * The window expressions are evaluated by the [[Window]], and [[PlanExternalUDFs]] subsequently
+ * converts the external UDFs in the new [[Project]] into evaluation nodes above it.
+ */
+private[sql] object ExtractExternalUDFFromWindow extends Rule[LogicalPlan] {
+
+ private def containsExternalUDFOverWindowExpression(expression: Expression): Boolean = {
+ expression.exists {
+ case udf: ExternalUserDefinedFunction =>
+ udf.exists(_.isInstanceOf[WindowExpression])
+ case _ => false
+ }
+ }
+
+ override def apply(plan: LogicalPlan): LogicalPlan = {
+ plan.transformWithPruning(
+ _.containsAllPatterns(EXTERNAL_UDF, WINDOW)) {
+ case window: Window
+ if window.windowExpressions.exists(containsExternalUDFOverWindowExpression) =>
+ val windowProjectExprIds = mutable.Set.empty[ExprId]
+ val windowProjectList = mutable.ArrayBuffer.empty[NamedExpression]
+ val externalUdfProjectList = window.windowExpressions.map { expression =>
+ if (containsExternalUDFOverWindowExpression(expression)) {
+ expression.transformDown {
+ case windowExpression: WindowExpression =>
+ val alias = Alias(windowExpression, s"w_${windowProjectList.size}")()
+ windowProjectList += alias
+ alias.toAttribute
+ case attribute: Attribute if !windowProjectExprIds.contains(attribute.exprId) =>
+ windowProjectList += attribute
Review Comment:
Fixed. Direct child attributes are no longer added to `Window.windowExpressions`. The focused regression keeps both UDF arguments and verifies that the Window output ExprIds are unique.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PlanExternalUDFs.scala:
##########
@@ -0,0 +1,266 @@
+/*
+ * 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.execution.externalUDF
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.LogKeys.JOIN_CONDITION
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
+import org.apache.spark.sql.catalyst.plans.InnerLike
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern.{AGGREGATE, EXTERNAL_UDF, JOIN}
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Converts each scalar external UDF expression into a separate logical evaluation node.
+ * Join-condition handling mirrors `ExtractPythonUDFFromJoinCondition`.
+ *
+ * TODO(SPARK-55278): Add an external UDF equivalent of `ExtractPythonUDFFromLambda`.
+ * TODO(SPARK-55278): Revisit sharing placement logic with the Python UDF extractors after
+ * external UDF planning semantics stabilize.
+ */
+private[sql] object PlanExternalUDFs
+ extends Rule[LogicalPlan] with Logging with PredicateHelper {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan match {
+ // A correlated subquery is rewritten as a join and revisits this rule later.
+ case subquery: Subquery if subquery.correlated => plan
+ case _ if !conf.getConf(SQLConf.UNIFIED_UDF_EXECUTION_ENABLED) =>
+ if (plan.containsPattern(EXTERNAL_UDF)) {
+ throw QueryCompilationErrors.externalUDFsDisabledError(
+ SQLConf.UNIFIED_UDF_EXECUTION_ENABLED.key)
+ }
+ plan
+ case _ =>
+ var preparedPlan = extractExternalUDFFromJoinCondition(plan)
+ preparedPlan = extractExternalUDFFromAggregate(preparedPlan)
+ preparedPlan = extractGroupingExternalUDFFromAggregate(preparedPlan)
+ preparedPlan.transformUpWithPruning(_.containsPattern(EXTERNAL_UDF)) {
+ // These nodes already own their external UDF expressions.
+ case udfPlan: ExternalUDF => udfPlan
+ case other => extract(other)
+ }
+ }
+
+ private def hasUnevaluableExternalUDF(expression: Expression, join: Join): Boolean = {
+ expression.exists {
+ case udf: ExternalUserDefinedFunction =>
+ !canEvaluate(udf, join.left) && !canEvaluate(udf, join.right)
+ case _ => false
+ }
+ }
+
+ private def extractExternalUDFFromJoinCondition(plan: LogicalPlan): LogicalPlan = {
+ plan.transformUpWithPruning(_.containsAllPatterns(EXTERNAL_UDF, JOIN)) {
+ case join @ Join(_, _, joinType, Some(condition), _)
+ if hasUnevaluableExternalUDF(condition, join) =>
+ if (!joinType.isInstanceOf[InnerLike]) {
+ // Match `PYTHON_UDF_IN_ON_CLAUSE`: moving a cross-side UDF to a post-join filter
+ // changes the semantics of non-inner joins.
+ throw QueryCompilationErrors.useExternalUDFInJoinConditionUnsupportedError(joinType)
+ }
+
+ val (udfConditions, otherConditions) = splitConjunctivePredicates(condition)
+ .partition(hasUnevaluableExternalUDF(_, join))
+ val newCondition = if (otherConditions.isEmpty) {
+ logWarning(log"The join condition:${MDC(JOIN_CONDITION, condition)} " +
+ log"of the join plan contains external UDFs only, " +
+ log"so it will be moved out and the join plan will become a cross join.")
+ None
+ } else {
+ Some(otherConditions.reduceLeft(And))
Review Comment:
Fixed. Added a mixed join-condition regression that verifies the equality remains on the `Join` while only the external-UDF predicate moves to the upper `Filter`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/ExecuteExternalUDFExec.scala:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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.execution.externalUDF
+
+import org.apache.spark.annotation.Experimental
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{
+ Attribute,
+ AttributeSet,
+ ExternalUserDefinedFunction
+}
+import org.apache.spark.sql.errors.QueryExecutionErrors
+import org.apache.spark.sql.execution.SparkPlan
+import org.apache.spark.udf.worker.UDFWorkerSpecification
+
+/**
+ * :: Experimental ::
+ * Physical plan node that evaluates one scalar UDF in an external worker process.
+ *
+ * @param udf UDF expression evaluated by the worker session.
+ * @param resultAttr Output attribute for the UDF expression.
+ * @param child Child plan providing input rows.
+ */
+@Experimental
+case class ExecuteExternalUDFExec(
+ udf: ExternalUserDefinedFunction,
Review Comment:
Fixed. The Scaladoc now describes this as a scalar-UDF planning node and explicitly says worker execution remains unimplemented.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58467][SQL] Add scalar external UDF planning support [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #57670:
URL: https://github.com/apache/spark/pull/57670#issuecomment-5535460154
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/85494541683d66541ed1357e05b64e9094c870dc
- merged into branch-4.x https://github.com/apache/spark/commit/73f4a975da1b2c3c82dfcbb8ba01d62e438fdac9
*Posted by `merge_spark_pr.py`*
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org