[SPARK-19712][SQL] Pushdown LeftSemi/LeftAnti below join #24331

dilipbiswal · 2019-04-09T21:34:00Z

What changes were proposed in this pull request?

This PR adds support for pushing down LeftSemi and LeftAnti joins below the Join operator.
This is a prerequisite work thats needed for the subsequent task of moving the subquery rewrites to the beginning of optimization phase.

The larger PR is here . This PR addresses the comment at link.

How was this patch tested?

Added tests under LeftSemiAntiJoinPushDownSuite.

SparkQA · 2019-04-09T21:37:06Z

Test build #104453 has finished for PR 24331 at commit a5f1f50.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2019-04-10T02:09:10Z

Test build #104456 has finished for PR 24331 at commit 14ed943.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

dilipbiswal · 2019-04-10T02:47:21Z

cc @cloud-fan

dilipbiswal · 2019-04-10T07:21:10Z

...talyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushDownLeftSemiAntiJoin.scala

+        case pushdownDirection.toLeftBranch
+          if (belowJoinType == LeftOuter || belowJoinType.isInstanceOf[InnerLike]) =>
+          // push down leftsemi/anti join to the left table
+          val newLeft = Join(gLeft, right, joinType, newJoinCond, parentHint)


@cloud-fan I have a question here. Will it be safe to propagate the hints here ? I am inclined to only do this optimization if no join hints are specified in either parent and child joins. Currently i am propagating them as is but thinking of changing it. Wanted to check your opinion before i made the change.

cloud-fan · 2019-04-11T12:05:58Z

...talyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushDownLeftSemiAntiJoin.scala

+   * Define an enumeration to identify whether a LeftSemi/LeftAnti join can be pushed down to
+   * the left leg or the right leg of the join.
+   */
+  object pushdownDirection extends Enumeration {


shall we upper case the first letter?

@cloud-fan Sure.

cloud-fan · 2019-04-11T12:11:00Z

...talyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushDownLeftSemiAntiJoin.scala

+
+  def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+    // push LeftSemi/LeftAnti down into the join below
+    case j @ Join(left @ Join(gLeft, gRight, AllowedJoinTypes(_), belowJoinCond, childHint),


this is hard to read when 2 joins are extracted together. How about

case j: Join(AllowedJoin(left), right, ...) =>

@cloud-fan Good idea wenchen.

cloud-fan · 2019-04-11T12:11:29Z

...talyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushDownLeftSemiAntiJoin.scala

+          val newRight = Join(gRight, right, joinType, newJoinCond, parentHint)
+          Join(gLeft, newRight, belowJoinType, belowJoinCond, childHint)
+        case _ =>
+          // Do nothing


when can this happen?

@cloud-fan When we decide that we can't pushdown the parent join. For example this test should exercise the default case.

cloud-fan · 2019-04-11T12:14:33Z

...talyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushDownLeftSemiAntiJoin.scala

+        // leftsemi/anti join then push the leftsemi/anti join to the right side.
+        (pushdownDirection.toRightBranch, rightConditions.reduceLeftOption(And))
+      } else {
+        noPushdown


PushPredicateThroughJoin may pushdown to both sides, do we have such a case here?

@cloud-fan To the best of my knowledge, we don't have this case. I actually tried to get a subquery to push down to both legs of the join but couldn't. Normal filter conditions can trigger pushing down to both legs currently though.

can we add a TODO and say we will revisit it?

@cloud-fan Sure. I will add a TODO

SparkQA · 2019-04-12T03:58:16Z

Test build #104530 has finished for PR 24331 at commit 813075f.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2019-04-15T15:03:43Z

...talyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushDownLeftSemiAntiJoin.scala

+   * the left leg or the right leg of the join.
+   */
+  object PushdownDirection extends Enumeration {
+    val toRightBranch, toLeftBranch, none = Value


upper-case the first letter for them too. I think this is the policy when defining enums.

actually, the policy is upper-case all the letters, e.g.

object PushdownDirection extends Enumeration { val TO_RIGHT_BRANCH, ... }

@cloud-fan Will change.

cloud-fan · 2019-04-15T15:05:34Z

...talyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushDownLeftSemiAntiJoin.scala

+    val toRightBranch, toLeftBranch, none = Value
+  }
+
+  object AllowedJoins {


nit: it matches a single join, so no s at the end.

@cloud-fan ok.. will make a change.

cloud-fan · 2019-04-15T16:57:04Z

...talyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushDownLeftSemiAntiJoin.scala

+    } else {
+      /**
+       * When the join condition is empty,
+       * 1) if this is a left outer join or inner join, push leftsemi/anti join down


why can't we push to both legs if it's inner join?

@cloud-fan Perhaps its possible. In this PR, i was focusing on what is happening today in PushPredicateThroughJoin and keep the behaviour same. We can look into improving both this rule and PushPredicateThroughJoin as follow-up. The reason i say it is, probably we need to test more and prove that pushdown to both sides don't create any side effects or can cause wrong results ?

cloud-fan · 2019-04-15T17:00:01Z

...talyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushDownLeftSemiAntiJoin.scala

+        // When the join conditions can be computed based on the left leg of
+        // leftsemi/anti join then push the leftsemi/anti join to the left side.
+        (PushdownDirection.TO_LEFT_BRANCH, leftConditions.reduceLeftOption(And))
+      } else if (leftConditions.isEmpty && rightConditions.nonEmpty && commonConditions.isEmpty) {


what if commonConditions is not empty? can we add a filter at top?

@cloud-fan Please refer to my answer above.

SparkQA · 2019-04-15T20:12:57Z

Test build #104593 has finished for PR 24331 at commit 9945c28.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2019-04-16T07:05:02Z

Test build #104612 has finished for PR 24331 at commit 4a5b4ca.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

dilipbiswal · 2019-04-16T07:07:57Z

retest this please

cloud-fan · 2019-04-16T11:24:59Z

...talyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushDownLeftSemiAntiJoin.scala

+  /**
+   * Determine which side of the join a LeftSemi/LeftAnti join can be pushed to.
+   */
+  private def pushTo(leftChild: Join, rightChild: LogicalPlan, joinCond: Option[Expression]) = {


since we can only push down left semi/anti join to either left or right side, there is no need to return the new join condition in this method, because the join condition won't change.

cloud-fan · 2019-04-16T11:26:41Z

...t/src/test/scala/org/apache/spark/sql/catalyst/optimizer/LeftSemiAntiJoinPushDownSuite.scala

+  Seq(LeftSemi, LeftAnti).foreach { case outerJT =>
+    Seq(Inner, LeftOuter, Cross).foreach { case innerJT =>
+      test(s"$outerJT pushdown with empty join condition join type $innerJT") {
+        val joinedRelation = testRelation1.join(testRelation2, joinType = innerJT, None)


the inner join can have a join condition, right?

If it supports, we should add Seq(Some('d === 'e), None).foreach at the begining too

cloud-fan · 2019-04-16T11:30:04Z

...t/src/test/scala/org/apache/spark/sql/catalyst/optimizer/LeftSemiAntiJoinPushDownSuite.scala

+  }
+
+  Seq(LeftSemi, LeftAnti).foreach { case jt =>
+      test(s"$jt pushdown with empty join condition join type RightOuter") {


can we merge the test case with above? something like

val correctAnswer = if (innerJT = RightOuter) { ... } else { ... }

SparkQA · 2019-04-16T11:31:59Z

Test build #104619 has finished for PR 24331 at commit 4a5b4ca.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2019-04-16T20:44:05Z

Test build #104627 has finished for PR 24331 at commit fe3e168.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2019-04-17T03:40:34Z

...talyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushDownLeftSemiAntiJoin.scala

+    val rightOutput = rightChild.outputSet
+
+    if (joinCond.nonEmpty) {
+      val noPushdown = PushdownDirection.NONE


We can use PushdownDirection.NONE directly.

cloud-fan · 2019-04-17T03:40:58Z

...talyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushDownLeftSemiAntiJoin.scala

+      if (rest.isEmpty && leftConditions.nonEmpty) {
+        // When the join conditions can be computed based on the left leg of
+        // leftsemi/anti join then push the leftsemi/anti join to the left side.
+        (PushdownDirection.TO_LEFT_BRANCH)


no need to add (...)

cloud-fan · 2019-04-17T03:41:03Z

...talyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushDownLeftSemiAntiJoin.scala

+      } else if (leftConditions.isEmpty && rightConditions.nonEmpty && commonConditions.isEmpty) {
+        // When the join conditions can be computed based on the attributes from right leg of
+        // leftsemi/anti join then push the leftsemi/anti join to the right side.
+        (PushdownDirection.TO_RIGHT_BRANCH)


cloud-fan · 2019-04-17T03:43:56Z

...t/src/test/scala/org/apache/spark/sql/catalyst/optimizer/LeftSemiAntiJoinPushDownSuite.scala

+
+  Seq(Some('d === 'e), None).foreach { case innerJoinCond =>
+    Seq(LeftSemi, LeftAnti).foreach { case outerJT =>
+      Seq(Inner, LeftOuter, Cross).foreach { case innerJT =>


These test cases almost cover the above one, except the RightOuter join type. Can we merge them.

nvm, I was wrong

SparkQA · 2019-04-17T07:05:02Z

Test build #104642 has finished for PR 24331 at commit a5ed3d9.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

dilipbiswal · 2019-04-17T07:24:29Z

retest this please

SparkQA · 2019-04-17T11:39:43Z

Test build #104650 has finished for PR 24331 at commit a5ed3d9.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2019-04-17T12:30:34Z

thanks, merging to master!

dilipbiswal · 2019-04-17T15:15:09Z

@cloud-fan Thank you very much.

[SPARK-19712] Pushdown LeftSemi/LeftAnti through join

a5f1f50

style

14ed943

dilipbiswal commented Apr 10, 2019

View reviewed changes

cloud-fan reviewed Apr 11, 2019

View reviewed changes

Code review

813075f

cloud-fan reviewed Apr 15, 2019

View reviewed changes

Code review

9945c28

cloud-fan reviewed Apr 15, 2019

View reviewed changes

Code review - add todo

4a5b4ca

cloud-fan reviewed Apr 16, 2019

View reviewed changes

Code review

fe3e168

cloud-fan reviewed Apr 17, 2019

View reviewed changes

Code review

a5ed3d9

cloud-fan approved these changes Apr 17, 2019

View reviewed changes

cloud-fan closed this in e1c90d6 Apr 17, 2019

[SPARK-19712][SQL] Pushdown LeftSemi/LeftAnti below join #24331

[SPARK-19712][SQL] Pushdown LeftSemi/LeftAnti below join #24331

Uh oh!

Conversation

dilipbiswal commented Apr 9, 2019

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

SparkQA commented Apr 9, 2019

Uh oh!

SparkQA commented Apr 10, 2019

Uh oh!

dilipbiswal commented Apr 10, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dilipbiswal Apr 11, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Apr 12, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Apr 15, 2019

Uh oh!

SparkQA commented Apr 16, 2019

Uh oh!

dilipbiswal commented Apr 16, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Apr 16, 2019

Uh oh!

SparkQA commented Apr 16, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dilipbiswal Apr 11, 2019 •

edited

Loading