Skip to content

[SPARK-32220][SQL]SHUFFLE_REPLICATE_NL Hint should not change Non-Cartesian Product join result #29035

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
AngersZhuuuu wants to merge 6 commits into from
Closed

[SPARK-32220][SQL]SHUFFLE_REPLICATE_NL Hint should not change Non-Cartesian Product join result #29035

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
5f1269f
Update JoinHintSuite.scala
AngersZhuuuu Jul 8, 2020
27b41e5
[SPARK-32220][SQL]SHUFFLE_REPLICATE_NL Hint should not change Non-Car…
AngersZhuuuu Jul 8, 2020
aeccc25
Update JoinHintSuite.scala
AngersZhuuuu Jul 8, 2020
36ff4ca
follow comment
AngersZhuuuu Jul 10, 2020
a65d332
Update JoinHintSuite.scala
AngersZhuuuu Jul 10, 2020
d9b4dcd
follow comment
AngersZhuuuu Jul 10, 2020
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
4 changes: 4 additions & 0 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/HintErrorLogger.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,4 +53,8 @@ object HintErrorLogger extends HintErrorHandler with Logging {
}
s"$name${prettyParameters.mkString("(", ", ", ")")}"
}

override def wrongHintForNonCartesianProductJoin(hint: HintInfo): Unit = {
logWarning(s"Hint $hint is wrong for Non Cartesian product Join")
}
}
16 changes: 16 additions & 0 deletions ...talyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/EliminateResolvedHint.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,22 @@ object EliminateResolvedHint extends Rule[LogicalPlan] {
case h: ResolvedHint =>
hintErrorHandler.joinNotFoundForJoinHint(h.hints)
h.child
case join: Join if join.condition.isDefined =>
join.copy(
hint = join.hint.copy(
leftHint = removeCartesianProductJoinHint(join.hint.leftHint),
rightHint = removeCartesianProductJoinHint(join.hint.rightHint)))
}
}

def removeCartesianProductJoinHint(hint: Option[HintInfo]): Option[HintInfo] = {

@maropu maropu Jul 9, 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.

I think we should pass a correct condition (leftKeys and rightKeys) into CartesianProductExec instead of removing the hint:

spark/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala

Line 202 in ac6406e

Some(Seq(joins.CartesianProductExec(planLater(left), planLater(right), condition)))

@maropu maropu Jul 9, 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.

It seems the spark strategy incrrectly removes a#0 = a#2;

== Optimized Logical Plan ==
Sort [a#0 ASC NULLS FIRST], true
+- Join Inner, (a#0 = a#2), leftHint=(strategy=shuffle_replicate_nl)
   :- Filter isnotnull(a#0)
   :  +- Relation[a#0,b#1] parquet
   +- Filter isnotnull(a#2)
      +- Relation[a#2,b#3] parquet

== Physical Plan ==
*(3) Sort [a#0 ASC NULLS FIRST], true, 0
+- Exchange rangepartitioning(a#0 ASC NULLS FIRST, 200), true, [id=#87]
   +- CartesianProduct
      :- *(1) Project [a#0, b#1]
      :  +- *(1) Filter isnotnull(a#0)
      :     +- *(1) ColumnarToRow
      :        +- FileScan parquet default.test4[a#0,b#1] Batched: true, DataFilters: ...
      +- *(2) Project [a#2, b#3]
         +- *(2) Filter isnotnull(a#2)
            +- *(2) ColumnarToRow
               +- FileScan parquet default.test5[a#2,b#3] Batched: true, DataFilters: ...

@AngersZhuuuu AngersZhuuuu Jul 10, 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 think we should pass a correct condition (leftKeys and rightKeys) into CartesianProductExec instead of removing the hint:

spark/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala

Line 202 in ac6406e

Some(Seq(joins.CartesianProductExec(planLater(left), planLater(right), condition)))

Yea, in default Cartesian Product Join situation, it didn't need condition at all. So in default, it seems don't have condition when build data.

@AngersZhuuuu AngersZhuuuu Jul 10, 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.

@maropu See latest change, it's ok to do like this?

@AngersZhuuuu AngersZhuuuu Jul 10, 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.

Since we don't know the join keys's condition is EqualTo or EaultNullSafe so it's better just not remove it in ExtractEqualJoinKeys

@maropu maropu Jul 10, 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.

You cannot use the original condition in logical.Join?

case p @ ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, condition, left, right, hint) =>
  p.condition <-- This?

@AngersZhuuuu AngersZhuuuu Jul 10, 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.

You cannot use the original condition in logical.Join?

case p @ ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, condition, left, right, hint) =>
  p.condition <-- This?

Don't know we can write like this....updated..

val isCartesianProductJoinHint =
hint.exists(_.strategy.exists(_ == SHUFFLE_REPLICATE_NL))
if (isCartesianProductJoinHint) {
hintErrorHandler.wrongHintForNonCartesianProductJoin(hint.get)
None
} else {
hint
}
}

Expand Down
6 changes: 6 additions & 0 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/hints.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -200,4 +200,10 @@ trait HintErrorHandler {
* @param hint the [[HintInfo]] being overridden
*/
def hintOverridden(hint: HintInfo): Unit

/**
* Callback for a wrong [[SHUFFLE_REPLICATE_NL]] hint for Non Cartesian product Join
* @param hint the [[HintInfo]]
*/
def wrongHintForNonCartesianProductJoin(hint: HintInfo): Unit
}
62 changes: 53 additions & 9 deletions sql/core/src/test/scala/org/apache/spark/sql/JoinHintSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -239,7 +239,7 @@ class JoinHintSuite extends PlanTest with SharedSparkSession with AdaptiveSparkP
.hint("shuffle_replicate_nl")
.join(df, "id"),
JoinHint(
Some(HintInfo(strategy = Some(SHUFFLE_REPLICATE_NL))),
None,
None) ::
JoinHint(
None,
Expand Down Expand Up @@ -291,7 +291,7 @@ class JoinHintSuite extends PlanTest with SharedSparkSession with AdaptiveSparkP
Some(HintInfo(strategy = Some(BROADCAST))),
Some(HintInfo(strategy = Some(SHUFFLE_MERGE)))) ::
JoinHint(
Some(HintInfo(strategy = Some(SHUFFLE_REPLICATE_NL))),
None,
Some(HintInfo(strategy = Some(SHUFFLE_HASH)))) :: Nil,
msgNoHintRelationFound("c", "broadcast(a, c)") ::
msgJoinHintOverridden("merge") ::
Expand Down Expand Up @@ -336,6 +336,10 @@ class JoinHintSuite extends PlanTest with SharedSparkSession with AdaptiveSparkP
hints.map("/*+ " + _ + " */").mkString(
"SELECT ", " ", s" * FROM t1 $joinType JOIN t2 ON t1.key = t2.key")

def nonCondJoinQueryWithHint(hints: Seq[String], joinType: String = "INNER"): String =
hints.map("/*+ " + _ + " */").mkString(
"SELECT ", " ", s" * FROM t1 $joinType JOIN t2")

def nonEquiJoinQueryWithHint(hints: Seq[String], joinType: String = "INNER"): String =
hints.map("/*+ " + _ + " */").mkString(
"SELECT ", " ", s" * FROM t1 $joinType JOIN t2 ON t1.key > t2.key")
Expand Down Expand Up @@ -375,12 +379,12 @@ class JoinHintSuite extends PlanTest with SharedSparkSession with AdaptiveSparkP
assert(shuffleMergeJoins.size == 1)
}

private def assertShuffleReplicateNLJoin(df: DataFrame): Unit = {
private def assertShuffleReplicateNLJoin(df: DataFrame, size: Int = 1): Unit = {
val executedPlan = df.queryExecution.executedPlan
val shuffleReplicateNLJoins = collect(executedPlan) {
case c: CartesianProductExec => c
}
assert(shuffleReplicateNLJoins.size == 1)
assert(shuffleReplicateNLJoins.size == size)
}

test("join strategy hint - broadcast") {
Expand Down Expand Up @@ -523,29 +527,51 @@ class JoinHintSuite extends PlanTest with SharedSparkSession with AdaptiveSparkP
withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> Int.MaxValue.toString) {
// Shuffle-replicate-nl hint specified on one side
assertShuffleReplicateNLJoin(
sql(equiJoinQueryWithHint("SHUFFLE_REPLICATE_NL(t1)" :: Nil)))
sql(nonCondJoinQueryWithHint("SHUFFLE_REPLICATE_NL(t1)" :: Nil)))
assertShuffleReplicateNLJoin(
sql(equiJoinQueryWithHint("SHUFFLE_REPLICATE_NL(t2)" :: Nil)))
sql(nonCondJoinQueryWithHint("SHUFFLE_REPLICATE_NL(t2)" :: Nil)))

// Shuffle-replicate-nl hint specified on both sides
assertShuffleReplicateNLJoin(
sql(equiJoinQueryWithHint("SHUFFLE_REPLICATE_NL(t1, t2)" :: Nil)))
sql(nonCondJoinQueryWithHint("SHUFFLE_REPLICATE_NL(t1, t2)" :: Nil)))

// Shuffle-merge hint prioritized over shuffle-replicate-nl hint, but shuffle-merge hint
// is not applicable
assertShuffleReplicateNLJoin(
sql(nonEquiJoinQueryWithHint("MERGE(t1)" :: "SHUFFLE_REPLICATE_NL(t2)" :: Nil)))
sql(nonCondJoinQueryWithHint("MERGE(t1)" :: "SHUFFLE_REPLICATE_NL(t2)" :: Nil)))

// Shuffle-hash hint prioritized over shuffle-replicate-nl hint, but shuffle-hash hint is
// not applicable
assertShuffleReplicateNLJoin(
sql(nonEquiJoinQueryWithHint("SHUFFLE_HASH(t2)" :: "SHUFFLE_REPLICATE_NL(t1)" :: Nil)))
sql(nonCondJoinQueryWithHint("SHUFFLE_HASH(t2)" :: "SHUFFLE_REPLICATE_NL(t1)" :: Nil)))

// Shuffle-replicate-nl hint specified but not doable
assertBroadcastHashJoin(
sql(equiJoinQueryWithHint("SHUFFLE_REPLICATE_NL(t1, t2)" :: Nil, "left")), BuildRight)
assertBroadcastNLJoin(
sql(nonEquiJoinQueryWithHint("SHUFFLE_REPLICATE_NL(t1, t2)" :: Nil, "right")), BuildLeft)


// Verify remove error shuffle_replicate_nl hint
// Shuffle-replicate-nl hint specified on one side
assertShuffleReplicateNLJoin(
sql(equiJoinQueryWithHint("SHUFFLE_REPLICATE_NL(t1)" :: Nil)), 0)
assertShuffleReplicateNLJoin(
sql(equiJoinQueryWithHint("SHUFFLE_REPLICATE_NL(t2)" :: Nil)), 0)

// Shuffle-replicate-nl hint specified on both sides
assertShuffleReplicateNLJoin(
sql(equiJoinQueryWithHint("SHUFFLE_REPLICATE_NL(t1, t2)" :: Nil)), 0)

// Shuffle-merge hint prioritized over shuffle-replicate-nl hint, but shuffle-merge hint
// is not applicable
assertShuffleReplicateNLJoin(
sql(nonEquiJoinQueryWithHint("MERGE(t1)" :: "SHUFFLE_REPLICATE_NL(t2)" :: Nil)), 0)

// Shuffle-hash hint prioritized over shuffle-replicate-nl hint, but shuffle-hash hint is
// not applicable
assertShuffleReplicateNLJoin(
sql(nonEquiJoinQueryWithHint("SHUFFLE_HASH(t2)" :: "SHUFFLE_REPLICATE_NL(t1)" :: Nil)), 0)
}
}
}
Expand All @@ -570,4 +596,22 @@ class JoinHintSuite extends PlanTest with SharedSparkSession with AdaptiveSparkP
assert(joinHints == expectedHints)
}
}

test("SPARK-32220: Non Cartesian Product Join Result Correct with SHUFFLE_REPLICATE_NL hint") {

@dongjoon-hyun dongjoon-hyun Jul 9, 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.

So, is this a correctness issue, @AngersZhuuuu ?

@maropu maropu Jul 9, 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.

Yea, I think so. Nice catch, @AngersZhuuuu

@AngersZhuuuu AngersZhuuuu Jul 10, 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.

Yea, when I try new join hint, I found this result is non-correct.

withTempView("t1", "t2") {
Seq((1, "4"), (2, "2")).toDF("key", "value").createTempView("t1")
Seq((1, "1"), (2, "12.3"), (2, "123")).toDF("key", "value").createTempView("t2")
val df1 = sql("SELECT /*+ shuffle_replicate_nl(t1) */ * from t1 join t2 ON t1.key = t2.key")
val df2 = sql("SELECT * from t1 join t2 ON t1.key = t2.key")
assert(df1.collect().size == df2.collect().size)

val df3 = sql("SELECT /*+ shuffle_replicate_nl(t1) */ * from t1 join t2")
val df4 = sql("SELECT * from t1 join t2")
assert(df3.collect().size == df4.collect().size)

val df5 = sql("SELECT /*+ shuffle_replicate_nl(t1) */ * from t1 join t2 ON t1.key < t2.key")
val df6 = sql("SELECT * from t1 join t2 ON t1.key < t2.key")
assert(df5.collect().size == df6.collect().size)
}
}
}