From 2ca57641400dfcb3a2cdddc5232f496bcda7183b Mon Sep 17 00:00:00 2001 From: Dilip Biswal Date: Fri, 26 Jul 2019 15:07:01 -0700 Subject: [PATCH 1/4] fix1 --- .../sql/catalyst/optimizer/subquery.scala | 20 ++++-- .../PullupCorrelatedPredicatesSuite.scala | 62 ++++++++++++++++--- 2 files changed, 71 insertions(+), 11 deletions(-) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala index 4f7333c3875f5..fa7be7ccec5d7 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala @@ -276,13 +276,25 @@ object PullupCorrelatedPredicates extends Rule[LogicalPlan] with PredicateHelper plan transformExpressions { case ScalarSubquery(sub, children, exprId) if children.nonEmpty => val (newPlan, newCond) = pullOutCorrelatedPredicates(sub, outerPlans) - ScalarSubquery(newPlan, newCond, exprId) + if (newCond.isEmpty) { + ScalarSubquery(newPlan, children, exprId) + } else { + ScalarSubquery(newPlan, newCond, exprId) + } case Exists(sub, children, exprId) if children.nonEmpty => val (newPlan, newCond) = pullOutCorrelatedPredicates(sub, outerPlans) - Exists(newPlan, newCond, exprId) - case ListQuery(sub, _, exprId, childOutputs) => + if (newCond.isEmpty) { + Exists(newPlan, children, exprId) + } else { + Exists(newPlan, newCond, exprId) + } + case ListQuery(sub, children, exprId, childOutputs) if children.nonEmpty => val (newPlan, newCond) = pullOutCorrelatedPredicates(sub, outerPlans) - ListQuery(newPlan, newCond, exprId, childOutputs) + if (newCond.isEmpty) { + ListQuery(newPlan, children, exprId, childOutputs) + } else { + ListQuery(newPlan, newCond, exprId, childOutputs) + } } } diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/PullupCorrelatedPredicatesSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/PullupCorrelatedPredicatesSuite.scala index 960162a5bfe44..4b1a8ac8688f0 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/PullupCorrelatedPredicatesSuite.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/PullupCorrelatedPredicatesSuite.scala @@ -19,9 +19,9 @@ package org.apache.spark.sql.catalyst.optimizer import org.apache.spark.sql.catalyst.dsl.expressions._ import org.apache.spark.sql.catalyst.dsl.plans._ -import org.apache.spark.sql.catalyst.expressions.{InSubquery, ListQuery} +import org.apache.spark.sql.catalyst.expressions._ import org.apache.spark.sql.catalyst.plans.PlanTest -import org.apache.spark.sql.catalyst.plans.logical.{LocalRelation, LogicalPlan} +import org.apache.spark.sql.catalyst.plans.logical.{Filter, LocalRelation, LogicalPlan} import org.apache.spark.sql.catalyst.rules.RuleExecutor class PullupCorrelatedPredicatesSuite extends PlanTest { @@ -38,17 +38,65 @@ class PullupCorrelatedPredicatesSuite extends PlanTest { val testRelation2 = LocalRelation('c.int, 'd.double) test("PullupCorrelatedPredicates should not produce unresolved plan") { - val correlatedSubquery = + val subPlan = testRelation2 .where('b < 'd) .select('c) - val outerQuery = + val inSubquery = testRelation - .where(InSubquery(Seq('a), ListQuery(correlatedSubquery))) + .where(InSubquery(Seq('a), ListQuery(subPlan))) .select('a).analyze - assert(outerQuery.resolved) + assert(inSubquery.resolved) - val optimized = Optimize.execute(outerQuery) + val optimized = Optimize.execute(inSubquery) assert(optimized.resolved) } + + test("SPARK-28375 PullupCorrelatedPredicates in correlated subquery idiompotency check") { + val subPlan = + testRelation2 + .where('b < 'd) + .select('c) + val inSubquery = + testRelation + .where(InSubquery(Seq('a), ListQuery(subPlan))) + .select('a).analyze + assert(inSubquery.resolved) + + val optimized = Optimize.execute(inSubquery) + val doubleOptimized = Optimize.execute(optimized) + comparePlans(optimized, doubleOptimized) + } + + test("SPARK-28375 PullupCorrelatedPredicates exists correlated subquery idiompotency check") { + val subPlan = + testRelation2 + .where('b === 'd && 'd === 1) + .select(Literal(1)) + val existsSubquery = + testRelation + .where(Exists(subPlan)) + .select('a).analyze + assert(existsSubquery.resolved) + + val optimized = Optimize.execute(existsSubquery) + val doubleOptimized = Optimize.execute(optimized) + comparePlans(optimized, doubleOptimized) + } + + test("SPARK-28375 PullupCorrelatedPredicates scalar correlated subquery idiompotency check") { + val subPlan = + testRelation2 + .where('b === 'd && 'd === 1) + .select(max('d)) + val scalarSubquery = + testRelation + .where(ScalarSubquery(subPlan)) + .select('a).analyze + assert(scalarSubquery.resolved) + + val optimized = Optimize.execute(scalarSubquery) + val doubleOptimized = Optimize.execute(optimized) + comparePlans(optimized, doubleOptimized, false) + } } From 19a38c59b1598fe0d42e3d109cd9f20ded88576c Mon Sep 17 00:00:00 2001 From: Dilip Biswal Date: Sat, 27 Jul 2019 14:47:43 -0700 Subject: [PATCH 2/4] fix --- .../sql/catalyst/optimizer/subquery.scala | 22 ++++++------------- .../PullupCorrelatedPredicatesSuite.scala | 6 ++--- 2 files changed, 10 insertions(+), 18 deletions(-) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala index fa7be7ccec5d7..06bb3ee8e404f 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala @@ -273,28 +273,20 @@ object PullupCorrelatedPredicates extends Rule[LogicalPlan] with PredicateHelper } private def rewriteSubQueries(plan: LogicalPlan, outerPlans: Seq[LogicalPlan]): LogicalPlan = { + def getJoinCondition(newCond: Seq[Expression], oldCond: Seq[Expression]): Seq[Expression] = { + if (newCond.isEmpty) oldCond else newCond + } + plan transformExpressions { case ScalarSubquery(sub, children, exprId) if children.nonEmpty => val (newPlan, newCond) = pullOutCorrelatedPredicates(sub, outerPlans) - if (newCond.isEmpty) { - ScalarSubquery(newPlan, children, exprId) - } else { - ScalarSubquery(newPlan, newCond, exprId) - } + ScalarSubquery(newPlan, getJoinCondition(newCond, children), exprId) case Exists(sub, children, exprId) if children.nonEmpty => val (newPlan, newCond) = pullOutCorrelatedPredicates(sub, outerPlans) - if (newCond.isEmpty) { - Exists(newPlan, children, exprId) - } else { - Exists(newPlan, newCond, exprId) - } + Exists(newPlan, getJoinCondition(newCond, children), exprId) case ListQuery(sub, children, exprId, childOutputs) if children.nonEmpty => val (newPlan, newCond) = pullOutCorrelatedPredicates(sub, outerPlans) - if (newCond.isEmpty) { - ListQuery(newPlan, children, exprId, childOutputs) - } else { - ListQuery(newPlan, newCond, exprId, childOutputs) - } + ListQuery(newPlan, getJoinCondition(newCond, children), exprId, childOutputs) } } diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/PullupCorrelatedPredicatesSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/PullupCorrelatedPredicatesSuite.scala index 4b1a8ac8688f0..9186f9eec336f 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/PullupCorrelatedPredicatesSuite.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/PullupCorrelatedPredicatesSuite.scala @@ -52,7 +52,7 @@ class PullupCorrelatedPredicatesSuite extends PlanTest { assert(optimized.resolved) } - test("SPARK-28375 PullupCorrelatedPredicates in correlated subquery idiompotency check") { + test("PullupCorrelatedPredicates in correlated subquery idiompotency check") { val subPlan = testRelation2 .where('b < 'd) @@ -68,7 +68,7 @@ class PullupCorrelatedPredicatesSuite extends PlanTest { comparePlans(optimized, doubleOptimized) } - test("SPARK-28375 PullupCorrelatedPredicates exists correlated subquery idiompotency check") { + test("PullupCorrelatedPredicates exists correlated subquery idiompotency check") { val subPlan = testRelation2 .where('b === 'd && 'd === 1) @@ -84,7 +84,7 @@ class PullupCorrelatedPredicatesSuite extends PlanTest { comparePlans(optimized, doubleOptimized) } - test("SPARK-28375 PullupCorrelatedPredicates scalar correlated subquery idiompotency check") { + test("PullupCorrelatedPredicates scalar correlated subquery idiompotency check") { val subPlan = testRelation2 .where('b === 'd && 'd === 1) From 5a19309a6ba2d8b844e8cf8ed3a299a7f4ddf390 Mon Sep 17 00:00:00 2001 From: Dilip Biswal Date: Sun, 28 Jul 2019 16:23:01 -0700 Subject: [PATCH 3/4] remove pullupCorrelatedPredicates from idiompotency backlist --- .../org/apache/spark/sql/catalyst/optimizer/Optimizer.scala | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala index 670fc92cb772d..346b2e6184c29 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala @@ -48,9 +48,7 @@ abstract class Optimizer(sessionCatalog: SessionCatalog) } override protected val blacklistedOnceBatches: Set[String] = - Set("Pullup Correlated Expressions", - "Extract Python UDFs" - ) + Set("Extract Python UDFs") protected def fixedPoint = FixedPoint(SQLConf.get.optimizerMaxIterations) From 3b224f094f50a5be61d77868ed9f6ee3e0202417 Mon Sep 17 00:00:00 2001 From: Dilip Biswal Date: Sun, 28 Jul 2019 22:33:43 -0700 Subject: [PATCH 4/4] Review comments --- .../apache/spark/sql/catalyst/optimizer/subquery.scala | 8 ++++++++ .../optimizer/PullupCorrelatedPredicatesSuite.scala | 6 +++--- 2 files changed, 11 insertions(+), 3 deletions(-) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala index 06bb3ee8e404f..32dbd389afd93 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala @@ -273,6 +273,14 @@ object PullupCorrelatedPredicates extends Rule[LogicalPlan] with PredicateHelper } private def rewriteSubQueries(plan: LogicalPlan, outerPlans: Seq[LogicalPlan]): LogicalPlan = { + /** + * This function is used as a aid to enforce idempotency of pullUpCorrelatedPredicate rule. + * In the first call to rewriteSubqueries, all the outer references from the subplan are + * pulled up and join predicates are recorded as children of the enclosing subquery expression. + * The subsequent call to rewriteSubqueries would simply re-records the `children` which would + * contains the pulled up correlated predicates (from the previous call) in the enclosing + * subquery expression. + */ def getJoinCondition(newCond: Seq[Expression], oldCond: Seq[Expression]): Seq[Expression] = { if (newCond.isEmpty) oldCond else newCond } diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/PullupCorrelatedPredicatesSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/PullupCorrelatedPredicatesSuite.scala index 9186f9eec336f..2d86d5a97e769 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/PullupCorrelatedPredicatesSuite.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/PullupCorrelatedPredicatesSuite.scala @@ -52,7 +52,7 @@ class PullupCorrelatedPredicatesSuite extends PlanTest { assert(optimized.resolved) } - test("PullupCorrelatedPredicates in correlated subquery idiompotency check") { + test("PullupCorrelatedPredicates in correlated subquery idempotency check") { val subPlan = testRelation2 .where('b < 'd) @@ -68,7 +68,7 @@ class PullupCorrelatedPredicatesSuite extends PlanTest { comparePlans(optimized, doubleOptimized) } - test("PullupCorrelatedPredicates exists correlated subquery idiompotency check") { + test("PullupCorrelatedPredicates exists correlated subquery idempotency check") { val subPlan = testRelation2 .where('b === 'd && 'd === 1) @@ -84,7 +84,7 @@ class PullupCorrelatedPredicatesSuite extends PlanTest { comparePlans(optimized, doubleOptimized) } - test("PullupCorrelatedPredicates scalar correlated subquery idiompotency check") { + test("PullupCorrelatedPredicates scalar correlated subquery idempotency check") { val subPlan = testRelation2 .where('b === 'd && 'd === 1)