Skip to content

[SPARK-32995][SQL] CostBasedJoinReorder optimizer rule should be idempotent #29871

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
tanelk wants to merge 22 commits into from
Closed

[SPARK-32995][SQL] CostBasedJoinReorder optimizer rule should be idempotent #29871

Show file tree
Hide file tree
Changes from 8 commits
Commits
Show all changes
22 commits
Select commit Hold shift + click to select a range
3628972
CBO once
tanelk Sep 24, 2020
d58966c
make CostBasedJoinReorder idempotent
tanelk Sep 25, 2020
7a44562
Add CostBasedJoinReorder idempotence check to UTs
tanelk Sep 25, 2020
1aec54b
Merge branch 'master' into SPARK-32995
tanelk Sep 25, 2020
c995703
Revert "make CostBasedJoinReorder idempotent"
tanelk Sep 27, 2020
c9028ad
make CostBasedJoinReorder idempotent
tanelk Sep 27, 2020
4619acc
style
tanelk Sep 27, 2020
ee944e9
Add UT
tanelk Sep 29, 2020
d77a024
Merge branch 'master' into SPARK-32995
Oct 7, 2020
8bd1c21
Merge branch 'master' into SPARK-32995_idempotent_cbo
tanelk Dec 30, 2020
624e355
Merge remote-tracking branch 'origin/SPARK-32995' into SPARK-32995
Jan 5, 2021
f208406
Merge branch 'master' into SPARK-32995
Jan 5, 2021
f204acb
Merge branch 'master' into SPARK-32995
Jan 5, 2021
6248e4a
Merge branch 'master' into SPARK-33122
Jan 5, 2021
fe7da48
Merge branch 'master' into SPARK-32995
Jan 5, 2021
09c5da8
Rerun PlanStabilitySuite
Jan 5, 2021
f815feb
Rerun PlanStabilitySuite
Jan 5, 2021
f7744db
Merge branch 'master' into SPARK-32995_idempotent_cbo
tanelk Jan 16, 2021
d09abd1
rerun plan stability
tanelk Jan 26, 2021
5fe5f3b
rerun plan stability
tanelk Jan 26, 2021
d6522d9
Rerun plan stability suite
tanelk Mar 20, 2021
d2a6128
Rerun plan stability suite
tanelk Apr 2, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 6 additions & 4 deletions ...atalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/CostBasedJoinReorder.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -206,13 +206,15 @@ object JoinReorderDP extends PredicateHelper with Logging {
filters: Option[JoinGraphInfo]): JoinPlanMap = {

val nextLevel = new JoinPlanMap
var k = 0
val lev = existingLevels.length - 1
var k = lev

@maropu maropu Sep 28, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

cc: @wzhfy

// Build plans for the next level from plans at level k (one side of the join) and level
// lev - k (the other side of the join).
// For the lower level k, we only need to search from 0 to lev - k, because when building
// For the higher level k, we only need to search from lev to lev - k, because when building
// a join from A and B, both A J B and B J A are handled.
while (k <= lev - k) {
// Start searching from highest level to make sure that optimally ordered input doesn't get
// reordered into another plan with the same cost.

@maropu maropu Sep 28, 2020 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could you describe more about why we need to change this search order strategy for making this rule idempotent? If how to update next-level join plans is a root cause, we cannot simply modify the update logic below when different plans having the same cost found?

spark/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/CostBasedJoinReorder.scala

Lines 227 to 233 in d15f504

buildJoin(oneSidePlan, otherSidePlan, conf, conditions, topOutput, filters) match {
case Some(newJoinPlan) =>
// Check if it's the first plan for the item set, or it's a better plan than
// the existing one due to lower cost.
val existingPlan = nextLevel.get(newJoinPlan.itemIds)
if (existingPlan.isEmpty || newJoinPlan.betterThan(existingPlan.get, conf)) {
nextLevel.update(newJoinPlan.itemIds, newJoinPlan)

@tanelk tanelk Sep 28, 2020 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This was also my first idea, but I couldn't figure it out.

I'll try to give two examples, where all pairwise joins have the same cost and we can join only plans with consecutive letters (AB, but bot AC)
Keep in mind, that we try build left-deep trees:

spark/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/CostBasedJoinReorder.scala

Line 301 in 4619acc

// Put the deeper side on the left, tend to build a left-deep tree.

Let the inital join be ((AB)C) - firstly join A and B and then join with C.
Lets look at the candidates generated at each iteration:

Firstly if we kept the first candidate (current code)

  1. iteration (the initial state):
    (A) (B) (C)

  2. iteration (nested loop over the single elements):
    (AB) (BC)
    Disgarded (BA) and (CB), because they were late

  3. iteration (outer loop is single element, inner is two element):
    ((BC)A)
    Disgarded ((AB)C) because it was late

Secondly if we kept the last one (possible change, that I considered)

  1. iteration (the initial state):
    (A) (B) (C)

  2. iteration (nested loop over the single elements):
    (BA) (CB)
    Disgarded (AB) and (BC), because they were overwritten

  3. iteration (outer loop is single element, inner is two element):
    ((BA)C)
    Disgarded ((CB)A), because it was overwritten

Neither of these gave the original result. But lets try iterating from the largest (while keeping the first candidate). This is my proposition.

  1. iteration (the initial state):
    (A) (B) (C)
    The initial state

  2. iteration (nested loop over the single elements):
    (AB) (BC)
    Disgarded (BA) and (CB), because they were late

  3. iteration (outer loop is two element, inner is single element):
    ((AB)C)
    Disgarded ((BC)A) because it was late

Now, if we would add another plan D, then in the 4. iteration it would be appended to the ((AB)C) and we would still have correct result.

Note that if the input would be ((AB)(CD)), then the result would also be (((AB)C)D), but we must be stable only on second time this is applied and because we output left-deep trees, then we must be stable on left-deep trees, to be idempotent.

Also by left-deep we mean "as left-deep" as possible.

And finally, yes, I know that this dummy example is no rigorous proof, but we have 2 things going for us:

  1. Out of all the test cases we have, none broke this rule
  2. The idempotentsi test is used only in UTs, so there is no chance of breaking the user side by marking this as idempotent.

@tanelk tanelk Sep 28, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I pondered about this a bit more. I think this is a bit more general explanaiton:

After the first round of optimisation the plans are ordered: ABCD....
Lets ignore the brackets to keep it general.
On the second run we just have to redraw the brackets without changing the order.

We know that the result is as left-deep as possible.

So, if we would be iterating from the smaller ones, then we could have (K) and (LM), that we want to join, but because it has to be left-deep, we must order it as ((LM)K).

But, if we start from the bigger ones, then we hit (KL) and (M) first and joining these will not reorder the letters.

For stability I think, that there are at least 4 factors, that must match with each others:
How do we join: left-deep, right-deep....
The iteration order in one "level": left-to-right, right-to-left.
The order of picking levels: bigger first, smaller first.
Which one we keep, when joins have equal cost: first, last...

There might be more. Changing the order of picking levels gets us to stability in one change.
There are most definitely more ways to achieve stability, but it seems, that it would require more than one change.

@maropu maropu Sep 29, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for your detailed explanation. IMO a fix should be simple and intuitive in terms of maintainability. Another idea that I came up with after I read the description above was to simply sort input candidate plans by some values (e.g., semanticHash) at the beginning of search. It seems there are multiple options to solve this (like the four factors you described above). Any reason to choose the current approach out from them? You chose the simplest one?

@maropu maropu Sep 29, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Btw, could you update the PR description based on the explanation above? That explanation looks useful for making others understood smoothly, I think.

@maropu maropu Sep 29, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

To my surprise the semanticHash changes between executions, so sorting by it makes the results of the PlanStabilitySuite not stable.

Ah, I see... This rule rewrites candidate plans after reordering plans, so IIUC semanticHash values might change in the second run...

TBH It wasn't my first idea, but it is the first one, I got to work - the other ones seemed to require too much code change and I did not go further with those.

okay, let's wait for the comments of other reviewers.

@tanelk tanelk Sep 29, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah, I see... This rule rewrites candidate plans after reordering plans, so IIUC semanticHash values might change in the second run...

Yeah, that was one issue, that complicated the approach.

But, what made it even worse is that the value for semanticHash will change every time you restart the JVM. I'm not sure if its intended or not and I could not find the source of randomness.

@maropu maropu Sep 29, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

But, what made it even worse is that the value for semanticHash will change every time you restart the JVM. I'm not sure if its intended or not and I could not find the source of randomness.

Yea, that's a design and why do we need to keep the same semanticHash values between different JVMs?

@tanelk tanelk Sep 29, 2020 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yea, that's a design and why do we need to keep the same semanticHash values between different JVMs?

It does not allow us sort by it and then compare results in the PlanStabilitySuite.

@maropu maropu Sep 29, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah, I see. I only cared about the idempotence check in RuleExecutor.

while (k >= lev - k) {
val oneSideCandidates = existingLevels(k).values.toSeq
for (i <- oneSideCandidates.indices) {
val oneSidePlan = oneSideCandidates(i)
Expand All @@ -236,7 +238,7 @@ object JoinReorderDP extends PredicateHelper with Logging {
}
}
}
k += 1
k -= 1
}
nextLevel
}
Expand Down
4 changes: 1 addition & 3 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -188,9 +188,7 @@ abstract class Optimizer(catalogManager: CatalogManager)
// plan may contain nodes that do not report stats. Anything that uses stats must run after
// this batch.
Batch("Early Filter and Projection Push-Down", Once, earlyScanPushDownRules: _*) :+
// Since join costs in AQP can change between multiple runs, there is no reason that we have an
// idempotence enforcement on this batch. We thus make it FixedPoint(1) instead of Once.
Batch("Join Reorder", FixedPoint(1),
Batch("Join Reorder", Once,
CostBasedJoinReorder) :+
Batch("Eliminate Sorts", Once,
EliminateSorts) :+
Expand Down
49 changes: 48 additions & 1 deletion ...src/test/scala/org/apache/spark/sql/catalyst/optimizer/joinReorder/JoinReorderSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ class JoinReorderSuite extends JoinReorderPlanTestBase with StatsEstimationTestB
PushPredicateThroughJoin,
ColumnPruning,
CollapseProject) ::
Batch("Join Reorder", FixedPoint(1),
Batch("Join Reorder", Once,
CostBasedJoinReorder) :: Nil
}

Expand Down Expand Up @@ -363,4 +363,51 @@ class JoinReorderSuite extends JoinReorderPlanTestBase with StatsEstimationTestB

assertEqualJoinPlans(Optimize, originalPlan3, bestPlan3)
}

test("SPARK-32995: Optimized plan should not get reordered to one with the same cost") {
// Define new columns and tables, so all join orders would have the same total cost
val columnInfo = AttributeMap(Seq(
attr("t1.col1") -> rangeColumnStat(5, 0),
attr("t1.col2") -> rangeColumnStat(5, 0),
attr("t2.col1") -> rangeColumnStat(5, 0),
attr("t2.col2") -> rangeColumnStat(5, 0),
attr("t3.col1") -> rangeColumnStat(5, 0),
attr("t3.col2") -> rangeColumnStat(5, 0)
))

val nameToAttr = columnInfo.map(kv => kv._1.name -> kv._1)
val nameToColInfo = columnInfo.map(kv => kv._1.name -> kv)

val t1 = StatsTestPlan(
outputList = Seq("t1.col1", "t1.col2").map(nameToAttr),
rowCount = 1000,
size = Some(1000),
attributeStats = AttributeMap(Seq("t1.col1", "t1.col2").map(nameToColInfo)))
val t2 = StatsTestPlan(
outputList = Seq("t2.col1", "t2.col2").map(nameToAttr),
rowCount = 1000,
size = Some(1000),
attributeStats = AttributeMap(Seq("t2.col1", "t2.col2").map(nameToColInfo)))
val t3 = StatsTestPlan(
outputList = Seq("t3.col1", "t3.col2").map(nameToAttr),
rowCount = 1000,
size = Some(1000),
attributeStats = AttributeMap(Seq("t3.col1", "t3.col2").map(nameToColInfo)))

// Right-deep
val originalPlan = t1.join(
t2.join(t3, Inner, Some(nameToAttr("t2.col2") === nameToAttr("t3.col2"))),
Inner, Some(nameToAttr("t1.col1") === nameToAttr("t2.col1")))

// Left-deep with the plans in the same order
val expectedPlan =
t1.join(t2, Inner, Some(nameToAttr("t1.col1") === nameToAttr("t2.col1")))
.join(t3, Inner, Some(nameToAttr("t2.col2") === nameToAttr("t3.col2")))

// Right-deep gets turned into a left-deep
assertEqualJoinPlans(Optimize, originalPlan, expectedPlan)

// Left-deep is not changed (the rule is idempotent)
assertEqualJoinPlans(Optimize, expectedPlan, expectedPlan)
}
}
10 changes: 5 additions & 5 deletions ...a/org/apache/spark/sql/catalyst/optimizer/joinReorder/StarJoinCostBasedReorderSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ class StarJoinCostBasedReorderSuite extends JoinReorderPlanTestBase with StatsEs
PushPredicateThroughJoin,
ColumnPruning,
CollapseProject) ::
Batch("Join Reorder", FixedPoint(1),
Batch("Join Reorder", Once,
CostBasedJoinReorder) :: Nil
}

Expand Down Expand Up @@ -345,8 +345,8 @@ class StarJoinCostBasedReorderSuite extends JoinReorderPlanTestBase with StatsEs

val expected =
f1.join(d3, Inner, Some(nameToAttr("f1_fk3") === nameToAttr("d3_pk")))
.join(d2, Inner, Some(nameToAttr("f1_fk2") === nameToAttr("d2_pk")))
.join(d1, Inner, Some(nameToAttr("f1_fk1") === nameToAttr("d1_pk")))
.join(d2, Inner, Some(nameToAttr("f1_fk2") === nameToAttr("d2_pk")))
.join(t4.join(t3, Inner, Some(nameToAttr("t3_c2") === nameToAttr("t4_c2"))), Inner,
Some(nameToAttr("d1_c2") === nameToAttr("t3_c1")))
.join(t2.join(t1, Inner, Some(nameToAttr("t1_c2") === nameToAttr("t2_c2"))), Inner,
Expand Down Expand Up @@ -380,8 +380,8 @@ class StarJoinCostBasedReorderSuite extends JoinReorderPlanTestBase with StatsEs

val expected =
f1.join(d3, Inner, Some(nameToAttr("f1_fk3") === nameToAttr("d3_pk")))
.join(d2, Inner, Some(nameToAttr("f1_fk2") === nameToAttr("d2_pk")))
.join(d1, Inner, Some(nameToAttr("f1_fk1") === nameToAttr("d1_pk")))
.join(d2, Inner, Some(nameToAttr("f1_fk2") === nameToAttr("d2_pk")))
.select(outputsOf(d1, d2, f1, d3): _*)

assertEqualJoinPlans(Optimize, query, expected)
Expand All @@ -405,8 +405,8 @@ class StarJoinCostBasedReorderSuite extends JoinReorderPlanTestBase with StatsEs
(nameToAttr("f1_fk3") === nameToAttr("t3_c1")))

val expected =
f1.join(t3, Inner, Some(nameToAttr("f1_fk3") === nameToAttr("t3_c1")))
.join(t2, Inner, Some(nameToAttr("f1_fk2") === nameToAttr("t2_c1")))
f1.join(t2, Inner, Some(nameToAttr("f1_fk2") === nameToAttr("t2_c1")))
.join(t3, Inner, Some(nameToAttr("f1_fk3") === nameToAttr("t3_c1")))
.join(t1, Inner, Some(nameToAttr("f1_fk1") === nameToAttr("t1_c1")))
.select(outputsOf(t1, f1, t2, t3): _*)

Expand Down
Loading