-
Notifications
You must be signed in to change notification settings - Fork 28.5k
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.
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
[SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross) #31736
Conversation
* Generate the (non-equi) condition used to filter joined rows. This is used in Inner, Left Semi | ||
* and Left Anti joins. | ||
*/ | ||
protected def getJoinCondition( |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This is copied from HashJoin.scala: getJoinCondition()
.
/** | ||
* Generates the code for variable of build side. | ||
*/ | ||
protected def genBuildSideVars( |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This is copied from HashJoin.scala: genBuildSideVars()
.
cc @cloud-fan and @maropu if you have time to take a look, thanks. |
@@ -178,6 +191,7 @@ object JoinBenchmark extends SqlBasedBenchmark { | |||
sortMergeJoin() | |||
sortMergeJoinWithDuplicates() | |||
shuffleHashJoin() | |||
broadcastNestedLoopJoin() |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Could we update the benchmark results(JoinBenchmark-results.txt and JoinBenchmark-jdk11-results.txt)?
Generate result:
SPARK_GENERATE_BENCHMARK_FILES=1 build/sbt "sql/test:runMain org.apache.spark.sql.execution.benchmark.JoinBenchmark"
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@wangyum - sure, I noticed the original benchmark was running on OpenJDK 64-Bit Server VM 1.8.0_232-8u232-b09-0ubuntu1~18.04.1-b09 on Linux 4.15.0-1044-aws
in result file. Should I try to run benchmark in the same type of machine? or this is not a hard requirement. cc @maropu as well.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
val M = 1 << 4 | ||
|
||
val dim = broadcast(spark.range(M).selectExpr("id as k", "cast(id as string) as v")) | ||
codegenBenchmark("broadcast nested loop join", N) { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Could you update benchmarks/JoinBenchmark-results.txt
, too?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@maropu - sure, replied back above.
assert(oneJoinDF.queryExecution.executedPlan.collect { | ||
case WholeStageCodegenExec(_ : BroadcastNestedLoopJoinExec) => true | ||
}.size === 1) | ||
checkAnswer(oneJoinDF, |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
For better test coverage, could you run tests with codegen enabled/disabled?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@maropu - sure, updated.
streamed.asInstanceOf[CodegenSupport].inputRDDs() | ||
} | ||
|
||
override def needCopyResult: Boolean = true |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
always true? Is there a trade-off between the overheads of uniqueness checks and result copys?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@maropu - in case of inner/cross broadcast nested loop join, one input row can potentially have multiple output rows, so I am following the comment here to set it true. btw sort merge join has same behavior. I think if we want to improve it, I can do in another followup PR. I am not very familiar with this part and I can check closer later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This is not a hash join and we don't know the uniqueness. I think it needs to be always true.
case _: InnerLike => codegenInner(ctx, input) | ||
case x => | ||
throw new IllegalArgumentException( | ||
s"BroadcastNestedLoopJoin code-gen should not take $x as the JoinType") |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
nit:
case _ =>
throw new IllegalArgumentException(
s"BroadcastNestedLoopJoin code-gen should not take $joinType as the JoinType")
?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@maropu - updated.
} | ||
|
||
private def codegenInner(ctx: CodegenContext, input: Seq[ExprCode]): String = { | ||
val arrayTerm = prepareBroadcast(ctx) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
nit: arrayTerm
-> buildRowArrayTerm
, buildRowsTerm
, ...?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@maropu - updated to buildRowArrayTerm
.
val broadcastTerm = ctx.addReferenceObj("broadcastTerm", broadcastArray) | ||
|
||
// Inline mutable state since not many join operations in a task | ||
ctx.addMutableState("InternalRow[]", "broadcastArray", |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
nit: broadcastArray
-> buildRows
, buidlRowArray
, ...?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@maropu - updated to buildRowArray
.
Kubernetes integration test starting |
Kubernetes integration test status success |
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala
Show resolved
Hide resolved
Kubernetes integration test starting |
|
||
/** | ||
* Generate the (non-equi) condition used to filter joined rows. This is used in Inner, Left Semi | ||
* and Left Anti joins. |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Since it's in a base trait now, can we add more doc to explain the return type?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@cloud-fan - makes sense, updated.
Kubernetes integration test status failure |
s""" | ||
|int $arrayIndex = 0; | ||
|UnsafeRow $buildRow; | ||
|while ($arrayIndex < $buildRowArrayTerm.length) { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Let's make it more java style
for (int $arrayIndex = 0; $arrayIndex < $buildRowArrayTerm.length; $arrayIndex++) {
UnsafeRow $buildRow = (UnsafeRow) $buildRowArrayTerm[$arrayIndex];
...
}
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I vaguely remembered some places in one PR that I saw while
is preferred over for
(or my memory can be wrong and it's opposite). @cloud-fan - do you remember related discussion?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
while
is preferred over for
in Scala. I don't think it's true for Java, at least from my experience of Java development in the past.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@cloud-fan - updated.
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala
Show resolved
Hide resolved
Addressed all comments, and the PR is ready to review again, thanks @cloud-fan , @maropu and @wangyum . |
Kubernetes integration test starting |
Kubernetes integration test status success |
* Generate the (non-equi) condition used to filter joined rows. | ||
* This is used in Inner, Left Semi and Left Anti joins. | ||
* | ||
* @return Variable name for row of build side. |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
nit: How about this instead?
* @return Tuple of variable name for row of build side, generated code for condition,
* and generated code for variables of build side.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@maropu - sure, updated.
Test build #135826 has finished for PR 31736 at commit
|
Kubernetes integration test starting |
Kubernetes integration test status success |
Test build #135847 has finished for PR 31736 at commit
|
*/ | ||
protected def getJoinCondition( | ||
ctx: CodegenContext, | ||
input: Seq[ExprCode], |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
nit: streamVars
?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@cloud-fan - updated.
Addressed all comments for now, and the PR is ready for check again, thanks. |
Close & reopen PR to trigger test rerun as some transient unit test failure happened. |
Kubernetes integration test starting |
Kubernetes integration test status success |
Test build #135901 has finished for PR 31736 at commit
|
thanks, merging to master! |
Thank you @cloud-fan , @maropu , @viirya and @wangyum for review! |
What changes were proposed in this pull request?
BroadcastNestedLoopJoinExec
does not have code-gen, and we can potentially boost the CPU performance for this operator if we add code-gen for it. https://databricks.com/blog/2017/02/16/processing-trillion-rows-per-second-single-machine-can-nested-loop-joins-fast.html also showed the evidence in one fork.The codegen for
BroadcastNestedLoopJoinExec
shared some code withHashJoin
, and the interfaceJoinCodegenSupport
is created to hold those common logic. This PR is only supporting inner and cross join. Other join types will be added later in followup PRs.Example query and generated code:
Why are the changes needed?
Improve query CPU performance. Added a micro benchmark query in
JoinBenchmark.scala
.Saw 1x of run time improvement:
Does this PR introduce any user-facing change?
No.
How was this patch tested?
WholeStageCodegenSuite.scala
, and existing unit tests forBroadcastNestedLoopJoinExec
.BroadcastNestedLoopJoinExec
is triggered.JoinBenchmark-jdk11-results.txt
andJoinBenchmark-results.txt
with new benchmark result. Followed previous benchmark PRs - [SPARK-30409][SPARK-29173][SQL][TESTS] UseNoOp
datasource in SQL benchmarks #27078 and [SPARK-29320][TESTS] Comparesql/core
module in JDK8/11 (Part 1) #26003 to use same type of machine: