You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2021/03/04 12:29:17 UTC
[GitHub] [spark] c21 opened a new pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
c21 opened a new pull request #31736:
URL: https://github.com/apache/spark/pull/31736
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
`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 with `HashJoin`, and the interface `JoinCodegenSupport` 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:
```
val df1 = spark.range(4).select($"id".as("k1"))
val df2 = spark.range(3).select($"id".as("k2"))
df1.join(df2, $"k1" + 1 =!= $"k2").explain("codegen")
```
```
== Subtree 2 / 2 (maxMethodCodeSize:282; maxConstantPoolSize:203(0.31% used); numInnerClasses:0) ==
*(2) BroadcastNestedLoopJoin BuildRight, Inner, NOT ((k1#2L + 1) = k2#6L)
:- *(2) Project [id#0L AS k1#2L]
: +- *(2) Range (0, 4, step=1, splits=2)
+- BroadcastExchange IdentityBroadcastMode, [id=#22]
+- *(1) Project [id#4L AS k2#6L]
+- *(1) Range (0, 3, step=1, splits=2)
Generated code:
/* 001 */ public Object generate(Object[] references) {
/* 002 */ return new GeneratedIteratorForCodegenStage2(references);
/* 003 */ }
/* 004 */
/* 005 */ // codegenStageId=2
/* 006 */ final class GeneratedIteratorForCodegenStage2 extends org.apache.spark.sql.execution.BufferedRowIterator {
/* 007 */ private Object[] references;
/* 008 */ private scala.collection.Iterator[] inputs;
/* 009 */ private boolean range_initRange_0;
/* 010 */ private long range_nextIndex_0;
/* 011 */ private TaskContext range_taskContext_0;
/* 012 */ private InputMetrics range_inputMetrics_0;
/* 013 */ private long range_batchEnd_0;
/* 014 */ private long range_numElementsTodo_0;
/* 015 */ private InternalRow[] bnlj_broadcastArray_0;
/* 016 */ private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter[] range_mutableStateArray_0 = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter[4];
/* 017 */
/* 018 */ public GeneratedIteratorForCodegenStage2(Object[] references) {
/* 019 */ this.references = references;
/* 020 */ }
/* 021 */
/* 022 */ public void init(int index, scala.collection.Iterator[] inputs) {
/* 023 */ partitionIndex = index;
/* 024 */ this.inputs = inputs;
/* 025 */
/* 026 */ range_taskContext_0 = TaskContext.get();
/* 027 */ range_inputMetrics_0 = range_taskContext_0.taskMetrics().inputMetrics();
/* 028 */ range_mutableStateArray_0[0] = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(1, 0);
/* 029 */ range_mutableStateArray_0[1] = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(1, 0);
/* 030 */ range_mutableStateArray_0[2] = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(1, 0);
/* 031 */ bnlj_broadcastArray_0 = (InternalRow[]) ((org.apache.spark.broadcast.TorrentBroadcast) references[1] /* broadcastTerm */).value();
/* 032 */ range_mutableStateArray_0[3] = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(2, 0);
/* 033 */
/* 034 */ }
/* 035 */
/* 036 */ private void bnlj_doConsume_0(long bnlj_expr_0_0) throws java.io.IOException {
/* 037 */ int bnlj_arrayIndex_0 = 0;
/* 038 */ UnsafeRow bnlj_buildRow_0;
/* 039 */ while (bnlj_arrayIndex_0 < bnlj_broadcastArray_0.length) {
/* 040 */ bnlj_buildRow_0 = (UnsafeRow) bnlj_broadcastArray_0[bnlj_arrayIndex_0];
/* 041 */
/* 042 */ long bnlj_value_1 = bnlj_buildRow_0.getLong(0);
/* 043 */
/* 044 */ long bnlj_value_4 = -1L;
/* 045 */
/* 046 */ bnlj_value_4 = bnlj_expr_0_0 + 1L;
/* 047 */
/* 048 */ boolean bnlj_value_3 = false;
/* 049 */ bnlj_value_3 = bnlj_value_4 == bnlj_value_1;
/* 050 */ boolean bnlj_value_2 = false;
/* 051 */ bnlj_value_2 = !(bnlj_value_3);
/* 052 */ if (!(false || !bnlj_value_2))
/* 053 */ {
/* 054 */ ((org.apache.spark.sql.execution.metric.SQLMetric) references[2] /* numOutputRows */).add(1);
/* 055 */
/* 056 */ range_mutableStateArray_0[3].reset();
/* 057 */
/* 058 */ range_mutableStateArray_0[3].write(0, bnlj_expr_0_0);
/* 059 */
/* 060 */ range_mutableStateArray_0[3].write(1, bnlj_value_1);
/* 061 */ append((range_mutableStateArray_0[3].getRow()).copy());
/* 062 */
/* 063 */ }
/* 064 */ bnlj_arrayIndex_0 += 1;
/* 065 */ }
/* 066 */
/* 067 */ }
/* 068 */
/* 069 */ private void initRange(int idx) {
/* 070 */ java.math.BigInteger index = java.math.BigInteger.valueOf(idx);
/* 071 */ java.math.BigInteger numSlice = java.math.BigInteger.valueOf(2L);
/* 072 */ java.math.BigInteger numElement = java.math.BigInteger.valueOf(4L);
/* 073 */ java.math.BigInteger step = java.math.BigInteger.valueOf(1L);
/* 074 */ java.math.BigInteger start = java.math.BigInteger.valueOf(0L);
/* 075 */ long partitionEnd;
/* 076 */
/* 077 */ java.math.BigInteger st = index.multiply(numElement).divide(numSlice).multiply(step).add(start);
/* 078 */ if (st.compareTo(java.math.BigInteger.valueOf(Long.MAX_VALUE)) > 0) {
/* 079 */ range_nextIndex_0 = Long.MAX_VALUE;
/* 080 */ } else if (st.compareTo(java.math.BigInteger.valueOf(Long.MIN_VALUE)) < 0) {
/* 081 */ range_nextIndex_0 = Long.MIN_VALUE;
/* 082 */ } else {
/* 083 */ range_nextIndex_0 = st.longValue();
/* 084 */ }
/* 085 */ range_batchEnd_0 = range_nextIndex_0;
/* 086 */
/* 087 */ java.math.BigInteger end = index.add(java.math.BigInteger.ONE).multiply(numElement).divide(numSlice)
/* 088 */ .multiply(step).add(start);
/* 089 */ if (end.compareTo(java.math.BigInteger.valueOf(Long.MAX_VALUE)) > 0) {
/* 090 */ partitionEnd = Long.MAX_VALUE;
/* 091 */ } else if (end.compareTo(java.math.BigInteger.valueOf(Long.MIN_VALUE)) < 0) {
/* 092 */ partitionEnd = Long.MIN_VALUE;
/* 093 */ } else {
/* 094 */ partitionEnd = end.longValue();
/* 095 */ }
/* 096 */
/* 097 */ java.math.BigInteger startToEnd = java.math.BigInteger.valueOf(partitionEnd).subtract(
/* 098 */ java.math.BigInteger.valueOf(range_nextIndex_0));
/* 099 */ range_numElementsTodo_0 = startToEnd.divide(step).longValue();
/* 100 */ if (range_numElementsTodo_0 < 0) {
/* 101 */ range_numElementsTodo_0 = 0;
/* 102 */ } else if (startToEnd.remainder(step).compareTo(java.math.BigInteger.valueOf(0L)) != 0) {
/* 103 */ range_numElementsTodo_0++;
/* 104 */ }
/* 105 */ }
/* 106 */
/* 107 */ protected void processNext() throws java.io.IOException {
/* 108 */ // initialize Range
/* 109 */ if (!range_initRange_0) {
/* 110 */ range_initRange_0 = true;
/* 111 */ initRange(partitionIndex);
/* 112 */ }
/* 113 */
/* 114 */ while (true) {
/* 115 */ if (range_nextIndex_0 == range_batchEnd_0) {
/* 116 */ long range_nextBatchTodo_0;
/* 117 */ if (range_numElementsTodo_0 > 1000L) {
/* 118 */ range_nextBatchTodo_0 = 1000L;
/* 119 */ range_numElementsTodo_0 -= 1000L;
/* 120 */ } else {
/* 121 */ range_nextBatchTodo_0 = range_numElementsTodo_0;
/* 122 */ range_numElementsTodo_0 = 0;
/* 123 */ if (range_nextBatchTodo_0 == 0) break;
/* 124 */ }
/* 125 */ range_batchEnd_0 += range_nextBatchTodo_0 * 1L;
/* 126 */ }
/* 127 */
/* 128 */ int range_localEnd_0 = (int)((range_batchEnd_0 - range_nextIndex_0) / 1L);
/* 129 */ for (int range_localIdx_0 = 0; range_localIdx_0 < range_localEnd_0; range_localIdx_0++) {
/* 130 */ long range_value_0 = ((long)range_localIdx_0 * 1L) + range_nextIndex_0;
/* 131 */
/* 132 */ // common sub-expressions
/* 133 */
/* 134 */ bnlj_doConsume_0(range_value_0);
/* 135 */
/* 136 */ if (shouldStop()) {
/* 137 */ range_nextIndex_0 = range_value_0 + 1L;
/* 138 */ ((org.apache.spark.sql.execution.metric.SQLMetric) references[0] /* numOutputRows */).add(range_localIdx_0 + 1);
/* 139 */ range_inputMetrics_0.incRecordsRead(range_localIdx_0 + 1);
/* 140 */ return;
/* 141 */ }
/* 142 */
/* 143 */ }
/* 144 */ range_nextIndex_0 = range_batchEnd_0;
/* 145 */ ((org.apache.spark.sql.execution.metric.SQLMetric) references[0] /* numOutputRows */).add(range_localEnd_0);
/* 146 */ range_inputMetrics_0.incRecordsRead(range_localEnd_0);
/* 147 */ range_taskContext_0.killTaskIfInterrupted();
/* 148 */ }
/* 149 */ }
/* 150 */
/* 151 */ }
```
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
Improve query CPU performance. Added a micro benchmark query in `JoinBenchmark.scala`.
Saw 70% of run time improvement:
```
Running benchmark: broadcast nested loop join
Running case: broadcast nested loop join wholestage off
Stopped after 2 iterations, 48367 ms
Running case: broadcast nested loop join wholestage on
Stopped after 5 iterations, 74502 ms
Java HotSpot(TM) 64-Bit Server VM 1.8.0_181-b13 on Mac OS X 10.15.7
Intel(R) Core(TM) i9-9980HK CPU @ 2.40GHz
broadcast nested loop join: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
-------------------------------------------------------------------------------------------------------------------------
broadcast nested loop join wholestage off 24130 24184 76 0.9 1150.6 1.0X
broadcast nested loop join wholestage on 14589 14900 220 1.4 695.7 1.7X
```
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as the documentation fix.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
No.
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
-->
Added unit test in `WholeStageCodegenSuite.scala`, and existing unit tests for `BroadcastNestedLoopJoinExec`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-792371243
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/40429/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-792357552
**[Test build #135847 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/135847/testReport)** for PR 31736 at commit [`4d26e49`](https://github.com/apache/spark/commit/4d26e49114ea2090f8725b394b4c060ef706a04c).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] c21 commented on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
c21 commented on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-791904879
Addressed all comments, and the PR is ready to review again, thanks @cloud-fan , @maropu and @wangyum .
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-790642159
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/40336/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #31736:
URL: https://github.com/apache/spark/pull/31736#discussion_r588154614
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import org.apache.spark.sql.catalyst.expressions.{BindReferences, BoundReference}
+import org.apache.spark.sql.catalyst.expressions.codegen._
+import org.apache.spark.sql.catalyst.expressions.codegen.Block._
+import org.apache.spark.sql.catalyst.plans.InnerLike
+import org.apache.spark.sql.execution.{CodegenSupport, SparkPlan}
+
+/**
+ * An interface for those join physical operators that support codegen.
+ */
+trait JoinCodegenSupport extends CodegenSupport with BaseJoinExec {
+
+ /**
+ * Generate the (non-equi) condition used to filter joined rows. This is used in Inner, Left Semi
+ * and Left Anti joins.
+ */
+ protected def getJoinCondition(
+ ctx: CodegenContext,
+ input: Seq[ExprCode],
+ streamedPlan: SparkPlan,
+ buildPlan: SparkPlan): (String, String, Seq[ExprCode]) = {
+ val buildRow = ctx.freshName("buildRow")
+ val buildVars = genBuildSideVars(ctx, buildRow, buildPlan)
+ val checkCondition = if (condition.isDefined) {
+ val expr = condition.get
+ // evaluate the variables from build side that used by condition
Review comment:
`input` also includes `buidVars`, my question is why this line of code here only handles build side not stream side.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-791925030
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/40408/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] c21 commented on a change in pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
c21 commented on a change in pull request #31736:
URL: https://github.com/apache/spark/pull/31736#discussion_r588148408
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastNestedLoopJoinExec.scala
##########
@@ -393,4 +394,65 @@ case class BroadcastNestedLoopJoinExec(
}
}
}
+
+ override def supportCodegen: Boolean = {
+ joinType.isInstanceOf[InnerLike]
+ }
+
+ override def inputRDDs(): Seq[RDD[InternalRow]] = {
+ streamed.asInstanceOf[CodegenSupport].inputRDDs()
+ }
+
+ override def needCopyResult: Boolean = true
+
+ override def doProduce(ctx: CodegenContext): String = {
+ streamed.asInstanceOf[CodegenSupport].produce(ctx, this)
+ }
+
+ override def doConsume(ctx: CodegenContext, input: Seq[ExprCode], row: ExprCode): String = {
+ joinType match {
+ case _: InnerLike => codegenInner(ctx, input)
+ case _ =>
+ throw new IllegalArgumentException(
+ s"BroadcastNestedLoopJoin code-gen should not take $joinType as the JoinType")
+ }
+ }
+
+ /**
+ * Returns the variable name for [[Broadcast]] side.
+ */
+ private def prepareBroadcast(ctx: CodegenContext): String = {
+ // Create a name for broadcast side
+ val broadcastArray = broadcast.executeBroadcast[Array[InternalRow]]()
+ val broadcastTerm = ctx.addReferenceObj("broadcastTerm", broadcastArray)
+
+ // Inline mutable state since not many join operations in a task
+ ctx.addMutableState("InternalRow[]", "buildRowArray",
+ v => s"$v = (InternalRow[]) $broadcastTerm.value();", forceInline = true)
+ }
+
+ private def codegenInner(ctx: CodegenContext, input: Seq[ExprCode]): String = {
+ val buildRowArrayTerm = prepareBroadcast(ctx)
+ val (buildRow, checkCondition, buildVars) = getJoinCondition(ctx, input, streamed, broadcast)
+
+ val resultVars = buildSide match {
+ case BuildLeft => buildVars ++ input
+ case BuildRight => input ++ buildVars
+ }
+ val arrayIndex = ctx.freshName("arrayIndex")
+ val numOutput = metricTerm(ctx, "numOutputRows")
+
+ s"""
+ |int $arrayIndex = 0;
+ |UnsafeRow $buildRow;
+ |while ($arrayIndex < $buildRowArrayTerm.length) {
Review comment:
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?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-791914317
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/40408/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-792371243
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/40429/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] c21 closed pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
c21 closed pull request #31736:
URL: https://github.com/apache/spark/pull/31736
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] c21 commented on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
c21 commented on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-790583890
cc @cloud-fan and @maropu if you have time to take a look, thanks.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-792369144
Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/40429/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-793530795
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/40483/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] c21 commented on a change in pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
c21 commented on a change in pull request #31736:
URL: https://github.com/apache/spark/pull/31736#discussion_r587429960
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import org.apache.spark.sql.catalyst.expressions.{BindReferences, BoundReference}
+import org.apache.spark.sql.catalyst.expressions.codegen._
+import org.apache.spark.sql.catalyst.expressions.codegen.Block._
+import org.apache.spark.sql.catalyst.plans.InnerLike
+import org.apache.spark.sql.execution.{CodegenSupport, SparkPlan}
+
+/**
+ * An interface for those join physical operators that support codegen.
+ */
+trait JoinCodegenSupport extends CodegenSupport with BaseJoinExec {
+
+ /**
+ * Generate the (non-equi) condition used to filter joined rows. This is used in Inner, Left Semi
+ * and Left Anti joins.
+ */
+ protected def getJoinCondition(
Review comment:
This is copied from `HashJoin.scala: getJoinCondition()`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-792357552
**[Test build #135847 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/135847/testReport)** for PR 31736 at commit [`4d26e49`](https://github.com/apache/spark/commit/4d26e49114ea2090f8725b394b4c060ef706a04c).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] c21 commented on a change in pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
c21 commented on a change in pull request #31736:
URL: https://github.com/apache/spark/pull/31736#discussion_r588083648
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala
##########
@@ -92,6 +92,39 @@ class WholeStageCodegenSuite extends QueryTest with SharedSparkSession
Seq(Row(0, 0, 0), Row(1, 1, 1), Row(2, 2, 2), Row(3, 3, 3), Row(4, 4, 4)))
}
+ test("Inner/Cross BroadcastNestedLoopJoinExec should be included in WholeStageCodegen") {
+ val df1 = spark.range(4).select($"id".as("k1"))
+ val df2 = spark.range(3).select($"id".as("k2"))
+ val df3 = spark.range(2).select($"id".as("k3"))
+
+ // test broadcast nested loop join without condition
+ val oneJoinDF = df1.join(df2)
+ assert(oneJoinDF.queryExecution.executedPlan.collect {
+ case WholeStageCodegenExec(_ : BroadcastNestedLoopJoinExec) => true
+ }.size === 1)
+ checkAnswer(oneJoinDF,
Review comment:
@maropu - sure, updated.
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastNestedLoopJoinExec.scala
##########
@@ -393,4 +394,65 @@ case class BroadcastNestedLoopJoinExec(
}
}
}
+
+ override def supportCodegen: Boolean = {
+ joinType.isInstanceOf[InnerLike]
+ }
+
+ override def inputRDDs(): Seq[RDD[InternalRow]] = {
+ streamed.asInstanceOf[CodegenSupport].inputRDDs()
+ }
+
+ override def needCopyResult: Boolean = true
Review comment:
@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](https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/WholeStageCodegenExec.scala#L346-L347) to set it true. btw sort merge join has [same behavior](https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoinExec.scala#L570). 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.
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastNestedLoopJoinExec.scala
##########
@@ -393,4 +394,65 @@ case class BroadcastNestedLoopJoinExec(
}
}
}
+
+ override def supportCodegen: Boolean = {
+ joinType.isInstanceOf[InnerLike]
+ }
+
+ override def inputRDDs(): Seq[RDD[InternalRow]] = {
+ streamed.asInstanceOf[CodegenSupport].inputRDDs()
+ }
+
+ override def needCopyResult: Boolean = true
+
+ override def doProduce(ctx: CodegenContext): String = {
+ streamed.asInstanceOf[CodegenSupport].produce(ctx, this)
+ }
+
+ override def doConsume(ctx: CodegenContext, input: Seq[ExprCode], row: ExprCode): String = {
+ joinType match {
+ case _: InnerLike => codegenInner(ctx, input)
+ case x =>
+ throw new IllegalArgumentException(
+ s"BroadcastNestedLoopJoin code-gen should not take $x as the JoinType")
+ }
+ }
+
+ /**
+ * Returns the variable name for [[Broadcast]] side.
+ */
+ private def prepareBroadcast(ctx: CodegenContext): String = {
+ // Create a name for broadcast side
+ val broadcastArray = broadcast.executeBroadcast[Array[InternalRow]]()
+ val broadcastTerm = ctx.addReferenceObj("broadcastTerm", broadcastArray)
+
+ // Inline mutable state since not many join operations in a task
+ ctx.addMutableState("InternalRow[]", "broadcastArray",
Review comment:
@maropu - updated to `buildRowArray`.
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastNestedLoopJoinExec.scala
##########
@@ -393,4 +394,65 @@ case class BroadcastNestedLoopJoinExec(
}
}
}
+
+ override def supportCodegen: Boolean = {
+ joinType.isInstanceOf[InnerLike]
+ }
+
+ override def inputRDDs(): Seq[RDD[InternalRow]] = {
+ streamed.asInstanceOf[CodegenSupport].inputRDDs()
+ }
+
+ override def needCopyResult: Boolean = true
+
+ override def doProduce(ctx: CodegenContext): String = {
+ streamed.asInstanceOf[CodegenSupport].produce(ctx, this)
+ }
+
+ override def doConsume(ctx: CodegenContext, input: Seq[ExprCode], row: ExprCode): String = {
+ joinType match {
+ case _: InnerLike => codegenInner(ctx, input)
+ case x =>
+ throw new IllegalArgumentException(
+ s"BroadcastNestedLoopJoin code-gen should not take $x as the JoinType")
Review comment:
@maropu - updated.
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastNestedLoopJoinExec.scala
##########
@@ -393,4 +394,65 @@ case class BroadcastNestedLoopJoinExec(
}
}
}
+
+ override def supportCodegen: Boolean = {
+ joinType.isInstanceOf[InnerLike]
+ }
+
+ override def inputRDDs(): Seq[RDD[InternalRow]] = {
+ streamed.asInstanceOf[CodegenSupport].inputRDDs()
+ }
+
+ override def needCopyResult: Boolean = true
+
+ override def doProduce(ctx: CodegenContext): String = {
+ streamed.asInstanceOf[CodegenSupport].produce(ctx, this)
+ }
+
+ override def doConsume(ctx: CodegenContext, input: Seq[ExprCode], row: ExprCode): String = {
+ joinType match {
+ case _: InnerLike => codegenInner(ctx, input)
+ case x =>
+ throw new IllegalArgumentException(
+ s"BroadcastNestedLoopJoin code-gen should not take $x as the JoinType")
+ }
+ }
+
+ /**
+ * Returns the variable name for [[Broadcast]] side.
+ */
+ private def prepareBroadcast(ctx: CodegenContext): String = {
+ // Create a name for broadcast side
+ val broadcastArray = broadcast.executeBroadcast[Array[InternalRow]]()
+ val broadcastTerm = ctx.addReferenceObj("broadcastTerm", broadcastArray)
+
+ // Inline mutable state since not many join operations in a task
+ ctx.addMutableState("InternalRow[]", "broadcastArray",
+ v => s"$v = (InternalRow[]) $broadcastTerm.value();", forceInline = true)
+ }
+
+ private def codegenInner(ctx: CodegenContext, input: Seq[ExprCode]): String = {
+ val arrayTerm = prepareBroadcast(ctx)
Review comment:
@maropu - updated to `buildRowArrayTerm`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-793799689
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/135901/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] maropu commented on a change in pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
maropu commented on a change in pull request #31736:
URL: https://github.com/apache/spark/pull/31736#discussion_r588883010
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala
##########
@@ -29,8 +29,12 @@ import org.apache.spark.sql.execution.{CodegenSupport, SparkPlan}
trait JoinCodegenSupport extends CodegenSupport with BaseJoinExec {
/**
- * Generate the (non-equi) condition used to filter joined rows. This is used in Inner, Left Semi
- * and Left Anti joins.
+ * 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.
Review comment:
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.
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-791272002
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/135803/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-791965254
**[Test build #135826 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/135826/testReport)** for PR 31736 at commit [`4e5341f`](https://github.com/apache/spark/commit/4e5341f1e7f5e3bf6c849ec587be7274679ce71a).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] c21 commented on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
c21 commented on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-794350669
Thank you @cloud-fan , @maropu , @viirya and @wangyum for review!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] c21 commented on a change in pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
c21 commented on a change in pull request #31736:
URL: https://github.com/apache/spark/pull/31736#discussion_r589087569
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala
##########
@@ -29,8 +29,12 @@ import org.apache.spark.sql.execution.{CodegenSupport, SparkPlan}
trait JoinCodegenSupport extends CodegenSupport with BaseJoinExec {
/**
- * Generate the (non-equi) condition used to filter joined rows. This is used in Inner, Left Semi
- * and Left Anti joins.
+ * 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.
Review comment:
@maropu - sure, updated.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-793562609
Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/40483/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-793596212
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/40483/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-793476281
**[Test build #135901 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/135901/testReport)** for PR 31736 at commit [`6f6d511`](https://github.com/apache/spark/commit/6f6d511e571a2109cba4dcd6b31fa3def4e44e21).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] c21 commented on a change in pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
c21 commented on a change in pull request #31736:
URL: https://github.com/apache/spark/pull/31736#discussion_r588861296
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/JoinBenchmark.scala
##########
@@ -166,6 +166,19 @@ object JoinBenchmark extends SqlBasedBenchmark {
}
}
+ def broadcastNestedLoopJoin(): Unit = {
+ val N = 20 << 20
+ 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) {
Review comment:
@maropu - sure, replied back above.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-790667069
Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/40336/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-793746543
**[Test build #135901 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/135901/testReport)** for PR 31736 at commit [`6f6d511`](https://github.com/apache/spark/commit/6f6d511e571a2109cba4dcd6b31fa3def4e44e21).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #31736:
URL: https://github.com/apache/spark/pull/31736#discussion_r589267382
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala
##########
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import org.apache.spark.sql.catalyst.expressions.{BindReferences, BoundReference}
+import org.apache.spark.sql.catalyst.expressions.codegen._
+import org.apache.spark.sql.catalyst.expressions.codegen.Block._
+import org.apache.spark.sql.catalyst.plans.InnerLike
+import org.apache.spark.sql.execution.{CodegenSupport, SparkPlan}
+
+/**
+ * An interface for those join physical operators that support codegen.
+ */
+trait JoinCodegenSupport extends CodegenSupport with BaseJoinExec {
+
+ /**
+ * Generate the (non-equi) condition used to filter joined rows.
+ * This is used in Inner, Left Semi and Left Anti joins.
+ *
+ * @return Tuple of variable name for row of build side, generated code for condition,
+ * and generated code for variables of build side.
+ */
+ protected def getJoinCondition(
+ ctx: CodegenContext,
+ input: Seq[ExprCode],
Review comment:
nit: `streamVars`?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #31736:
URL: https://github.com/apache/spark/pull/31736#discussion_r589954425
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import org.apache.spark.sql.catalyst.expressions.{BindReferences, BoundReference}
+import org.apache.spark.sql.catalyst.expressions.codegen._
+import org.apache.spark.sql.catalyst.expressions.codegen.Block._
+import org.apache.spark.sql.catalyst.plans.InnerLike
+import org.apache.spark.sql.execution.{CodegenSupport, SparkPlan}
+
+/**
+ * An interface for those join physical operators that support codegen.
+ */
+trait JoinCodegenSupport extends CodegenSupport with BaseJoinExec {
+
+ /**
+ * Generate the (non-equi) condition used to filter joined rows. This is used in Inner, Left Semi
+ * and Left Anti joins.
+ */
+ protected def getJoinCondition(
+ ctx: CodegenContext,
+ input: Seq[ExprCode],
+ streamedPlan: SparkPlan,
+ buildPlan: SparkPlan): (String, String, Seq[ExprCode]) = {
+ val buildRow = ctx.freshName("buildRow")
+ val buildVars = genBuildSideVars(ctx, buildRow, buildPlan)
+ val checkCondition = if (condition.isDefined) {
+ val expr = condition.get
+ // evaluate the variables from build side that used by condition
Review comment:
ah, good catch! @viirya is right, we don't need to do that.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #31736:
URL: https://github.com/apache/spark/pull/31736#discussion_r588944070
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import org.apache.spark.sql.catalyst.expressions.{BindReferences, BoundReference}
+import org.apache.spark.sql.catalyst.expressions.codegen._
+import org.apache.spark.sql.catalyst.expressions.codegen.Block._
+import org.apache.spark.sql.catalyst.plans.InnerLike
+import org.apache.spark.sql.execution.{CodegenSupport, SparkPlan}
+
+/**
+ * An interface for those join physical operators that support codegen.
+ */
+trait JoinCodegenSupport extends CodegenSupport with BaseJoinExec {
+
+ /**
+ * Generate the (non-equi) condition used to filter joined rows. This is used in Inner, Left Semi
+ * and Left Anti joins.
+ */
+ protected def getJoinCondition(
+ ctx: CodegenContext,
+ input: Seq[ExprCode],
+ streamedPlan: SparkPlan,
+ buildPlan: SparkPlan): (String, String, Seq[ExprCode]) = {
+ val buildRow = ctx.freshName("buildRow")
+ val buildVars = genBuildSideVars(ctx, buildRow, buildPlan)
+ val checkCondition = if (condition.isDefined) {
+ val expr = condition.get
+ // evaluate the variables from build side that used by condition
Review comment:
The variables from stream side are be already evaluated before `doConsume` is called. So you don't need to include stream side variables again.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-790606969
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/135753/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] c21 commented on a change in pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
c21 commented on a change in pull request #31736:
URL: https://github.com/apache/spark/pull/31736#discussion_r588144023
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import org.apache.spark.sql.catalyst.expressions.{BindReferences, BoundReference}
+import org.apache.spark.sql.catalyst.expressions.codegen._
+import org.apache.spark.sql.catalyst.expressions.codegen.Block._
+import org.apache.spark.sql.catalyst.plans.InnerLike
+import org.apache.spark.sql.execution.{CodegenSupport, SparkPlan}
+
+/**
+ * An interface for those join physical operators that support codegen.
+ */
+trait JoinCodegenSupport extends CodegenSupport with BaseJoinExec {
+
+ /**
+ * Generate the (non-equi) condition used to filter joined rows. This is used in Inner, Left Semi
+ * and Left Anti joins.
+ */
+ protected def getJoinCondition(
+ ctx: CodegenContext,
+ input: Seq[ExprCode],
+ streamedPlan: SparkPlan,
+ buildPlan: SparkPlan): (String, String, Seq[ExprCode]) = {
+ val buildRow = ctx.freshName("buildRow")
+ val buildVars = genBuildSideVars(ctx, buildRow, buildPlan)
+ val checkCondition = if (condition.isDefined) {
+ val expr = condition.get
+ // evaluate the variables from build side that used by condition
Review comment:
@cloud-fan - `input` here is the generated code for streamed side. The code is generated below with `ctx.currentVars = input ++ buildVars`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #31736:
URL: https://github.com/apache/spark/pull/31736#discussion_r588141760
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import org.apache.spark.sql.catalyst.expressions.{BindReferences, BoundReference}
+import org.apache.spark.sql.catalyst.expressions.codegen._
+import org.apache.spark.sql.catalyst.expressions.codegen.Block._
+import org.apache.spark.sql.catalyst.plans.InnerLike
+import org.apache.spark.sql.execution.{CodegenSupport, SparkPlan}
+
+/**
+ * An interface for those join physical operators that support codegen.
+ */
+trait JoinCodegenSupport extends CodegenSupport with BaseJoinExec {
+
+ /**
+ * Generate the (non-equi) condition used to filter joined rows. This is used in Inner, Left Semi
+ * and Left Anti joins.
+ */
+ protected def getJoinCondition(
+ ctx: CodegenContext,
+ input: Seq[ExprCode],
+ streamedPlan: SparkPlan,
+ buildPlan: SparkPlan): (String, String, Seq[ExprCode]) = {
+ val buildRow = ctx.freshName("buildRow")
+ val buildVars = genBuildSideVars(ctx, buildRow, buildPlan)
+ val checkCondition = if (condition.isDefined) {
+ val expr = condition.get
+ // evaluate the variables from build side that used by condition
Review comment:
This is coped so my question may not be related to this PR: the join condition can refer to columns from stream side, why don't we need to evaluate variables from the stream side?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-791274134
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/40384/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #31736:
URL: https://github.com/apache/spark/pull/31736#discussion_r588134938
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import org.apache.spark.sql.catalyst.expressions.{BindReferences, BoundReference}
+import org.apache.spark.sql.catalyst.expressions.codegen._
+import org.apache.spark.sql.catalyst.expressions.codegen.Block._
+import org.apache.spark.sql.catalyst.plans.InnerLike
+import org.apache.spark.sql.execution.{CodegenSupport, SparkPlan}
+
+/**
+ * An interface for those join physical operators that support codegen.
+ */
+trait JoinCodegenSupport extends CodegenSupport with BaseJoinExec {
+
+ /**
+ * Generate the (non-equi) condition used to filter joined rows. This is used in Inner, Left Semi
+ * and Left Anti joins.
Review comment:
Since it's in a base trait now, can we add more doc to explain the return type?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #31736:
URL: https://github.com/apache/spark/pull/31736#discussion_r588134104
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import org.apache.spark.sql.catalyst.expressions.{BindReferences, BoundReference}
+import org.apache.spark.sql.catalyst.expressions.codegen._
+import org.apache.spark.sql.catalyst.expressions.codegen.Block._
+import org.apache.spark.sql.catalyst.plans.InnerLike
+import org.apache.spark.sql.execution.{CodegenSupport, SparkPlan}
+
+/**
+ * An interface for those join physical operators that support codegen.
+ */
+trait JoinCodegenSupport extends CodegenSupport with BaseJoinExec {
+
+ /**
+ * Generate the (non-equi) condition used to filter joined rows. This is used in Inner, Left Semi
+ * and Left Anti joins.
+ */
+ protected def getJoinCondition(
+ ctx: CodegenContext,
+ input: Seq[ExprCode],
+ streamedPlan: SparkPlan,
+ buildPlan: SparkPlan): (String, String, Seq[ExprCode]) = {
+ val buildRow = ctx.freshName("buildRow")
+ val buildVars = genBuildSideVars(ctx, buildRow, buildPlan)
+ val checkCondition = if (condition.isDefined) {
+ val expr = condition.get
+ // evaluate the variables from build side that used by condition
+ val eval = evaluateRequiredVariables(buildPlan.output, buildVars, expr.references)
+ // filter the output via condition
+ ctx.currentVars = input ++ buildVars
+ val ev =
+ BindReferences.bindReference(expr, streamedPlan.output ++ buildPlan.output).genCode(ctx)
+ val skipRow = s"${ev.isNull} || !${ev.value}"
+ s"""
+ |$eval
+ |${ev.code}
+ |if (!($skipRow))
+ """.stripMargin
+ } else {
+ ""
+ }
+ (buildRow, checkCondition, buildVars)
+ }
+
+ /**
+ * Generates the code for variable of build side.
+ */
+ protected def genBuildSideVars(
+ ctx: CodegenContext,
+ buildRow: String,
+ buildPlan: SparkPlan): Seq[ExprCode] = {
+ ctx.currentVars = null
+ ctx.INPUT_ROW = buildRow
+ buildPlan.output.zipWithIndex.map { case (a, i) =>
+ val ev = BoundReference(i, a.dataType, a.nullable).genCode(ctx)
+ if (joinType.isInstanceOf[InnerLike]) {
+ ev
+ } else {
+ // the variables are needed even there is no matched rows
Review comment:
This is not true for left-semi/anti, but is fine as this PR only supports inner like.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-790676290
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/40336/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-791968238
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/135826/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #31736:
URL: https://github.com/apache/spark/pull/31736#discussion_r588143547
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import org.apache.spark.sql.catalyst.expressions.{BindReferences, BoundReference}
+import org.apache.spark.sql.catalyst.expressions.codegen._
+import org.apache.spark.sql.catalyst.expressions.codegen.Block._
+import org.apache.spark.sql.catalyst.plans.InnerLike
+import org.apache.spark.sql.execution.{CodegenSupport, SparkPlan}
+
+/**
+ * An interface for those join physical operators that support codegen.
+ */
+trait JoinCodegenSupport extends CodegenSupport with BaseJoinExec {
+
+ /**
+ * Generate the (non-equi) condition used to filter joined rows. This is used in Inner, Left Semi
+ * and Left Anti joins.
+ */
+ protected def getJoinCondition(
+ ctx: CodegenContext,
+ input: Seq[ExprCode],
+ streamedPlan: SparkPlan,
+ buildPlan: SparkPlan): (String, String, Seq[ExprCode]) = {
+ val buildRow = ctx.freshName("buildRow")
+ val buildVars = genBuildSideVars(ctx, buildRow, buildPlan)
+ val checkCondition = if (condition.isDefined) {
+ val expr = condition.get
+ // evaluate the variables from build side that used by condition
Review comment:
This seems like a problem. In `HashJoin`, we have this at the beginning
```
|// generate join key for stream side
|${keyEv.code}
```
Shall we do something similar in BroadcastNestedLoopJoinExec?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #31736:
URL: https://github.com/apache/spark/pull/31736#discussion_r589264829
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import org.apache.spark.sql.catalyst.expressions.{BindReferences, BoundReference}
+import org.apache.spark.sql.catalyst.expressions.codegen._
+import org.apache.spark.sql.catalyst.expressions.codegen.Block._
+import org.apache.spark.sql.catalyst.plans.InnerLike
+import org.apache.spark.sql.execution.{CodegenSupport, SparkPlan}
+
+/**
+ * An interface for those join physical operators that support codegen.
+ */
+trait JoinCodegenSupport extends CodegenSupport with BaseJoinExec {
+
+ /**
+ * Generate the (non-equi) condition used to filter joined rows. This is used in Inner, Left Semi
+ * and Left Anti joins.
+ */
+ protected def getJoinCondition(
+ ctx: CodegenContext,
+ input: Seq[ExprCode],
+ streamedPlan: SparkPlan,
+ buildPlan: SparkPlan): (String, String, Seq[ExprCode]) = {
+ val buildRow = ctx.freshName("buildRow")
+ val buildVars = genBuildSideVars(ctx, buildRow, buildPlan)
+ val checkCondition = if (condition.isDefined) {
+ val expr = condition.get
+ // evaluate the variables from build side that used by condition
Review comment:
@viirya hmmm are you sure about it? AFAIK for operators like `Filter`, it doesn't evaluate all the child outputs, only the ones required by the condition, though `Filter.output` is `child.output`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-790676290
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/40336/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-791908921
**[Test build #135826 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/135826/testReport)** for PR 31736 at commit [`4e5341f`](https://github.com/apache/spark/commit/4e5341f1e7f5e3bf6c849ec587be7274679ce71a).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan closed pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
cloud-fan closed pull request #31736:
URL: https://github.com/apache/spark/pull/31736
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] maropu commented on a change in pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
maropu commented on a change in pull request #31736:
URL: https://github.com/apache/spark/pull/31736#discussion_r587461811
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala
##########
@@ -92,6 +92,39 @@ class WholeStageCodegenSuite extends QueryTest with SharedSparkSession
Seq(Row(0, 0, 0), Row(1, 1, 1), Row(2, 2, 2), Row(3, 3, 3), Row(4, 4, 4)))
}
+ test("Inner/Cross BroadcastNestedLoopJoinExec should be included in WholeStageCodegen") {
+ val df1 = spark.range(4).select($"id".as("k1"))
+ val df2 = spark.range(3).select($"id".as("k2"))
+ val df3 = spark.range(2).select($"id".as("k3"))
+
+ // test broadcast nested loop join without condition
+ val oneJoinDF = df1.join(df2)
+ assert(oneJoinDF.queryExecution.executedPlan.collect {
+ case WholeStageCodegenExec(_ : BroadcastNestedLoopJoinExec) => true
+ }.size === 1)
+ checkAnswer(oneJoinDF,
Review comment:
For better test coverage, could you run tests with codegen enabled/disabled?
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/JoinBenchmark.scala
##########
@@ -166,6 +166,19 @@ object JoinBenchmark extends SqlBasedBenchmark {
}
}
+ def broadcastNestedLoopJoin(): Unit = {
+ val N = 20 << 20
+ 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) {
Review comment:
Could you update `benchmarks/JoinBenchmark-results.txt`, too?
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastNestedLoopJoinExec.scala
##########
@@ -393,4 +394,65 @@ case class BroadcastNestedLoopJoinExec(
}
}
}
+
+ override def supportCodegen: Boolean = {
+ joinType.isInstanceOf[InnerLike]
+ }
+
+ override def inputRDDs(): Seq[RDD[InternalRow]] = {
+ streamed.asInstanceOf[CodegenSupport].inputRDDs()
+ }
+
+ override def needCopyResult: Boolean = true
+
+ override def doProduce(ctx: CodegenContext): String = {
+ streamed.asInstanceOf[CodegenSupport].produce(ctx, this)
+ }
+
+ override def doConsume(ctx: CodegenContext, input: Seq[ExprCode], row: ExprCode): String = {
+ joinType match {
+ case _: InnerLike => codegenInner(ctx, input)
+ case x =>
+ throw new IllegalArgumentException(
+ s"BroadcastNestedLoopJoin code-gen should not take $x as the JoinType")
+ }
+ }
+
+ /**
+ * Returns the variable name for [[Broadcast]] side.
+ */
+ private def prepareBroadcast(ctx: CodegenContext): String = {
+ // Create a name for broadcast side
+ val broadcastArray = broadcast.executeBroadcast[Array[InternalRow]]()
+ val broadcastTerm = ctx.addReferenceObj("broadcastTerm", broadcastArray)
+
+ // Inline mutable state since not many join operations in a task
+ ctx.addMutableState("InternalRow[]", "broadcastArray",
+ v => s"$v = (InternalRow[]) $broadcastTerm.value();", forceInline = true)
+ }
+
+ private def codegenInner(ctx: CodegenContext, input: Seq[ExprCode]): String = {
+ val arrayTerm = prepareBroadcast(ctx)
Review comment:
nit: `arrayTerm` -> `buildRowArrayTerm`, `buildRowsTerm`, ...?
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastNestedLoopJoinExec.scala
##########
@@ -393,4 +394,65 @@ case class BroadcastNestedLoopJoinExec(
}
}
}
+
+ override def supportCodegen: Boolean = {
+ joinType.isInstanceOf[InnerLike]
+ }
+
+ override def inputRDDs(): Seq[RDD[InternalRow]] = {
+ streamed.asInstanceOf[CodegenSupport].inputRDDs()
+ }
+
+ override def needCopyResult: Boolean = true
+
+ override def doProduce(ctx: CodegenContext): String = {
+ streamed.asInstanceOf[CodegenSupport].produce(ctx, this)
+ }
+
+ override def doConsume(ctx: CodegenContext, input: Seq[ExprCode], row: ExprCode): String = {
+ joinType match {
+ case _: InnerLike => codegenInner(ctx, input)
+ case x =>
+ throw new IllegalArgumentException(
+ s"BroadcastNestedLoopJoin code-gen should not take $x as the JoinType")
+ }
+ }
+
+ /**
+ * Returns the variable name for [[Broadcast]] side.
+ */
+ private def prepareBroadcast(ctx: CodegenContext): String = {
+ // Create a name for broadcast side
+ val broadcastArray = broadcast.executeBroadcast[Array[InternalRow]]()
+ val broadcastTerm = ctx.addReferenceObj("broadcastTerm", broadcastArray)
+
+ // Inline mutable state since not many join operations in a task
+ ctx.addMutableState("InternalRow[]", "broadcastArray",
Review comment:
nit: `broadcastArray` -> `buildRows`, `buidlRowArray`, ...?
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastNestedLoopJoinExec.scala
##########
@@ -393,4 +394,65 @@ case class BroadcastNestedLoopJoinExec(
}
}
}
+
+ override def supportCodegen: Boolean = {
+ joinType.isInstanceOf[InnerLike]
+ }
+
+ override def inputRDDs(): Seq[RDD[InternalRow]] = {
+ streamed.asInstanceOf[CodegenSupport].inputRDDs()
+ }
+
+ override def needCopyResult: Boolean = true
Review comment:
always true? Is there a trade-off between the overheads of uniqueness checks and result copys?
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastNestedLoopJoinExec.scala
##########
@@ -393,4 +394,65 @@ case class BroadcastNestedLoopJoinExec(
}
}
}
+
+ override def supportCodegen: Boolean = {
+ joinType.isInstanceOf[InnerLike]
+ }
+
+ override def inputRDDs(): Seq[RDD[InternalRow]] = {
+ streamed.asInstanceOf[CodegenSupport].inputRDDs()
+ }
+
+ override def needCopyResult: Boolean = true
+
+ override def doProduce(ctx: CodegenContext): String = {
+ streamed.asInstanceOf[CodegenSupport].produce(ctx, this)
+ }
+
+ override def doConsume(ctx: CodegenContext, input: Seq[ExprCode], row: ExprCode): String = {
+ joinType match {
+ case _: InnerLike => codegenInner(ctx, input)
+ case x =>
+ throw new IllegalArgumentException(
+ s"BroadcastNestedLoopJoin code-gen should not take $x as the JoinType")
Review comment:
nit:
```
case _ =>
throw new IllegalArgumentException(
s"BroadcastNestedLoopJoin code-gen should not take $joinType as the JoinType")
```
?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] c21 commented on a change in pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
c21 commented on a change in pull request #31736:
URL: https://github.com/apache/spark/pull/31736#discussion_r588144023
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import org.apache.spark.sql.catalyst.expressions.{BindReferences, BoundReference}
+import org.apache.spark.sql.catalyst.expressions.codegen._
+import org.apache.spark.sql.catalyst.expressions.codegen.Block._
+import org.apache.spark.sql.catalyst.plans.InnerLike
+import org.apache.spark.sql.execution.{CodegenSupport, SparkPlan}
+
+/**
+ * An interface for those join physical operators that support codegen.
+ */
+trait JoinCodegenSupport extends CodegenSupport with BaseJoinExec {
+
+ /**
+ * Generate the (non-equi) condition used to filter joined rows. This is used in Inner, Left Semi
+ * and Left Anti joins.
+ */
+ protected def getJoinCondition(
+ ctx: CodegenContext,
+ input: Seq[ExprCode],
+ streamedPlan: SparkPlan,
+ buildPlan: SparkPlan): (String, String, Seq[ExprCode]) = {
+ val buildRow = ctx.freshName("buildRow")
+ val buildVars = genBuildSideVars(ctx, buildRow, buildPlan)
+ val checkCondition = if (condition.isDefined) {
+ val expr = condition.get
+ // evaluate the variables from build side that used by condition
Review comment:
> why don't we need to evaluate variables from the stream side?
@cloud-fan - `input` here is the generated code for streamed side. The code is generated below with `ctx.currentVars = input ++ buildVars`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-791277716
Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/40384/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-792411421
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/135847/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] c21 commented on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
c21 commented on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-793509585
Close & reopen PR to trigger test rerun as some transient unit test failure happened.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #31736:
URL: https://github.com/apache/spark/pull/31736#discussion_r588944070
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import org.apache.spark.sql.catalyst.expressions.{BindReferences, BoundReference}
+import org.apache.spark.sql.catalyst.expressions.codegen._
+import org.apache.spark.sql.catalyst.expressions.codegen.Block._
+import org.apache.spark.sql.catalyst.plans.InnerLike
+import org.apache.spark.sql.execution.{CodegenSupport, SparkPlan}
+
+/**
+ * An interface for those join physical operators that support codegen.
+ */
+trait JoinCodegenSupport extends CodegenSupport with BaseJoinExec {
+
+ /**
+ * Generate the (non-equi) condition used to filter joined rows. This is used in Inner, Left Semi
+ * and Left Anti joins.
+ */
+ protected def getJoinCondition(
+ ctx: CodegenContext,
+ input: Seq[ExprCode],
+ streamedPlan: SparkPlan,
+ buildPlan: SparkPlan): (String, String, Seq[ExprCode]) = {
+ val buildRow = ctx.freshName("buildRow")
+ val buildVars = genBuildSideVars(ctx, buildRow, buildPlan)
+ val checkCondition = if (condition.isDefined) {
+ val expr = condition.get
+ // evaluate the variables from build side that used by condition
Review comment:
The variables from stream side are be already evaluated before `consume` is called. So you don't need to include stream side variables again.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-791925030
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/40408/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-793476281
**[Test build #135901 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/135901/testReport)** for PR 31736 at commit [`6f6d511`](https://github.com/apache/spark/commit/6f6d511e571a2109cba4dcd6b31fa3def4e44e21).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] c21 commented on a change in pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
c21 commented on a change in pull request #31736:
URL: https://github.com/apache/spark/pull/31736#discussion_r587430064
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import org.apache.spark.sql.catalyst.expressions.{BindReferences, BoundReference}
+import org.apache.spark.sql.catalyst.expressions.codegen._
+import org.apache.spark.sql.catalyst.expressions.codegen.Block._
+import org.apache.spark.sql.catalyst.plans.InnerLike
+import org.apache.spark.sql.execution.{CodegenSupport, SparkPlan}
+
+/**
+ * An interface for those join physical operators that support codegen.
+ */
+trait JoinCodegenSupport extends CodegenSupport with BaseJoinExec {
+
+ /**
+ * Generate the (non-equi) condition used to filter joined rows. This is used in Inner, Left Semi
+ * and Left Anti joins.
+ */
+ protected def getJoinCondition(
+ ctx: CodegenContext,
+ input: Seq[ExprCode],
+ streamedPlan: SparkPlan,
+ buildPlan: SparkPlan): (String, String, Seq[ExprCode]) = {
+ val buildRow = ctx.freshName("buildRow")
+ val buildVars = genBuildSideVars(ctx, buildRow, buildPlan)
+ val checkCondition = if (condition.isDefined) {
+ val expr = condition.get
+ // evaluate the variables from build side that used by condition
+ val eval = evaluateRequiredVariables(buildPlan.output, buildVars, expr.references)
+ // filter the output via condition
+ ctx.currentVars = input ++ buildVars
+ val ev =
+ BindReferences.bindReference(expr, streamedPlan.output ++ buildPlan.output).genCode(ctx)
+ val skipRow = s"${ev.isNull} || !${ev.value}"
+ s"""
+ |$eval
+ |${ev.code}
+ |if (!($skipRow))
+ """.stripMargin
+ } else {
+ ""
+ }
+ (buildRow, checkCondition, buildVars)
+ }
+
+ /**
+ * Generates the code for variable of build side.
+ */
+ protected def genBuildSideVars(
Review comment:
This is copied from `HashJoin.scala: genBuildSideVars()`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #31736:
URL: https://github.com/apache/spark/pull/31736#discussion_r588153099
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastNestedLoopJoinExec.scala
##########
@@ -393,4 +394,65 @@ case class BroadcastNestedLoopJoinExec(
}
}
}
+
+ override def supportCodegen: Boolean = {
+ joinType.isInstanceOf[InnerLike]
+ }
+
+ override def inputRDDs(): Seq[RDD[InternalRow]] = {
+ streamed.asInstanceOf[CodegenSupport].inputRDDs()
+ }
+
+ override def needCopyResult: Boolean = true
+
+ override def doProduce(ctx: CodegenContext): String = {
+ streamed.asInstanceOf[CodegenSupport].produce(ctx, this)
+ }
+
+ override def doConsume(ctx: CodegenContext, input: Seq[ExprCode], row: ExprCode): String = {
+ joinType match {
+ case _: InnerLike => codegenInner(ctx, input)
+ case _ =>
+ throw new IllegalArgumentException(
+ s"BroadcastNestedLoopJoin code-gen should not take $joinType as the JoinType")
+ }
+ }
+
+ /**
+ * Returns the variable name for [[Broadcast]] side.
+ */
+ private def prepareBroadcast(ctx: CodegenContext): String = {
+ // Create a name for broadcast side
+ val broadcastArray = broadcast.executeBroadcast[Array[InternalRow]]()
+ val broadcastTerm = ctx.addReferenceObj("broadcastTerm", broadcastArray)
+
+ // Inline mutable state since not many join operations in a task
+ ctx.addMutableState("InternalRow[]", "buildRowArray",
+ v => s"$v = (InternalRow[]) $broadcastTerm.value();", forceInline = true)
+ }
+
+ private def codegenInner(ctx: CodegenContext, input: Seq[ExprCode]): String = {
+ val buildRowArrayTerm = prepareBroadcast(ctx)
+ val (buildRow, checkCondition, buildVars) = getJoinCondition(ctx, input, streamed, broadcast)
+
+ val resultVars = buildSide match {
+ case BuildLeft => buildVars ++ input
+ case BuildRight => input ++ buildVars
+ }
+ val arrayIndex = ctx.freshName("arrayIndex")
+ val numOutput = metricTerm(ctx, "numOutputRows")
+
+ s"""
+ |int $arrayIndex = 0;
+ |UnsafeRow $buildRow;
+ |while ($arrayIndex < $buildRowArrayTerm.length) {
Review comment:
`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.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #31736:
URL: https://github.com/apache/spark/pull/31736#discussion_r588155848
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import org.apache.spark.sql.catalyst.expressions.{BindReferences, BoundReference}
+import org.apache.spark.sql.catalyst.expressions.codegen._
+import org.apache.spark.sql.catalyst.expressions.codegen.Block._
+import org.apache.spark.sql.catalyst.plans.InnerLike
+import org.apache.spark.sql.execution.{CodegenSupport, SparkPlan}
+
+/**
+ * An interface for those join physical operators that support codegen.
+ */
+trait JoinCodegenSupport extends CodegenSupport with BaseJoinExec {
+
+ /**
+ * Generate the (non-equi) condition used to filter joined rows. This is used in Inner, Left Semi
+ * and Left Anti joins.
+ */
+ protected def getJoinCondition(
+ ctx: CodegenContext,
+ input: Seq[ExprCode],
+ streamedPlan: SparkPlan,
+ buildPlan: SparkPlan): (String, String, Seq[ExprCode]) = {
+ val buildRow = ctx.freshName("buildRow")
+ val buildVars = genBuildSideVars(ctx, buildRow, buildPlan)
+ val checkCondition = if (condition.isDefined) {
+ val expr = condition.get
+ // evaluate the variables from build side that used by condition
Review comment:
`evaluateRequiredVariables` is for performance purpose (eagerly evaluate some expressios and put them in vars), so this is not about correctness and is hard to test.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-793768312
thanks, merging to master!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] c21 commented on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
c21 commented on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-793454012
Addressed all comments for now, and the PR is ready for check again, thanks.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-793799689
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/135901/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] wangyum commented on a change in pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
wangyum commented on a change in pull request #31736:
URL: https://github.com/apache/spark/pull/31736#discussion_r587480753
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/JoinBenchmark.scala
##########
@@ -178,6 +191,7 @@ object JoinBenchmark extends SqlBasedBenchmark {
sortMergeJoin()
sortMergeJoinWithDuplicates()
shuffleHashJoin()
+ broadcastNestedLoopJoin()
Review comment:
Could we update the benchmark results([JoinBenchmark-results.txt](https://github.com/apache/spark/blob/f5118f81e395bde0cd8253dbef6a9e6455c3958a/sql/core/benchmarks/JoinBenchmark-results.txt) and [JoinBenchmark-jdk11-results.txt](https://github.com/apache/spark/blob/f5118f81e395bde0cd8253dbef6a9e6455c3958a/sql/core/benchmarks/JoinBenchmark-jdk11-results.txt))?
Generate result:
```
SPARK_GENERATE_BENCHMARK_FILES=1 build/sbt "sql/test:runMain org.apache.spark.sql.execution.benchmark.JoinBenchmark"
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] c21 commented on a change in pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
c21 commented on a change in pull request #31736:
URL: https://github.com/apache/spark/pull/31736#discussion_r589814250
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import org.apache.spark.sql.catalyst.expressions.{BindReferences, BoundReference}
+import org.apache.spark.sql.catalyst.expressions.codegen._
+import org.apache.spark.sql.catalyst.expressions.codegen.Block._
+import org.apache.spark.sql.catalyst.plans.InnerLike
+import org.apache.spark.sql.execution.{CodegenSupport, SparkPlan}
+
+/**
+ * An interface for those join physical operators that support codegen.
+ */
+trait JoinCodegenSupport extends CodegenSupport with BaseJoinExec {
+
+ /**
+ * Generate the (non-equi) condition used to filter joined rows. This is used in Inner, Left Semi
+ * and Left Anti joins.
+ */
+ protected def getJoinCondition(
+ ctx: CodegenContext,
+ input: Seq[ExprCode],
+ streamedPlan: SparkPlan,
+ buildPlan: SparkPlan): (String, String, Seq[ExprCode]) = {
+ val buildRow = ctx.freshName("buildRow")
+ val buildVars = genBuildSideVars(ctx, buildRow, buildPlan)
+ val checkCondition = if (condition.isDefined) {
+ val expr = condition.get
+ // evaluate the variables from build side that used by condition
Review comment:
After a second check, I think we don't need to evaluate variable from stream side here, as `BroadcastNestedLoopJoinExec`'s stream child node should already have evaluated them in [consume()](https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/WholeStageCodegenExec.scala#L177) before calling `BroadcastNestedLoopJoinExec.doConsume()`. As @viirya pointed out, given `BroadcastNestedLoopJoinExec.usedInputs()` is `references`, it should contain all attributes of `condition` from stream side.
Thanks @viirya for pointing this out and careful review. @cloud-fan wondering what do you think?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #31736:
URL: https://github.com/apache/spark/pull/31736#discussion_r589277164
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import org.apache.spark.sql.catalyst.expressions.{BindReferences, BoundReference}
+import org.apache.spark.sql.catalyst.expressions.codegen._
+import org.apache.spark.sql.catalyst.expressions.codegen.Block._
+import org.apache.spark.sql.catalyst.plans.InnerLike
+import org.apache.spark.sql.execution.{CodegenSupport, SparkPlan}
+
+/**
+ * An interface for those join physical operators that support codegen.
+ */
+trait JoinCodegenSupport extends CodegenSupport with BaseJoinExec {
+
+ /**
+ * Generate the (non-equi) condition used to filter joined rows. This is used in Inner, Left Semi
+ * and Left Anti joins.
+ */
+ protected def getJoinCondition(
+ ctx: CodegenContext,
+ input: Seq[ExprCode],
+ streamedPlan: SparkPlan,
+ buildPlan: SparkPlan): (String, String, Seq[ExprCode]) = {
+ val buildRow = ctx.freshName("buildRow")
+ val buildVars = genBuildSideVars(ctx, buildRow, buildPlan)
+ val checkCondition = if (condition.isDefined) {
+ val expr = condition.get
+ // evaluate the variables from build side that used by condition
Review comment:
For Filter or Project, we do some evaluation deferring. Normally `CodegenSupport.usedInputs` is set to `references`. `HashJoin` doesn't override it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-791277742
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/40384/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-792411421
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/135847/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-792364411
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/40429/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] c21 commented on a change in pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
c21 commented on a change in pull request #31736:
URL: https://github.com/apache/spark/pull/31736#discussion_r589976077
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import org.apache.spark.sql.catalyst.expressions.{BindReferences, BoundReference}
+import org.apache.spark.sql.catalyst.expressions.codegen._
+import org.apache.spark.sql.catalyst.expressions.codegen.Block._
+import org.apache.spark.sql.catalyst.plans.InnerLike
+import org.apache.spark.sql.execution.{CodegenSupport, SparkPlan}
+
+/**
+ * An interface for those join physical operators that support codegen.
+ */
+trait JoinCodegenSupport extends CodegenSupport with BaseJoinExec {
+
+ /**
+ * Generate the (non-equi) condition used to filter joined rows. This is used in Inner, Left Semi
+ * and Left Anti joins.
+ */
+ protected def getJoinCondition(
+ ctx: CodegenContext,
+ input: Seq[ExprCode],
+ streamedPlan: SparkPlan,
+ buildPlan: SparkPlan): (String, String, Seq[ExprCode]) = {
+ val buildRow = ctx.freshName("buildRow")
+ val buildVars = genBuildSideVars(ctx, buildRow, buildPlan)
+ val checkCondition = if (condition.isDefined) {
+ val expr = condition.get
+ // evaluate the variables from build side that used by condition
Review comment:
Updated to evaluate build side only. Same as before in `HashJoin.scala`.
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala
##########
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import org.apache.spark.sql.catalyst.expressions.{BindReferences, BoundReference}
+import org.apache.spark.sql.catalyst.expressions.codegen._
+import org.apache.spark.sql.catalyst.expressions.codegen.Block._
+import org.apache.spark.sql.catalyst.plans.InnerLike
+import org.apache.spark.sql.execution.{CodegenSupport, SparkPlan}
+
+/**
+ * An interface for those join physical operators that support codegen.
+ */
+trait JoinCodegenSupport extends CodegenSupport with BaseJoinExec {
+
+ /**
+ * Generate the (non-equi) condition used to filter joined rows.
+ * This is used in Inner, Left Semi and Left Anti joins.
+ *
+ * @return Tuple of variable name for row of build side, generated code for condition,
+ * and generated code for variables of build side.
+ */
+ protected def getJoinCondition(
+ ctx: CodegenContext,
+ input: Seq[ExprCode],
Review comment:
@cloud-fan - updated.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #31736:
URL: https://github.com/apache/spark/pull/31736#discussion_r588126729
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastNestedLoopJoinExec.scala
##########
@@ -393,4 +394,65 @@ case class BroadcastNestedLoopJoinExec(
}
}
}
+
+ override def supportCodegen: Boolean = {
+ joinType.isInstanceOf[InnerLike]
+ }
+
+ override def inputRDDs(): Seq[RDD[InternalRow]] = {
+ streamed.asInstanceOf[CodegenSupport].inputRDDs()
+ }
+
+ override def needCopyResult: Boolean = true
Review comment:
This is not a hash join and we don't know the uniqueness. I think it needs to be always true.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-791272002
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/135803/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-790606969
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/135753/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #31736:
URL: https://github.com/apache/spark/pull/31736#discussion_r588140194
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastNestedLoopJoinExec.scala
##########
@@ -393,4 +394,65 @@ case class BroadcastNestedLoopJoinExec(
}
}
}
+
+ override def supportCodegen: Boolean = {
+ joinType.isInstanceOf[InnerLike]
+ }
+
+ override def inputRDDs(): Seq[RDD[InternalRow]] = {
+ streamed.asInstanceOf[CodegenSupport].inputRDDs()
+ }
+
+ override def needCopyResult: Boolean = true
+
+ override def doProduce(ctx: CodegenContext): String = {
+ streamed.asInstanceOf[CodegenSupport].produce(ctx, this)
+ }
+
+ override def doConsume(ctx: CodegenContext, input: Seq[ExprCode], row: ExprCode): String = {
+ joinType match {
+ case _: InnerLike => codegenInner(ctx, input)
+ case _ =>
+ throw new IllegalArgumentException(
+ s"BroadcastNestedLoopJoin code-gen should not take $joinType as the JoinType")
+ }
+ }
+
+ /**
+ * Returns the variable name for [[Broadcast]] side.
+ */
+ private def prepareBroadcast(ctx: CodegenContext): String = {
+ // Create a name for broadcast side
+ val broadcastArray = broadcast.executeBroadcast[Array[InternalRow]]()
+ val broadcastTerm = ctx.addReferenceObj("broadcastTerm", broadcastArray)
+
+ // Inline mutable state since not many join operations in a task
+ ctx.addMutableState("InternalRow[]", "buildRowArray",
+ v => s"$v = (InternalRow[]) $broadcastTerm.value();", forceInline = true)
+ }
+
+ private def codegenInner(ctx: CodegenContext, input: Seq[ExprCode]): String = {
+ val buildRowArrayTerm = prepareBroadcast(ctx)
+ val (buildRow, checkCondition, buildVars) = getJoinCondition(ctx, input, streamed, broadcast)
+
+ val resultVars = buildSide match {
+ case BuildLeft => buildVars ++ input
+ case BuildRight => input ++ buildVars
+ }
+ val arrayIndex = ctx.freshName("arrayIndex")
+ val numOutput = metricTerm(ctx, "numOutputRows")
+
+ s"""
+ |int $arrayIndex = 0;
+ |UnsafeRow $buildRow;
+ |while ($arrayIndex < $buildRowArrayTerm.length) {
Review comment:
Let's make it more java style
```
for (int $arrayIndex = 0; $arrayIndex < $buildRowArrayTerm.length; $arrayIndex++) {
UnsafeRow $buildRow = (UnsafeRow) $buildRowArrayTerm[$arrayIndex];
...
}
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] c21 commented on a change in pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
c21 commented on a change in pull request #31736:
URL: https://github.com/apache/spark/pull/31736#discussion_r587861479
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/JoinBenchmark.scala
##########
@@ -178,6 +191,7 @@ object JoinBenchmark extends SqlBasedBenchmark {
sortMergeJoin()
sortMergeJoinWithDuplicates()
shuffleHashJoin()
+ broadcastNestedLoopJoin()
Review comment:
@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](https://github.com/apache/spark/blob/f5118f81e395bde0cd8253dbef6a9e6455c3958a/sql/core/benchmarks/JoinBenchmark-results.txt#L5). Should I try to run benchmark in the same type of machine? or this is not a hard requirement. cc @maropu as well.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-791968238
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/135826/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-791921615
Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/40408/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] c21 commented on a change in pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
c21 commented on a change in pull request #31736:
URL: https://github.com/apache/spark/pull/31736#discussion_r588861262
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/JoinBenchmark.scala
##########
@@ -178,6 +191,7 @@ object JoinBenchmark extends SqlBasedBenchmark {
sortMergeJoin()
sortMergeJoinWithDuplicates()
shuffleHashJoin()
+ broadcastNestedLoopJoin()
Review comment:
@wangyum, @maropu - updated the join benchmark results with same type machines as previous PRs. The description of PR has more details, thanks.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] c21 commented on a change in pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
c21 commented on a change in pull request #31736:
URL: https://github.com/apache/spark/pull/31736#discussion_r588144711
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import org.apache.spark.sql.catalyst.expressions.{BindReferences, BoundReference}
+import org.apache.spark.sql.catalyst.expressions.codegen._
+import org.apache.spark.sql.catalyst.expressions.codegen.Block._
+import org.apache.spark.sql.catalyst.plans.InnerLike
+import org.apache.spark.sql.execution.{CodegenSupport, SparkPlan}
+
+/**
+ * An interface for those join physical operators that support codegen.
+ */
+trait JoinCodegenSupport extends CodegenSupport with BaseJoinExec {
+
+ /**
+ * Generate the (non-equi) condition used to filter joined rows. This is used in Inner, Left Semi
+ * and Left Anti joins.
+ */
+ protected def getJoinCondition(
+ ctx: CodegenContext,
+ input: Seq[ExprCode],
+ streamedPlan: SparkPlan,
+ buildPlan: SparkPlan): (String, String, Seq[ExprCode]) = {
+ val buildRow = ctx.freshName("buildRow")
+ val buildVars = genBuildSideVars(ctx, buildRow, buildPlan)
+ val checkCondition = if (condition.isDefined) {
+ val expr = condition.get
+ // evaluate the variables from build side that used by condition
Review comment:
> Shall we do something similar in BroadcastNestedLoopJoinExec?
We don't have join key for streamed side in `BroadcastNestedLoopJoinExec`, i.e. non-equal joins.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-791277742
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/40384/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-792407299
**[Test build #135847 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/135847/testReport)** for PR 31736 at commit [`4d26e49`](https://github.com/apache/spark/commit/4d26e49114ea2090f8725b394b4c060ef706a04c).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-793596212
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/40483/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #31736:
URL: https://github.com/apache/spark/pull/31736#issuecomment-791908921
**[Test build #135826 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/135826/testReport)** for PR 31736 at commit [`4e5341f`](https://github.com/apache/spark/commit/4e5341f1e7f5e3bf6c849ec587be7274679ce71a).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] c21 commented on a change in pull request #31736: [SPARK-34620][SQL] Code-gen broadcast nested loop join (inner/cross)
Posted by GitBox <gi...@apache.org>.
c21 commented on a change in pull request #31736:
URL: https://github.com/apache/spark/pull/31736#discussion_r588796195
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import org.apache.spark.sql.catalyst.expressions.{BindReferences, BoundReference}
+import org.apache.spark.sql.catalyst.expressions.codegen._
+import org.apache.spark.sql.catalyst.expressions.codegen.Block._
+import org.apache.spark.sql.catalyst.plans.InnerLike
+import org.apache.spark.sql.execution.{CodegenSupport, SparkPlan}
+
+/**
+ * An interface for those join physical operators that support codegen.
+ */
+trait JoinCodegenSupport extends CodegenSupport with BaseJoinExec {
+
+ /**
+ * Generate the (non-equi) condition used to filter joined rows. This is used in Inner, Left Semi
+ * and Left Anti joins.
+ */
+ protected def getJoinCondition(
+ ctx: CodegenContext,
+ input: Seq[ExprCode],
+ streamedPlan: SparkPlan,
+ buildPlan: SparkPlan): (String, String, Seq[ExprCode]) = {
+ val buildRow = ctx.freshName("buildRow")
+ val buildVars = genBuildSideVars(ctx, buildRow, buildPlan)
+ val checkCondition = if (condition.isDefined) {
+ val expr = condition.get
+ // evaluate the variables from build side that used by condition
+ val eval = evaluateRequiredVariables(buildPlan.output, buildVars, expr.references)
+ // filter the output via condition
+ ctx.currentVars = input ++ buildVars
+ val ev =
+ BindReferences.bindReference(expr, streamedPlan.output ++ buildPlan.output).genCode(ctx)
+ val skipRow = s"${ev.isNull} || !${ev.value}"
+ s"""
+ |$eval
+ |${ev.code}
+ |if (!($skipRow))
+ """.stripMargin
+ } else {
+ ""
+ }
+ (buildRow, checkCondition, buildVars)
+ }
+
+ /**
+ * Generates the code for variable of build side.
+ */
+ protected def genBuildSideVars(
+ ctx: CodegenContext,
+ buildRow: String,
+ buildPlan: SparkPlan): Seq[ExprCode] = {
+ ctx.currentVars = null
+ ctx.INPUT_ROW = buildRow
+ buildPlan.output.zipWithIndex.map { case (a, i) =>
+ val ev = BoundReference(i, a.dataType, a.nullable).genCode(ctx)
+ if (joinType.isInstanceOf[InnerLike]) {
+ ev
+ } else {
+ // the variables are needed even there is no matched rows
Review comment:
@cloud-fan - I will clean up in another PR then.
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import org.apache.spark.sql.catalyst.expressions.{BindReferences, BoundReference}
+import org.apache.spark.sql.catalyst.expressions.codegen._
+import org.apache.spark.sql.catalyst.expressions.codegen.Block._
+import org.apache.spark.sql.catalyst.plans.InnerLike
+import org.apache.spark.sql.execution.{CodegenSupport, SparkPlan}
+
+/**
+ * An interface for those join physical operators that support codegen.
+ */
+trait JoinCodegenSupport extends CodegenSupport with BaseJoinExec {
+
+ /**
+ * Generate the (non-equi) condition used to filter joined rows. This is used in Inner, Left Semi
+ * and Left Anti joins.
Review comment:
@cloud-fan - makes sense, updated.
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastNestedLoopJoinExec.scala
##########
@@ -393,4 +394,65 @@ case class BroadcastNestedLoopJoinExec(
}
}
}
+
+ override def supportCodegen: Boolean = {
+ joinType.isInstanceOf[InnerLike]
+ }
+
+ override def inputRDDs(): Seq[RDD[InternalRow]] = {
+ streamed.asInstanceOf[CodegenSupport].inputRDDs()
+ }
+
+ override def needCopyResult: Boolean = true
+
+ override def doProduce(ctx: CodegenContext): String = {
+ streamed.asInstanceOf[CodegenSupport].produce(ctx, this)
+ }
+
+ override def doConsume(ctx: CodegenContext, input: Seq[ExprCode], row: ExprCode): String = {
+ joinType match {
+ case _: InnerLike => codegenInner(ctx, input)
+ case _ =>
+ throw new IllegalArgumentException(
+ s"BroadcastNestedLoopJoin code-gen should not take $joinType as the JoinType")
+ }
+ }
+
+ /**
+ * Returns the variable name for [[Broadcast]] side.
+ */
+ private def prepareBroadcast(ctx: CodegenContext): String = {
+ // Create a name for broadcast side
+ val broadcastArray = broadcast.executeBroadcast[Array[InternalRow]]()
+ val broadcastTerm = ctx.addReferenceObj("broadcastTerm", broadcastArray)
+
+ // Inline mutable state since not many join operations in a task
+ ctx.addMutableState("InternalRow[]", "buildRowArray",
+ v => s"$v = (InternalRow[]) $broadcastTerm.value();", forceInline = true)
+ }
+
+ private def codegenInner(ctx: CodegenContext, input: Seq[ExprCode]): String = {
+ val buildRowArrayTerm = prepareBroadcast(ctx)
+ val (buildRow, checkCondition, buildVars) = getJoinCondition(ctx, input, streamed, broadcast)
+
+ val resultVars = buildSide match {
+ case BuildLeft => buildVars ++ input
+ case BuildRight => input ++ buildVars
+ }
+ val arrayIndex = ctx.freshName("arrayIndex")
+ val numOutput = metricTerm(ctx, "numOutputRows")
+
+ s"""
+ |int $arrayIndex = 0;
+ |UnsafeRow $buildRow;
+ |while ($arrayIndex < $buildRowArrayTerm.length) {
Review comment:
@cloud-fan - updated.
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import org.apache.spark.sql.catalyst.expressions.{BindReferences, BoundReference}
+import org.apache.spark.sql.catalyst.expressions.codegen._
+import org.apache.spark.sql.catalyst.expressions.codegen.Block._
+import org.apache.spark.sql.catalyst.plans.InnerLike
+import org.apache.spark.sql.execution.{CodegenSupport, SparkPlan}
+
+/**
+ * An interface for those join physical operators that support codegen.
+ */
+trait JoinCodegenSupport extends CodegenSupport with BaseJoinExec {
+
+ /**
+ * Generate the (non-equi) condition used to filter joined rows. This is used in Inner, Left Semi
+ * and Left Anti joins.
+ */
+ protected def getJoinCondition(
+ ctx: CodegenContext,
+ input: Seq[ExprCode],
+ streamedPlan: SparkPlan,
+ buildPlan: SparkPlan): (String, String, Seq[ExprCode]) = {
+ val buildRow = ctx.freshName("buildRow")
+ val buildVars = genBuildSideVars(ctx, buildRow, buildPlan)
+ val checkCondition = if (condition.isDefined) {
+ val expr = condition.get
+ // evaluate the variables from build side that used by condition
Review comment:
Discussed offline, added code to eagerly evaluate streamed side as well. Thanks @cloud-fan for pointing it out.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org