(spark) branch master updated: [SPARK-49646][SQL] add spark config for fixing subquery decorrelation for union/set operations when parentOuterReferences has references not covered in collectedChildOuterReferences

[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 6bde48c3f891 [SPARK-49646][SQL] add spark config for fixing subquery decorrelation for union/set operations when parentOuterReferences has references not covered in collectedChildOuterReferences
6bde48c3f891 is described below

commit 6bde48c3f891263c1bd7518fb9ea5fb737d404fb
Author: Avery Qi <av...@gmail.com>
AuthorDate: Wed Jan 22 22:34:24 2025 +0800

    [SPARK-49646][SQL] add spark config for fixing subquery decorrelation for union/set operations when parentOuterReferences has references not covered in collectedChildOuterReferences
    
    ### What changes were proposed in this pull request?
    
    Spark config added for this change: https://github.com/apache/spark/pull/48109
    
    ### Why are the changes needed?
    
    For safer backports
    
    ### Does this PR introduce _any_ user-facing change?
    
    yes, adds a user-facing config `spark.sql.optimizer.decorrelateUnionOrSetOpUnderLimit.enabled`. Set it to true will enable decorrelating subqueries having correlated references under Union/Set operators which are under Limit operators. It is by default true, setting it to false make spark reverting to incorrect legacy behavior which raises exceptions when decorrelate the above query patterns.
    
    ### How was this patch tested?
    
    N/A
    
    ### Was this patch authored or co-authored using generative AI tooling?
    
    no
    
    Closes #49536 from AveryQi115/SPARK-49646-2.
    
    Lead-authored-by: Avery Qi <av...@gmail.com>
    Co-authored-by: Avery <av...@gmail.com>
    Signed-off-by: Wenchen Fan <we...@databricks.com>
---
 .../spark/sql/catalyst/optimizer/DecorrelateInnerQuery.scala   |  9 ++++++++-
 .../src/main/scala/org/apache/spark/sql/internal/SQLConf.scala | 10 ++++++++++
 2 files changed, 18 insertions(+), 1 deletion(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/DecorrelateInnerQuery.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/DecorrelateInnerQuery.scala
index 2b97b2621b5b..47cee2e789c7 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/DecorrelateInnerQuery.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/DecorrelateInnerQuery.scala
@@ -1064,7 +1064,14 @@ object DecorrelateInnerQuery extends PredicateHelper {
                 // Project, they could get added at the beginning or the end of the output columns
                 // depending on the child plan.
                 // The inner expressions for the domain are the values of newOuterReferenceMap.
-                val domainProjections = newOuterReferences.map(newOuterReferenceMap(_))
+                val domainProjections =
+                  if (SQLConf.get.getConf(
+                    SQLConf.DECORRELATE_UNION_OR_SET_OP_UNDER_LIMIT_ENABLED
+                  )) {
+                    newOuterReferences.map(newOuterReferenceMap(_))
+                  } else {
+                    collectedChildOuterReferences.map(newOuterReferenceMap(_))
+                  }
                 val newChild = Project(child.output ++ domainProjections, decorrelatedChild)
                 (newChild, newJoinCond, newOuterReferenceMap)
               }
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index 066006ce7082..4907e7ee6276 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -3998,6 +3998,16 @@ object SQLConf {
       .booleanConf
       .createWithDefault(true)
 
+  val DECORRELATE_UNION_OR_SET_OP_UNDER_LIMIT_ENABLED =
+    buildConf("spark.sql.optimizer.decorrelateUnionOrSetOpUnderLimit.enabled")
+      .internal()
+      .doc("Decorrelate UNION or SET operation under LIMIT operator. If not enabled," +
+        "revert to legacy incorrect behavior for certain subqueries with correlation under" +
+        "UNION/SET operator with a LIMIT operator above it.")
+      .version("4.0.0")
+      .booleanConf
+      .createWithDefault(true)
+
   val DECORRELATE_EXISTS_IN_SUBQUERY_LEGACY_INCORRECT_COUNT_HANDLING_ENABLED =
     buildConf("spark.sql.optimizer.decorrelateExistsSubqueryLegacyIncorrectCountHandling.enabled")
       .internal()


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