You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by davies <gi...@git.apache.org> on 2016/02/18 08:43:21 UTC
[GitHub] spark pull request: [SPARK-13375] [SQL] generate sort merge join
GitHub user davies opened a pull request:
https://github.com/apache/spark/pull/11248
[SPARK-13375] [SQL] generate sort merge join
## What changes were proposed in this pull request?
Generates code for SortMergeJoin.
## How was the this patch tested?
Unit tests and manually tested with TPCDS Q72, which showed 3X performance improvements, but micro benchmark only show minor improvements, it may depends the distribution of data and number of columns.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/davies/spark gen_smj
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/11248.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #11248
----
commit 2a83ff5dafd3dbd57faa718bc11fd8f8b4ace751
Author: Davies Liu <da...@databricks.com>
Date: 2016-02-18T07:37:01Z
generate sort merge join
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13375] [SQL] generate sort merge join
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-185587877
Merged build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13375] [SQL] generate sort merge join
Posted by davies <gi...@git.apache.org>.
Github user davies commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-185885503
Here is part of the code generated from Q72
```
/* 160 */ private boolean findNextInnerJoinRows(
/* 161 */ scala.collection.Iterator leftIter,
/* 162 */ scala.collection.Iterator rightIter) {
/* 163 */ smj_leftRow = null;
/* 164 */ int comp = 0;
/* 165 */ while (smj_leftRow == null) {
/* 166 */ if (!leftIter.hasNext()) return false;
/* 167 */ smj_leftRow = (InternalRow) leftIter.next();
/* 168 */ /* input[7, int] */
/* 169 */ boolean smj_isNull = smj_leftRow.isNullAt(7);
/* 170 */ int smj_value = smj_isNull ? -1 : (smj_leftRow.getInt(7));
/* 171 */ if (smj_isNull) {
/* 172 */ smj_leftRow = null;
/* 173 */ continue;
/* 174 */ }
/* 175 */ if (!smj_matches.isEmpty()) {
/* 176 */ comp = 0;
/* 177 */ if (comp == 0) {
/* 178 */ comp = (smj_value > smj_value3 ? 1 : smj_value < smj_value3 ? -1 : 0);
/* 179 */ }
/* 180 */
/* 181 */ if (comp == 0) {
/* 182 */ return true;
/* 183 */ }
/* 184 */ smj_matches.clear();
/* 185 */ }
/* 186 */
/* 187 */ do {
/* 188 */ if (smj_rightRow == null) {
/* 189 */ if (!rightIter.hasNext()) {
/* 190 */ smj_value3 = smj_value;
/* 191 */
/* 192 */ return !smj_matches.isEmpty();
/* 193 */ }
/* 194 */ smj_rightRow = (InternalRow) rightIter.next();
/* 195 */ /* input[1, int] */
/* 196 */ boolean smj_isNull1 = smj_rightRow.isNullAt(1);
/* 197 */ int smj_value1 = smj_isNull1 ? -1 : (smj_rightRow.getInt(1));
/* 198 */ if (smj_isNull1) {
/* 199 */ smj_rightRow = null;
/* 200 */ continue;
/* 201 */ }
/* 202 */
/* 203 */ smj_value2 = smj_value1;
/* 204 */
/* 205 */ }
/* 206 */
/* 207 */ comp = 0;
/* 208 */ if (comp == 0) {
/* 209 */ comp = (smj_value > smj_value2 ? 1 : smj_value < smj_value2 ? -1 : 0);
/* 210 */ }
/* 211 */
/* 212 */ if (comp > 0) {
/* 213 */ smj_rightRow = null;
/* 214 */ } else if (comp < 0) {
/* 215 */ if (!smj_matches.isEmpty()) {
/* 216 */ smj_value3 = smj_value;
/* 217 */
/* 218 */ return true;
/* 219 */ }
/* 220 */ smj_leftRow = null;
/* 221 */ } else {
/* 222 */ smj_matches.add(smj_rightRow.copy());
/* 223 */ smj_rightRow = null;;
/* 224 */ }
/* 225 */ } while (smj_leftRow != null);
/* 226 */ }
/* 227 */ return false; // unreachable
/* 228 */ }
/* 229 */
/* 230 */ protected void processNext() throws java.io.IOException {
/* 231 */ while (findNextInnerJoinRows(smj_leftInput, smj_rightInput)) {
/* 232 */ int smj_size = smj_matches.size();
/* 233 */ boolean smj_loaded = false;
/* 234 */
/* 235 */ smj_isNull7 = smj_leftRow.isNullAt(5);
/* 236 */ smj_value9 = smj_isNull7 ? -1 : (smj_leftRow.getInt(5));
/* 237 */
/* 238 */ for (int smj_i = 0; smj_i < smj_size; smj_i ++) {
/* 239 */ InternalRow smj_rightRow1 = (InternalRow) smj_matches.get(smj_i);
/* 240 */ /* input[3, int] */
/* 241 */ boolean smj_isNull13 = smj_rightRow1.isNullAt(3);
/* 242 */ int smj_value15 = smj_isNull13 ? -1 : (smj_rightRow1.getInt(3));
/* 243 */ /* (input[11, int] < input[5, int]) */
/* 244 */ boolean smj_isNull14 = true;
/* 245 */ boolean smj_value16 = false;
/* 246 */
/* 247 */ if (!smj_isNull13) {
/* 248 */ if (!smj_isNull7) {
/* 249 */ smj_isNull14 = false; // resultCode could change nullability.
/* 250 */ smj_value16 = smj_value15 < smj_value9;
/* 251 */
/* 252 */ }
/* 253 */
/* 254 */ }
/* 255 */ if (smj_isNull14 || !smj_value16) continue;
/* 256 */ if (!smj_loaded) {
/* 257 */ smj_loaded = true;
/* 258 */
/* 259 */ smj_isNull2 = smj_leftRow.isNullAt(0);
/* 260 */ smj_value4 = smj_isNull2 ? -1 : (smj_leftRow.getInt(0));
/* 261 */
/* 262 */ smj_isNull3 = smj_leftRow.isNullAt(1);
/* 263 */ smj_value5 = smj_isNull3 ? -1 : (smj_leftRow.getInt(1));
/* 264 */
/* 265 */ smj_isNull4 = smj_leftRow.isNullAt(2);
/* 266 */ smj_value6 = smj_isNull4 ? -1 : (smj_leftRow.getInt(2));
/* 267 */
/* 268 */ smj_isNull5 = smj_leftRow.isNullAt(3);
/* 269 */ smj_value7 = smj_isNull5 ? -1 : (smj_leftRow.getInt(3));
/* 270 */
/* 271 */ smj_isNull6 = smj_leftRow.isNullAt(4);
/* 272 */ smj_value8 = smj_isNull6 ? -1 : (smj_leftRow.getInt(4));
/* 273 */
/* 274 */ smj_isNull8 = smj_leftRow.isNullAt(6);
/* 275 */ smj_value10 = smj_isNull8 ? -1 : (smj_leftRow.getInt(6));
/* 276 */
/* 277 */ smj_isNull9 = smj_leftRow.isNullAt(7);
/* 278 */ smj_value11 = smj_isNull9 ? -1 : (smj_leftRow.getInt(7));
/* 279 */
/* 280 */ }
/* 281 */ /* input[0, int] */
/* 282 */ boolean smj_isNull10 = smj_rightRow1.isNullAt(0);
/* 283 */ int smj_value12 = smj_isNull10 ? -1 : (smj_rightRow1.getInt(0));
/* 284 */ /* input[1, int] */
/* 285 */ boolean smj_isNull11 = smj_rightRow1.isNullAt(1);
/* 286 */ int smj_value13 = smj_isNull11 ? -1 : (smj_rightRow1.getInt(1));
/* 287 */ /* input[2, int] */
/* 288 */ boolean smj_isNull12 = smj_rightRow1.isNullAt(2);
/* 289 */ int smj_value14 = smj_isNull12 ? -1 : (smj_rightRow1.getInt(2));
/* 290 */ smj_metricValue.add(1);
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13375] [SQL] generate sort merge join
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-185840569
**[Test build #51490 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/51490/consoleFull)** for PR 11248 at commit [`8898957`](https://github.com/apache/spark/commit/8898957bd49c73216637f1b7597046a1e2090f3d).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13373] [SQL] generate sort merge join
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/11248#discussion_r53718141
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoin.scala ---
@@ -125,6 +126,246 @@ case class SortMergeJoin(
}.toScala
}
}
+
+ override def upstreams(): Seq[RDD[InternalRow]] = {
+ left.execute() :: right.execute() :: Nil
+ }
+
+ private def createJoinKey(
+ ctx: CodegenContext,
+ row: String,
+ keys: Seq[Expression],
+ input: Seq[Attribute]): Seq[ExprCode] = {
+ ctx.INPUT_ROW = row
+ keys.map(BindReferences.bindReference(_, input).gen(ctx))
+ }
+
+ private def copyKeys(ctx: CodegenContext, vars: Seq[ExprCode]): Seq[ExprCode] = {
+ vars.zipWithIndex.map { case (ev, i) =>
+ val value = ctx.freshName("value")
+ ctx.addMutableState(ctx.javaType(leftKeys(i).dataType), value, "")
+ val code =
+ s"""
+ |$value = ${ev.value};
+ """.stripMargin
+ ExprCode(code, "false", value)
+ }
+ }
+
+ private def genComparision(ctx: CodegenContext, a: Seq[ExprCode], b: Seq[ExprCode]): String = {
+ val comparisons = a.zip(b).zipWithIndex.map { case ((l, r), i) =>
+ s"""
+ |if (comp == 0) {
+ | comp = ${ctx.genComp(leftKeys(i).dataType, l.value, r.value)};
+ |}
+ """.stripMargin.trim
+ }
+ s"""
+ |comp = 0;
+ |${comparisons.mkString("\n")}
+ """.stripMargin
+ }
+
+ /**
+ * Generate a function to scan both left and right to find a match, returns the term for
+ * matched one row from left side and buffered rows from right side.
+ */
+ private def genScanner(ctx: CodegenContext): (String, String) = {
+ // Create class member for next row from both sides.
+ val leftRow = ctx.freshName("leftRow")
+ ctx.addMutableState("InternalRow", leftRow, "")
+ val rightRow = ctx.freshName("rightRow")
+ ctx.addMutableState("InternalRow", rightRow, s"$rightRow = null;")
+
+ // Create variables for join keys from both sides.
+ val leftKeyVars = createJoinKey(ctx, leftRow, leftKeys, left.output)
+ val leftAnyNull = leftKeyVars.map(_.isNull).mkString(" || ")
+ val rightKeyTmpVars = createJoinKey(ctx, rightRow, rightKeys, right.output)
+ val rightAnyNull = rightKeyTmpVars.map(_.isNull).mkString(" || ")
+ // Copy the right key as class members so they could be used in next function call.
+ val rightKeyVars = copyKeys(ctx, rightKeyTmpVars)
+
+ // A list to hold all matched rows from right side.
+ val matches = ctx.freshName("matches")
+ val clsName = classOf[java.util.ArrayList[InternalRow]].getName
+ ctx.addMutableState(clsName, matches, s"$matches = new $clsName();")
+ // Copy the left keys as class members so they could be used in next function call.
+ val matchedKeyVars = copyKeys(ctx, leftKeyVars)
+
+ ctx.addNewFunction("findNextInnerJoinRows",
+ s"""
+ |private boolean findNextInnerJoinRows(
+ | scala.collection.Iterator leftIter,
+ | scala.collection.Iterator rightIter) {
+ | $leftRow = null;
+ | int comp = 0;
+ | while ($leftRow == null) {
+ | if (!leftIter.hasNext()) return false;
+ | $leftRow = (InternalRow) leftIter.next();
+ | ${leftKeyVars.map(_.code).mkString("\n")}
+ | if ($leftAnyNull) {
+ | $leftRow = null;
+ | continue;
+ | }
+ | if (!$matches.isEmpty()) {
+ | ${genComparision(ctx, leftKeyVars, matchedKeyVars)}
+ | if (comp == 0) {
+ | return true;
+ | }
+ | $matches.clear();
+ | }
+ |
+ | do {
+ | if ($rightRow == null) {
+ | if (!rightIter.hasNext()) {
+ | ${matchedKeyVars.map(_.code).mkString("\n")}
+ | return !$matches.isEmpty();
+ | }
+ | $rightRow = (InternalRow) rightIter.next();
+ | ${rightKeyTmpVars.map(_.code).mkString("\n")}
+ | if ($rightAnyNull) {
+ | $rightRow = null;
+ | continue;
+ | }
+ | ${rightKeyVars.map(_.code).mkString("\n")}
+ | }
+ | ${genComparision(ctx, leftKeyVars, rightKeyVars)}
+ | if (comp > 0) {
+ | $rightRow = null;
+ | } else if (comp < 0) {
+ | if (!$matches.isEmpty()) {
+ | ${matchedKeyVars.map(_.code).mkString("\n")}
+ | return true;
+ | }
+ | $leftRow = null;
+ | } else {
+ | $matches.add($rightRow.copy());
+ | $rightRow = null;;
+ | }
+ | } while ($leftRow != null);
+ | }
+ | return false; // unreachable
+ |}
+ """.stripMargin)
+
+ (leftRow, matches)
+ }
+
+ /**
+ * Creates variables for left part of result row.
+ *
+ * In order to defer the access after condition and also only access once in the loop,
+ * the variables should be declared separately from accessing the columns, we can't use the
+ * codegen of BoundReference here.
+ */
+ private def createLeftVars(ctx: CodegenContext, leftRow: String): Seq[ExprCode] = {
+ ctx.INPUT_ROW = leftRow
+ left.output.zipWithIndex.map { case (a, i) =>
+ val value = ctx.freshName("value")
+ val valueCode = ctx.getValue(leftRow, a.dataType, i.toString)
+ // declare it as class member, so we can access the column before or in the loop.
+ ctx.addMutableState(ctx.javaType(a.dataType), value, "")
+ if (a.nullable) {
+ val isNull = ctx.freshName("isNull")
+ ctx.addMutableState("boolean", isNull, "")
+ val code =
+ s"""
+ |$isNull = $leftRow.isNullAt($i);
+ |$value = $isNull ? ${ctx.defaultValue(a.dataType)} : ($valueCode);
+ """.stripMargin
+ ExprCode(code, isNull, value)
+ } else {
+ ExprCode(s"$value = $valueCode;", "false", value)
+ }
+ }
+ }
+
+ /**
+ * Creates the variables for right part of result row, using BoundReference, since the right
+ * part are accessed inside the loop.
+ */
+ private def createRightVar(ctx: CodegenContext, rightRow: String): Seq[ExprCode] = {
+ ctx.INPUT_ROW = rightRow
+ right.output.zipWithIndex.map { case (a, i) =>
+ BoundReference(i, a.dataType, a.nullable).gen(ctx)
+ }
+ }
+
+ /**
+ * Splits variables based on whether it's used by condition or not, returns the code to create
+ * these variables before the condition and after the condition.
+ *
+ * Only a few columns are used by condition, then we can skip the accessing of those columns
+ * that are not used by condition also filtered out by condition.
+ */
+ private def splitVarsByCondition(
+ attributes: Seq[Attribute],
+ variables: Seq[ExprCode]): (String, String) = {
+ if (condition.isDefined) {
+ val condRefs = condition.get.references
+ val (used, notUsed) = attributes.zip(variables).partition{ case (a, ev) =>
+ condRefs.contains(a)
+ }
+ val beforeCond = used.map(_._2.code).mkString("\n")
+ val afterCond = notUsed.map(_._2.code).mkString("\n")
+ (beforeCond, afterCond)
+ } else {
+ (variables.map(_.code).mkString("\n"), "")
+ }
+ }
+
+ override def doProduce(ctx: CodegenContext): String = {
+ val leftInput = ctx.freshName("leftInput")
+ ctx.addMutableState("scala.collection.Iterator", leftInput, s"$leftInput = inputs[0];")
+ val rightInput = ctx.freshName("rightInput")
+ ctx.addMutableState("scala.collection.Iterator", rightInput, s"$rightInput = inputs[1];")
+
+ val (leftRow, matches) = genScanner(ctx)
+
+ // Create variables for row from both sides.
+ val leftVars = createLeftVars(ctx, leftRow)
+ val rightRow = ctx.freshName("rightRow")
+ val rightVars = createRightVar(ctx, rightRow)
+ val resultVars = leftVars ++ rightVars
+
+ // Check condition
+ ctx.currentVars = resultVars
+ val cond = if (condition.isDefined) {
+ BindReferences.bindReference(condition.get, output).gen(ctx)
+ } else {
+ ExprCode("", "false", "true")
+ }
+ // Split the code of creating variables based on whether it's used by condition or not.
+ val loaded = ctx.freshName("loaded")
+ val (leftBefore, leftAfter) = splitVarsByCondition(left.output, leftVars)
+ val (rightBefore, rightAfter) = splitVarsByCondition(right.output, rightVars)
+
+
+ val size = ctx.freshName("size")
+ val i = ctx.freshName("i")
+ val numOutput = metricTerm(ctx, "numOutputRows")
+ s"""
+ |while (findNextInnerJoinRows($leftInput, $rightInput)) {
+ | int $size = $matches.size();
+ | boolean $loaded = false;
+ | $leftBefore
+ | for (int $i = 0; $i < $size; $i ++) {
+ | InternalRow $rightRow = (InternalRow) $matches.get($i);
+ | $rightBefore
+ | ${cond.code}
+ | if (${cond.isNull} || !${cond.value}) continue;
+ | if (!$loaded) {
+ | $loaded = true;
+ | $leftAfter
+ | }
+ | $rightAfter
+ | $numOutput.add(1);
--- End diff --
see https://github.com/apache/spark/pull/11170
It costs 0.1 - 0.2 nanoseconds per rows.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13375] [SQL] generate sort merge join
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-185955164
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13373] [SQL] generate sort merge join
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-187840890
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/51771/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13375] [SQL] generate sort merge join
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-185838161
**[Test build #2544 has started](https://amplab.cs.berkeley.edu/jenkins/job/NewSparkPullRequestBuilder/2544/consoleFull)** for PR 11248 at commit [`8898957`](https://github.com/apache/spark/commit/8898957bd49c73216637f1b7597046a1e2090f3d).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13375] [SQL] generate sort merge join
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-185954194
**[Test build #51495 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/51495/consoleFull)** for PR 11248 at commit [`09a97d4`](https://github.com/apache/spark/commit/09a97d42e17699d11bc5f372e9e0aaef780f2c66).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13375] [SQL] generate sort merge join
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-186143129
@davies this is linked to the wrong JIRA
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13375] [SQL] generate sort merge join
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-185619610
**[Test build #51477 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/51477/consoleFull)** for PR 11248 at commit [`3c8b35c`](https://github.com/apache/spark/commit/3c8b35ce0f44f64da4d7fb4d12d8e28448744599).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13375] [SQL] generate sort merge join
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-185952297
**[Test build #2545 has started](https://amplab.cs.berkeley.edu/jenkins/job/NewSparkPullRequestBuilder/2545/consoleFull)** for PR 11248 at commit [`09a97d4`](https://github.com/apache/spark/commit/09a97d42e17699d11bc5f372e9e0aaef780f2c66).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13375] [SQL] generate sort merge join
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-185587548
**[Test build #51471 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/51471/consoleFull)** for PR 11248 at commit [`2a83ff5`](https://github.com/apache/spark/commit/2a83ff5dafd3dbd57faa718bc11fd8f8b4ace751).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13373] [SQL] generate sort merge join
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/11248
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13375] [SQL] generate sort merge join
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-185665675
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/51478/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13373] [SQL] generate sort merge join
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-187840884
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13375] [SQL] generate sort merge join
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-185665669
Merged build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13373] [SQL] generate sort merge join
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on a diff in the pull request:
https://github.com/apache/spark/pull/11248#discussion_r53710181
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/WholeStageCodegen.scala ---
@@ -276,19 +286,32 @@ case class WholeStageCodegen(plan: CodegenSupport, children: Seq[SparkPlan])
// println(s"${CodeFormatter.format(cleanedSource)}")
CodeGenerator.compile(cleanedSource)
- plan.upstream().mapPartitions { iter =>
-
- val clazz = CodeGenerator.compile(source)
- val buffer = clazz.generate(references).asInstanceOf[BufferedRowIterator]
- buffer.setInput(iter)
- new Iterator[InternalRow] {
- override def hasNext: Boolean = buffer.hasNext
- override def next: InternalRow = buffer.next()
+ val rdds = plan.upstreams()
+ if (rdds.length == 1) {
+ rdds.head.mapPartitions { iter =>
+ val clazz = CodeGenerator.compile(cleanedSource)
+ val buffer = clazz.generate(references).asInstanceOf[BufferedRowIterator]
+ buffer.init(Array(iter))
+ new Iterator[InternalRow] {
+ override def hasNext: Boolean = buffer.hasNext
+ override def next: InternalRow = buffer.next()
+ }
+ }
+ } else {
+ // Right now, we support up to two upstreams.
--- End diff --
Add an assert
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13375] [SQL] generate sort merge join
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-185950526
**[Test build #51490 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/51490/consoleFull)** for PR 11248 at commit [`8898957`](https://github.com/apache/spark/commit/8898957bd49c73216637f1b7597046a1e2090f3d).
* This patch **fails from timeout after a configured wait of \`250m\`**.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13373] [SQL] generate sort merge join
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on a diff in the pull request:
https://github.com/apache/spark/pull/11248#discussion_r53858750
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/execution/BenchmarkWholeStageCodegen.scala ---
@@ -38,6 +38,7 @@ import org.apache.spark.util.Benchmark
class BenchmarkWholeStageCodegen extends SparkFunSuite {
lazy val conf = new SparkConf().setMaster("local[1]").setAppName("benchmark")
.set("spark.sql.shuffle.partitions", "1")
+ .set("spark.sql.autoBroadcastJoinThreshold", "0")
--- End diff --
Should this be set inside the sort merge join benchmark?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13375] [SQL] generate sort merge join
Posted by davies <gi...@git.apache.org>.
Github user davies commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-185836844
@hvanhovell I think the purpose of micro benchmark is to show the baseline cost of each operator, the result will be very different from case to case, it's hard to say which one is more realistic.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13375] [SQL] generate sort merge join
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-185951458
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/51490/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13375] [SQL] generate sort merge join
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-185951451
Merged build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13375] [SQL] generate sort merge join
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-185624806
**[Test build #51478 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/51478/consoleFull)** for PR 11248 at commit [`b3669f2`](https://github.com/apache/spark/commit/b3669f293fd904b716e2dc919375319519c0894f).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13373] [SQL] generate sort merge join
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/11248#discussion_r53718033
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoin.scala ---
@@ -125,6 +126,246 @@ case class SortMergeJoin(
}.toScala
}
}
+
+ override def upstreams(): Seq[RDD[InternalRow]] = {
+ left.execute() :: right.execute() :: Nil
+ }
+
+ private def createJoinKey(
+ ctx: CodegenContext,
+ row: String,
+ keys: Seq[Expression],
+ input: Seq[Attribute]): Seq[ExprCode] = {
+ ctx.INPUT_ROW = row
+ keys.map(BindReferences.bindReference(_, input).gen(ctx))
+ }
+
+ private def copyKeys(ctx: CodegenContext, vars: Seq[ExprCode]): Seq[ExprCode] = {
+ vars.zipWithIndex.map { case (ev, i) =>
+ val value = ctx.freshName("value")
+ ctx.addMutableState(ctx.javaType(leftKeys(i).dataType), value, "")
+ val code =
+ s"""
+ |$value = ${ev.value};
+ """.stripMargin
+ ExprCode(code, "false", value)
+ }
+ }
+
+ private def genComparision(ctx: CodegenContext, a: Seq[ExprCode], b: Seq[ExprCode]): String = {
+ val comparisons = a.zip(b).zipWithIndex.map { case ((l, r), i) =>
+ s"""
+ |if (comp == 0) {
+ | comp = ${ctx.genComp(leftKeys(i).dataType, l.value, r.value)};
+ |}
+ """.stripMargin.trim
+ }
+ s"""
+ |comp = 0;
+ |${comparisons.mkString("\n")}
+ """.stripMargin
+ }
+
+ /**
+ * Generate a function to scan both left and right to find a match, returns the term for
+ * matched one row from left side and buffered rows from right side.
+ */
+ private def genScanner(ctx: CodegenContext): (String, String) = {
+ // Create class member for next row from both sides.
+ val leftRow = ctx.freshName("leftRow")
+ ctx.addMutableState("InternalRow", leftRow, "")
+ val rightRow = ctx.freshName("rightRow")
+ ctx.addMutableState("InternalRow", rightRow, s"$rightRow = null;")
+
+ // Create variables for join keys from both sides.
+ val leftKeyVars = createJoinKey(ctx, leftRow, leftKeys, left.output)
+ val leftAnyNull = leftKeyVars.map(_.isNull).mkString(" || ")
+ val rightKeyTmpVars = createJoinKey(ctx, rightRow, rightKeys, right.output)
+ val rightAnyNull = rightKeyTmpVars.map(_.isNull).mkString(" || ")
+ // Copy the right key as class members so they could be used in next function call.
+ val rightKeyVars = copyKeys(ctx, rightKeyTmpVars)
+
+ // A list to hold all matched rows from right side.
+ val matches = ctx.freshName("matches")
+ val clsName = classOf[java.util.ArrayList[InternalRow]].getName
+ ctx.addMutableState(clsName, matches, s"$matches = new $clsName();")
+ // Copy the left keys as class members so they could be used in next function call.
+ val matchedKeyVars = copyKeys(ctx, leftKeyVars)
+
+ ctx.addNewFunction("findNextInnerJoinRows",
+ s"""
+ |private boolean findNextInnerJoinRows(
+ | scala.collection.Iterator leftIter,
+ | scala.collection.Iterator rightIter) {
+ | $leftRow = null;
+ | int comp = 0;
+ | while ($leftRow == null) {
+ | if (!leftIter.hasNext()) return false;
+ | $leftRow = (InternalRow) leftIter.next();
+ | ${leftKeyVars.map(_.code).mkString("\n")}
+ | if ($leftAnyNull) {
+ | $leftRow = null;
+ | continue;
+ | }
+ | if (!$matches.isEmpty()) {
+ | ${genComparision(ctx, leftKeyVars, matchedKeyVars)}
+ | if (comp == 0) {
+ | return true;
+ | }
+ | $matches.clear();
+ | }
+ |
+ | do {
+ | if ($rightRow == null) {
+ | if (!rightIter.hasNext()) {
+ | ${matchedKeyVars.map(_.code).mkString("\n")}
+ | return !$matches.isEmpty();
+ | }
+ | $rightRow = (InternalRow) rightIter.next();
+ | ${rightKeyTmpVars.map(_.code).mkString("\n")}
+ | if ($rightAnyNull) {
+ | $rightRow = null;
+ | continue;
+ | }
+ | ${rightKeyVars.map(_.code).mkString("\n")}
+ | }
+ | ${genComparision(ctx, leftKeyVars, rightKeyVars)}
+ | if (comp > 0) {
+ | $rightRow = null;
+ | } else if (comp < 0) {
+ | if (!$matches.isEmpty()) {
+ | ${matchedKeyVars.map(_.code).mkString("\n")}
+ | return true;
+ | }
+ | $leftRow = null;
+ | } else {
+ | $matches.add($rightRow.copy());
+ | $rightRow = null;;
+ | }
+ | } while ($leftRow != null);
+ | }
+ | return false; // unreachable
+ |}
+ """.stripMargin)
+
+ (leftRow, matches)
+ }
+
+ /**
+ * Creates variables for left part of result row.
+ *
+ * In order to defer the access after condition and also only access once in the loop,
+ * the variables should be declared separately from accessing the columns, we can't use the
+ * codegen of BoundReference here.
+ */
+ private def createLeftVars(ctx: CodegenContext, leftRow: String): Seq[ExprCode] = {
+ ctx.INPUT_ROW = leftRow
+ left.output.zipWithIndex.map { case (a, i) =>
+ val value = ctx.freshName("value")
+ val valueCode = ctx.getValue(leftRow, a.dataType, i.toString)
+ // declare it as class member, so we can access the column before or in the loop.
+ ctx.addMutableState(ctx.javaType(a.dataType), value, "")
+ if (a.nullable) {
+ val isNull = ctx.freshName("isNull")
+ ctx.addMutableState("boolean", isNull, "")
+ val code =
+ s"""
+ |$isNull = $leftRow.isNullAt($i);
+ |$value = $isNull ? ${ctx.defaultValue(a.dataType)} : ($valueCode);
+ """.stripMargin
+ ExprCode(code, isNull, value)
+ } else {
+ ExprCode(s"$value = $valueCode;", "false", value)
+ }
+ }
+ }
+
+ /**
+ * Creates the variables for right part of result row, using BoundReference, since the right
+ * part are accessed inside the loop.
+ */
+ private def createRightVar(ctx: CodegenContext, rightRow: String): Seq[ExprCode] = {
+ ctx.INPUT_ROW = rightRow
+ right.output.zipWithIndex.map { case (a, i) =>
+ BoundReference(i, a.dataType, a.nullable).gen(ctx)
+ }
+ }
+
+ /**
+ * Splits variables based on whether it's used by condition or not, returns the code to create
+ * these variables before the condition and after the condition.
+ *
+ * Only a few columns are used by condition, then we can skip the accessing of those columns
+ * that are not used by condition also filtered out by condition.
+ */
+ private def splitVarsByCondition(
+ attributes: Seq[Attribute],
+ variables: Seq[ExprCode]): (String, String) = {
+ if (condition.isDefined) {
+ val condRefs = condition.get.references
+ val (used, notUsed) = attributes.zip(variables).partition{ case (a, ev) =>
+ condRefs.contains(a)
+ }
+ val beforeCond = used.map(_._2.code).mkString("\n")
+ val afterCond = notUsed.map(_._2.code).mkString("\n")
+ (beforeCond, afterCond)
+ } else {
+ (variables.map(_.code).mkString("\n"), "")
+ }
+ }
+
+ override def doProduce(ctx: CodegenContext): String = {
+ val leftInput = ctx.freshName("leftInput")
+ ctx.addMutableState("scala.collection.Iterator", leftInput, s"$leftInput = inputs[0];")
+ val rightInput = ctx.freshName("rightInput")
+ ctx.addMutableState("scala.collection.Iterator", rightInput, s"$rightInput = inputs[1];")
+
+ val (leftRow, matches) = genScanner(ctx)
+
+ // Create variables for row from both sides.
+ val leftVars = createLeftVars(ctx, leftRow)
+ val rightRow = ctx.freshName("rightRow")
+ val rightVars = createRightVar(ctx, rightRow)
+ val resultVars = leftVars ++ rightVars
+
+ // Check condition
+ ctx.currentVars = resultVars
+ val cond = if (condition.isDefined) {
+ BindReferences.bindReference(condition.get, output).gen(ctx)
+ } else {
+ ExprCode("", "false", "true")
+ }
+ // Split the code of creating variables based on whether it's used by condition or not.
+ val loaded = ctx.freshName("loaded")
+ val (leftBefore, leftAfter) = splitVarsByCondition(left.output, leftVars)
+ val (rightBefore, rightAfter) = splitVarsByCondition(right.output, rightVars)
+
+
+ val size = ctx.freshName("size")
+ val i = ctx.freshName("i")
+ val numOutput = metricTerm(ctx, "numOutputRows")
+ s"""
+ |while (findNextInnerJoinRows($leftInput, $rightInput)) {
+ | int $size = $matches.size();
+ | boolean $loaded = false;
+ | $leftBefore
+ | for (int $i = 0; $i < $size; $i ++) {
+ | InternalRow $rightRow = (InternalRow) $matches.get($i);
+ | $rightBefore
+ | ${cond.code}
+ | if (${cond.isNull} || !${cond.value}) continue;
+ | if (!$loaded) {
+ | $loaded = true;
+ | $leftAfter
+ | }
+ | $rightAfter
+ | $numOutput.add(1);
+ | ${consume(ctx, resultVars)}
+ | }
+ | if (shouldStop()) return;
--- End diff --
I think we can't break the loop in last operator, because some loops in the middle can't be interrupted. We do that in the beginning, then switch to current way since we generated joins.
for example:
```
while (input.hasNext) {
for (i <- matches) {
// can't return at here
}
// can only return at here
}
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13375] [SQL] generate sort merge join
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-185654038
**[Test build #51477 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/51477/consoleFull)** for PR 11248 at commit [`3c8b35c`](https://github.com/apache/spark/commit/3c8b35ce0f44f64da4d7fb4d12d8e28448744599).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13375] [SQL] generate sort merge join
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-185837581
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/51489/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13373] [SQL] generate sort merge join
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-186723731
**[Test build #51607 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/51607/consoleFull)** for PR 11248 at commit [`55148d4`](https://github.com/apache/spark/commit/55148d442b5d181d03d43687972ebc5bd30f18b6).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13375] [SQL] generate sort merge join
Posted by hvanhovell <gi...@git.apache.org>.
Github user hvanhovell commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-185827032
Shouldn't we add a more realistic benchmark then?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13375] [SQL] generate sort merge join
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-185905076
**[Test build #2544 has finished](https://amplab.cs.berkeley.edu/jenkins/job/NewSparkPullRequestBuilder/2544/consoleFull)** for PR 11248 at commit [`8898957`](https://github.com/apache/spark/commit/8898957bd49c73216637f1b7597046a1e2090f3d).
* This patch **fails PySpark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13373] [SQL] generate sort merge join
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-187938391
LGTM
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13373] [SQL] generate sort merge join
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-186747421
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/51607/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13373] [SQL] generate sort merge join
Posted by davies <gi...@git.apache.org>.
Github user davies commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-187448749
@nongli We already avoid the copy until found the match. In order to support left N rows match with right M rows, we have to copy some of them. Right now, we copy the keys from left, and the rows from right. It's possible to have less copies, I'd like to leave that out of this PR, because SMJ usually required Exchange and Sort, those are more expensive than coping the matched rows from right.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13373] [SQL] generate sort merge join
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-187768132
**[Test build #51771 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/51771/consoleFull)** for PR 11248 at commit [`0aa075f`](https://github.com/apache/spark/commit/0aa075fedee2d3e34eddee44a5ad196003d22896).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13375] [SQL] generate sort merge join
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-185955172
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/51495/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13375] [SQL] generate sort merge join
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-185999147
**[Test build #2545 has finished](https://amplab.cs.berkeley.edu/jenkins/job/NewSparkPullRequestBuilder/2545/consoleFull)** for PR 11248 at commit [`09a97d4`](https://github.com/apache/spark/commit/09a97d42e17699d11bc5f372e9e0aaef780f2c66).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13373] [SQL] generate sort merge join
Posted by davies <gi...@git.apache.org>.
Github user davies commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-187954148
Merging this into master, thanks!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13373] [SQL] generate sort merge join
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-186747420
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13373] [SQL] generate sort merge join
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-187840342
**[Test build #51771 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/51771/consoleFull)** for PR 11248 at commit [`0aa075f`](https://github.com/apache/spark/commit/0aa075fedee2d3e34eddee44a5ad196003d22896).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13375] [SQL] generate sort merge join
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-185654397
Merged build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13373] [SQL] generate sort merge join
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on a diff in the pull request:
https://github.com/apache/spark/pull/11248#discussion_r53710871
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoin.scala ---
@@ -125,6 +126,246 @@ case class SortMergeJoin(
}.toScala
}
}
+
+ override def upstreams(): Seq[RDD[InternalRow]] = {
+ left.execute() :: right.execute() :: Nil
+ }
+
+ private def createJoinKey(
+ ctx: CodegenContext,
+ row: String,
+ keys: Seq[Expression],
+ input: Seq[Attribute]): Seq[ExprCode] = {
+ ctx.INPUT_ROW = row
+ keys.map(BindReferences.bindReference(_, input).gen(ctx))
+ }
+
+ private def copyKeys(ctx: CodegenContext, vars: Seq[ExprCode]): Seq[ExprCode] = {
+ vars.zipWithIndex.map { case (ev, i) =>
+ val value = ctx.freshName("value")
+ ctx.addMutableState(ctx.javaType(leftKeys(i).dataType), value, "")
+ val code =
+ s"""
+ |$value = ${ev.value};
+ """.stripMargin
+ ExprCode(code, "false", value)
+ }
+ }
+
+ private def genComparision(ctx: CodegenContext, a: Seq[ExprCode], b: Seq[ExprCode]): String = {
+ val comparisons = a.zip(b).zipWithIndex.map { case ((l, r), i) =>
+ s"""
+ |if (comp == 0) {
+ | comp = ${ctx.genComp(leftKeys(i).dataType, l.value, r.value)};
+ |}
+ """.stripMargin.trim
+ }
+ s"""
+ |comp = 0;
+ |${comparisons.mkString("\n")}
+ """.stripMargin
+ }
+
+ /**
+ * Generate a function to scan both left and right to find a match, returns the term for
+ * matched one row from left side and buffered rows from right side.
+ */
+ private def genScanner(ctx: CodegenContext): (String, String) = {
+ // Create class member for next row from both sides.
+ val leftRow = ctx.freshName("leftRow")
+ ctx.addMutableState("InternalRow", leftRow, "")
+ val rightRow = ctx.freshName("rightRow")
+ ctx.addMutableState("InternalRow", rightRow, s"$rightRow = null;")
+
+ // Create variables for join keys from both sides.
+ val leftKeyVars = createJoinKey(ctx, leftRow, leftKeys, left.output)
+ val leftAnyNull = leftKeyVars.map(_.isNull).mkString(" || ")
+ val rightKeyTmpVars = createJoinKey(ctx, rightRow, rightKeys, right.output)
+ val rightAnyNull = rightKeyTmpVars.map(_.isNull).mkString(" || ")
+ // Copy the right key as class members so they could be used in next function call.
+ val rightKeyVars = copyKeys(ctx, rightKeyTmpVars)
+
+ // A list to hold all matched rows from right side.
+ val matches = ctx.freshName("matches")
+ val clsName = classOf[java.util.ArrayList[InternalRow]].getName
+ ctx.addMutableState(clsName, matches, s"$matches = new $clsName();")
+ // Copy the left keys as class members so they could be used in next function call.
+ val matchedKeyVars = copyKeys(ctx, leftKeyVars)
+
+ ctx.addNewFunction("findNextInnerJoinRows",
+ s"""
+ |private boolean findNextInnerJoinRows(
+ | scala.collection.Iterator leftIter,
+ | scala.collection.Iterator rightIter) {
+ | $leftRow = null;
+ | int comp = 0;
+ | while ($leftRow == null) {
+ | if (!leftIter.hasNext()) return false;
+ | $leftRow = (InternalRow) leftIter.next();
+ | ${leftKeyVars.map(_.code).mkString("\n")}
+ | if ($leftAnyNull) {
+ | $leftRow = null;
+ | continue;
+ | }
+ | if (!$matches.isEmpty()) {
+ | ${genComparision(ctx, leftKeyVars, matchedKeyVars)}
+ | if (comp == 0) {
+ | return true;
+ | }
+ | $matches.clear();
+ | }
+ |
+ | do {
+ | if ($rightRow == null) {
+ | if (!rightIter.hasNext()) {
+ | ${matchedKeyVars.map(_.code).mkString("\n")}
+ | return !$matches.isEmpty();
+ | }
+ | $rightRow = (InternalRow) rightIter.next();
+ | ${rightKeyTmpVars.map(_.code).mkString("\n")}
+ | if ($rightAnyNull) {
+ | $rightRow = null;
+ | continue;
+ | }
+ | ${rightKeyVars.map(_.code).mkString("\n")}
+ | }
+ | ${genComparision(ctx, leftKeyVars, rightKeyVars)}
+ | if (comp > 0) {
+ | $rightRow = null;
+ | } else if (comp < 0) {
+ | if (!$matches.isEmpty()) {
+ | ${matchedKeyVars.map(_.code).mkString("\n")}
+ | return true;
+ | }
+ | $leftRow = null;
+ | } else {
+ | $matches.add($rightRow.copy());
+ | $rightRow = null;;
+ | }
+ | } while ($leftRow != null);
+ | }
+ | return false; // unreachable
+ |}
+ """.stripMargin)
+
+ (leftRow, matches)
+ }
+
+ /**
+ * Creates variables for left part of result row.
+ *
+ * In order to defer the access after condition and also only access once in the loop,
+ * the variables should be declared separately from accessing the columns, we can't use the
+ * codegen of BoundReference here.
+ */
+ private def createLeftVars(ctx: CodegenContext, leftRow: String): Seq[ExprCode] = {
+ ctx.INPUT_ROW = leftRow
+ left.output.zipWithIndex.map { case (a, i) =>
+ val value = ctx.freshName("value")
+ val valueCode = ctx.getValue(leftRow, a.dataType, i.toString)
+ // declare it as class member, so we can access the column before or in the loop.
+ ctx.addMutableState(ctx.javaType(a.dataType), value, "")
+ if (a.nullable) {
+ val isNull = ctx.freshName("isNull")
+ ctx.addMutableState("boolean", isNull, "")
+ val code =
+ s"""
+ |$isNull = $leftRow.isNullAt($i);
+ |$value = $isNull ? ${ctx.defaultValue(a.dataType)} : ($valueCode);
+ """.stripMargin
+ ExprCode(code, isNull, value)
+ } else {
+ ExprCode(s"$value = $valueCode;", "false", value)
+ }
+ }
+ }
+
+ /**
+ * Creates the variables for right part of result row, using BoundReference, since the right
+ * part are accessed inside the loop.
+ */
+ private def createRightVar(ctx: CodegenContext, rightRow: String): Seq[ExprCode] = {
+ ctx.INPUT_ROW = rightRow
+ right.output.zipWithIndex.map { case (a, i) =>
+ BoundReference(i, a.dataType, a.nullable).gen(ctx)
+ }
+ }
+
+ /**
+ * Splits variables based on whether it's used by condition or not, returns the code to create
+ * these variables before the condition and after the condition.
+ *
+ * Only a few columns are used by condition, then we can skip the accessing of those columns
+ * that are not used by condition also filtered out by condition.
+ */
+ private def splitVarsByCondition(
+ attributes: Seq[Attribute],
+ variables: Seq[ExprCode]): (String, String) = {
+ if (condition.isDefined) {
+ val condRefs = condition.get.references
+ val (used, notUsed) = attributes.zip(variables).partition{ case (a, ev) =>
+ condRefs.contains(a)
+ }
+ val beforeCond = used.map(_._2.code).mkString("\n")
+ val afterCond = notUsed.map(_._2.code).mkString("\n")
+ (beforeCond, afterCond)
+ } else {
+ (variables.map(_.code).mkString("\n"), "")
+ }
+ }
+
+ override def doProduce(ctx: CodegenContext): String = {
+ val leftInput = ctx.freshName("leftInput")
+ ctx.addMutableState("scala.collection.Iterator", leftInput, s"$leftInput = inputs[0];")
+ val rightInput = ctx.freshName("rightInput")
+ ctx.addMutableState("scala.collection.Iterator", rightInput, s"$rightInput = inputs[1];")
+
+ val (leftRow, matches) = genScanner(ctx)
+
+ // Create variables for row from both sides.
+ val leftVars = createLeftVars(ctx, leftRow)
+ val rightRow = ctx.freshName("rightRow")
+ val rightVars = createRightVar(ctx, rightRow)
+ val resultVars = leftVars ++ rightVars
+
+ // Check condition
+ ctx.currentVars = resultVars
+ val cond = if (condition.isDefined) {
+ BindReferences.bindReference(condition.get, output).gen(ctx)
+ } else {
+ ExprCode("", "false", "true")
+ }
+ // Split the code of creating variables based on whether it's used by condition or not.
+ val loaded = ctx.freshName("loaded")
+ val (leftBefore, leftAfter) = splitVarsByCondition(left.output, leftVars)
+ val (rightBefore, rightAfter) = splitVarsByCondition(right.output, rightVars)
+
+
+ val size = ctx.freshName("size")
+ val i = ctx.freshName("i")
+ val numOutput = metricTerm(ctx, "numOutputRows")
+ s"""
+ |while (findNextInnerJoinRows($leftInput, $rightInput)) {
+ | int $size = $matches.size();
+ | boolean $loaded = false;
+ | $leftBefore
+ | for (int $i = 0; $i < $size; $i ++) {
+ | InternalRow $rightRow = (InternalRow) $matches.get($i);
+ | $rightBefore
+ | ${cond.code}
+ | if (${cond.isNull} || !${cond.value}) continue;
+ | if (!$loaded) {
+ | $loaded = true;
+ | $leftAfter
+ | }
+ | $rightAfter
+ | $numOutput.add(1);
--- End diff --
how expensive is this? row by row counters gets expensive quickly (for later)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13375] [SQL] generate sort merge join
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-185837578
Merged build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13375] [SQL] generate sort merge join
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-185587872
**[Test build #51471 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/51471/consoleFull)** for PR 11248 at commit [`2a83ff5`](https://github.com/apache/spark/commit/2a83ff5dafd3dbd57faa718bc11fd8f8b4ace751).
* This patch **fails Scala style tests**.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* `public abstract class BufferedRowIterator `
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13373] [SQL] generate sort merge join
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-187445997
Does it make sense to do something like this and remove the copies?
```
/* 056 */ private void agg_doAggregateWithoutKey() throws java.io.IOException {
/* 057 */ // initialize aggregation buffer
/* 058 */
/* 059 */ agg_bufIsNull = false;
/* 060 */ agg_bufValue = 0L;
/* 061 */
smj_left = leftIter.next();
smj_right = leftIter.next();
while (true) {
int comp = compare(smj_left, smj_right);
if (comp > 0) {
if (!leftIter.hasNext()) break;
smj_left = leftIter.next();
} else if (comp < 0) {
if (!rightIter.hasNext()) break;
rightIter = rightIter.next();
}
// Match
/* 071 */ long smj_value5 = smj_isNull3 ? -1L : (smj_rightRow1.getLong(0));
/* 072 */
/* 073 */ if (false || !true) continue;
/* 074 */ smj_metricValue.add(1);
/* 075 */
/* 076 */ // do aggregate
/* 077 */ /* (input[0, bigint] + 1) */
/* 078 */ long agg_value1 = -1L;
/* 079 */ agg_value1 = agg_bufValue + 1L;
/* 080 */ // update aggregation buffer
/* 081 */ agg_bufIsNull = false;
/* 082 */ agg_bufValue = agg_value1;
/* 083 */
// Advance 1
if (!leftIter.hasNext()) break;
smj_left = leftIter.next();
}
/* 088 */ }
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13373] [SQL] generate sort merge join
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on a diff in the pull request:
https://github.com/apache/spark/pull/11248#discussion_r53709739
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/WholeStageCodegen.scala ---
@@ -172,19 +175,23 @@ case class InputAdapter(child: SparkPlan) extends LeafNode with CodegenSupport {
override def supportCodegen: Boolean = false
- override def upstream(): RDD[InternalRow] = {
- child.execute()
+ override def upstreams(): Seq[RDD[InternalRow]] = {
+ child.execute() :: Nil
}
override def doProduce(ctx: CodegenContext): String = {
+ val input = ctx.freshName("input")
+ // Right now, Range is only used when there is one upstream.
--- End diff --
Range -> InputAdapter?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13373] [SQL] generate sort merge join
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-186747269
**[Test build #51607 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/51607/consoleFull)** for PR 11248 at commit [`55148d4`](https://github.com/apache/spark/commit/55148d442b5d181d03d43687972ebc5bd30f18b6).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13375] [SQL] generate sort merge join
Posted by davies <gi...@git.apache.org>.
Github user davies commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-185583137
Generated code for
```
val df1 = sqlContext.range(N).selectExpr(s"id as k1")
val df2 = sqlContext.range(N).selectExpr(s"id as k2")
df1.join(df2, col("k1") === col("k2")).count()
```
```
/* 001 */
/* 002 */ public Object generate(Object[] references) {
/* 003 */ return new GeneratedIterator(references);
/* 004 */ }
/* 005 */
/* 006 */ /** Codegened pipeline for:
/* 007 */ * TungstenAggregate(key=[], functions=[(count(1),mode=Partial,isDistinct=false)], output=[count#191L])
/* 008 */ +- Project
/* 009 */ +- SortMergeJoin [k1#185L], [k2#187L], None
/* 010 */ :- INPUT
/* 011 */ +- INPUT
/* 012 */ */
/* 013 */ class GeneratedIterator extends org.apache.spark.sql.execution.BufferedRowIterator {
/* 014 */ private Object[] references;
/* 015 */ private boolean agg_initAgg;
/* 016 */ private boolean agg_bufIsNull;
/* 017 */ private long agg_bufValue;
/* 018 */ private scala.collection.Iterator smj_leftInput;
/* 019 */ private scala.collection.Iterator smj_rightInput;
/* 020 */ private InternalRow smj_leftRow;
/* 021 */ private InternalRow smj_rightRow;
/* 022 */ private java.util.LinkedList smj_matches;
/* 023 */ private boolean smj_isNull;
/* 024 */ private long smj_value;
/* 025 */ private boolean smj_isNull2;
/* 026 */ private long smj_value2;
/* 027 */ private org.apache.spark.sql.execution.metric.LongSQLMetric smj_numOutputRows;
/* 028 */ private org.apache.spark.sql.execution.metric.LongSQLMetricValue smj_metricValue;
/* 029 */ private org.apache.spark.sql.execution.metric.LongSQLMetric agg_numOutputRows;
/* 030 */ private org.apache.spark.sql.execution.metric.LongSQLMetricValue agg_metricValue;
/* 031 */ private UnsafeRow agg_result;
/* 032 */ private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder agg_holder;
/* 033 */ private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter agg_rowWriter;
/* 034 */
/* 035 */ public GeneratedIterator(Object[] references) {
/* 036 */ this.references = references;
/* 037 */ }
/* 038 */
/* 039 */ public void init(scala.collection.Iterator inputs[]) {
/* 040 */ agg_initAgg = false;
/* 041 */
/* 042 */ smj_leftInput = inputs[0];
/* 043 */ smj_rightInput = inputs[1];
/* 044 */
/* 045 */ smj_rightRow = null;
/* 046 */ smj_matches = new java.util.LinkedList();
/* 047 */
/* 048 */ this.smj_numOutputRows = (org.apache.spark.sql.execution.metric.LongSQLMetric) references[0];
/* 049 */ smj_metricValue = (org.apache.spark.sql.execution.metric.LongSQLMetricValue) smj_numOutputRows.localValue();
/* 050 */ this.agg_numOutputRows = (org.apache.spark.sql.execution.metric.LongSQLMetric) references[1];
/* 051 */ agg_metricValue = (org.apache.spark.sql.execution.metric.LongSQLMetricValue) agg_numOutputRows.localValue();
/* 052 */ agg_result = new UnsafeRow(1);
/* 053 */ this.agg_holder = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(agg_result, 0);
/* 054 */ this.agg_rowWriter = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(agg_holder, 1);
/* 055 */ }
/* 056 */
/* 057 */ private void agg_doAggregateWithoutKey() throws java.io.IOException {
/* 058 */ // initialize aggregation buffer
/* 059 */
/* 060 */ agg_bufIsNull = false;
/* 061 */ agg_bufValue = 0L;
/* 062 */
/* 063 */ while (findNextInnerJoinRows(smj_leftInput, smj_rightInput)) {
/* 064 */ /* input[0, bigint] */
/* 065 */ boolean smj_isNull4 = smj_leftRow.isNullAt(0);
/* 066 */ long smj_value4 = smj_isNull4 ? -1L : (smj_leftRow.getLong(0));
/* 067 */ while (!smj_matches.isEmpty()) {
/* 068 */ InternalRow smj_rightRow1 = (InternalRow) smj_matches.remove();
/* 069 */ /* input[0, bigint] */
/* 070 */ boolean smj_isNull5 = smj_rightRow1.isNullAt(0);
/* 071 */ long smj_value5 = smj_isNull5 ? -1L : (smj_rightRow1.getLong(0));
/* 072 */ smj_metricValue.add(1);
/* 073 */
/* 074 */ // do aggregate
/* 075 */ /* (input[0, bigint] + 1) */
/* 076 */ long agg_value1 = -1L;
/* 077 */ agg_value1 = agg_bufValue + 1L;
/* 078 */ // update aggregation buffer
/* 079 */ agg_bufIsNull = false;
/* 080 */ agg_bufValue = agg_value1;
/* 081 */
/* 082 */ }
/* 083 */ }
/* 084 */
/* 085 */ }
/* 086 */
/* 087 */ private boolean findNextInnerJoinRows(
/* 088 */ scala.collection.Iterator leftIter,
/* 089 */ scala.collection.Iterator rightIter) {
/* 090 */ smj_leftRow = null;
/* 091 */ smj_matches.clear();
/* 092 */ while (smj_leftRow == null) {
/* 093 */ if (!leftIter.hasNext()) return false;
/* 094 */ smj_leftRow = (InternalRow) leftIter.next();
/* 095 */
/* 096 */ /* input[0, bigint] */
/* 097 */ boolean smj_isNull1 = smj_leftRow.isNullAt(0);
/* 098 */ long smj_value1 = smj_isNull1 ? -1L : (smj_leftRow.getLong(0));
/* 099 */ smj_isNull = smj_isNull1;
/* 100 */ smj_value = smj_value1;
/* 101 */
/* 102 */ if (smj_isNull) {
/* 103 */ smj_leftRow = null;
/* 104 */ continue;
/* 105 */ }
/* 106 */
/* 107 */ do {
/* 108 */ if (smj_rightRow == null) {
/* 109 */ if (!rightIter.hasNext()) {
/* 110 */ return !smj_matches.isEmpty();
/* 111 */ }
/* 112 */ smj_rightRow = (InternalRow) rightIter.next();
/* 113 */ }
/* 114 */
/* 115 */ /* input[0, bigint] */
/* 116 */ boolean smj_isNull3 = smj_rightRow.isNullAt(0);
/* 117 */ long smj_value3 = smj_isNull3 ? -1L : (smj_rightRow.getLong(0));
/* 118 */ smj_isNull2 = smj_isNull3;
/* 119 */ smj_value2 = smj_value3;
/* 120 */
/* 121 */ if (smj_isNull2) {
/* 122 */ smj_rightRow = null;
/* 123 */ continue;
/* 124 */ }
/* 125 */
/* 126 */ int smj_comp = 0;
/* 127 */ if (smj_comp == 0) {
/* 128 */ smj_comp = (smj_value > smj_value2 ? 1 : smj_value < smj_value2 ? -1 : 0);
/* 129 */ }
/* 130 */
/* 131 */ if (smj_comp > 0) {
/* 132 */ smj_rightRow = null;
/* 133 */ } else if (smj_comp < 0) {
/* 134 */ if (!smj_matches.isEmpty()) return true;
/* 135 */ smj_leftRow = null;
/* 136 */ } else {
/* 137 */ smj_matches.add(smj_rightRow.copy());
/* 138 */ smj_rightRow = null;;
/* 139 */ }
/* 140 */ } while (smj_leftRow != null);
/* 141 */ }
/* 142 */ return false; // unreachable
/* 143 */ }
/* 144 */
/* 145 */ protected void processNext() throws java.io.IOException {
/* 146 */ if (!agg_initAgg) {
/* 147 */ agg_initAgg = true;
/* 148 */ agg_doAggregateWithoutKey();
/* 149 */
/* 150 */ // output the result
/* 151 */
/* 152 */ agg_metricValue.add(1);
/* 153 */ agg_rowWriter.zeroOutNullBytes();
/* 154 */
/* 155 */ if (agg_bufIsNull) {
/* 156 */ agg_rowWriter.setNullAt(0);
/* 157 */ } else {
/* 158 */ agg_rowWriter.write(0, agg_bufValue);
/* 159 */ }
/* 160 */ append(agg_result.copy());
/* 161 */ }
/* 162 */ }
/* 163 */ }
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13375] [SQL] generate sort merge join
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-185889625
**[Test build #51495 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/51495/consoleFull)** for PR 11248 at commit [`09a97d4`](https://github.com/apache/spark/commit/09a97d42e17699d11bc5f372e9e0aaef780f2c66).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13375] [SQL] generate sort merge join
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-185587880
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/51471/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13373] [SQL] generate sort merge join
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on a diff in the pull request:
https://github.com/apache/spark/pull/11248#discussion_r53710796
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoin.scala ---
@@ -125,6 +126,246 @@ case class SortMergeJoin(
}.toScala
}
}
+
+ override def upstreams(): Seq[RDD[InternalRow]] = {
+ left.execute() :: right.execute() :: Nil
+ }
+
+ private def createJoinKey(
+ ctx: CodegenContext,
+ row: String,
+ keys: Seq[Expression],
+ input: Seq[Attribute]): Seq[ExprCode] = {
+ ctx.INPUT_ROW = row
+ keys.map(BindReferences.bindReference(_, input).gen(ctx))
+ }
+
+ private def copyKeys(ctx: CodegenContext, vars: Seq[ExprCode]): Seq[ExprCode] = {
+ vars.zipWithIndex.map { case (ev, i) =>
+ val value = ctx.freshName("value")
+ ctx.addMutableState(ctx.javaType(leftKeys(i).dataType), value, "")
+ val code =
+ s"""
+ |$value = ${ev.value};
+ """.stripMargin
+ ExprCode(code, "false", value)
+ }
+ }
+
+ private def genComparision(ctx: CodegenContext, a: Seq[ExprCode], b: Seq[ExprCode]): String = {
+ val comparisons = a.zip(b).zipWithIndex.map { case ((l, r), i) =>
+ s"""
+ |if (comp == 0) {
+ | comp = ${ctx.genComp(leftKeys(i).dataType, l.value, r.value)};
+ |}
+ """.stripMargin.trim
+ }
+ s"""
+ |comp = 0;
+ |${comparisons.mkString("\n")}
+ """.stripMargin
+ }
+
+ /**
+ * Generate a function to scan both left and right to find a match, returns the term for
+ * matched one row from left side and buffered rows from right side.
+ */
+ private def genScanner(ctx: CodegenContext): (String, String) = {
+ // Create class member for next row from both sides.
+ val leftRow = ctx.freshName("leftRow")
+ ctx.addMutableState("InternalRow", leftRow, "")
+ val rightRow = ctx.freshName("rightRow")
+ ctx.addMutableState("InternalRow", rightRow, s"$rightRow = null;")
+
+ // Create variables for join keys from both sides.
+ val leftKeyVars = createJoinKey(ctx, leftRow, leftKeys, left.output)
+ val leftAnyNull = leftKeyVars.map(_.isNull).mkString(" || ")
+ val rightKeyTmpVars = createJoinKey(ctx, rightRow, rightKeys, right.output)
+ val rightAnyNull = rightKeyTmpVars.map(_.isNull).mkString(" || ")
+ // Copy the right key as class members so they could be used in next function call.
+ val rightKeyVars = copyKeys(ctx, rightKeyTmpVars)
+
+ // A list to hold all matched rows from right side.
+ val matches = ctx.freshName("matches")
+ val clsName = classOf[java.util.ArrayList[InternalRow]].getName
+ ctx.addMutableState(clsName, matches, s"$matches = new $clsName();")
+ // Copy the left keys as class members so they could be used in next function call.
+ val matchedKeyVars = copyKeys(ctx, leftKeyVars)
+
+ ctx.addNewFunction("findNextInnerJoinRows",
+ s"""
+ |private boolean findNextInnerJoinRows(
+ | scala.collection.Iterator leftIter,
+ | scala.collection.Iterator rightIter) {
+ | $leftRow = null;
+ | int comp = 0;
+ | while ($leftRow == null) {
+ | if (!leftIter.hasNext()) return false;
+ | $leftRow = (InternalRow) leftIter.next();
+ | ${leftKeyVars.map(_.code).mkString("\n")}
+ | if ($leftAnyNull) {
+ | $leftRow = null;
+ | continue;
+ | }
+ | if (!$matches.isEmpty()) {
+ | ${genComparision(ctx, leftKeyVars, matchedKeyVars)}
+ | if (comp == 0) {
+ | return true;
+ | }
+ | $matches.clear();
+ | }
+ |
+ | do {
+ | if ($rightRow == null) {
+ | if (!rightIter.hasNext()) {
+ | ${matchedKeyVars.map(_.code).mkString("\n")}
+ | return !$matches.isEmpty();
+ | }
+ | $rightRow = (InternalRow) rightIter.next();
+ | ${rightKeyTmpVars.map(_.code).mkString("\n")}
+ | if ($rightAnyNull) {
+ | $rightRow = null;
+ | continue;
+ | }
+ | ${rightKeyVars.map(_.code).mkString("\n")}
+ | }
+ | ${genComparision(ctx, leftKeyVars, rightKeyVars)}
+ | if (comp > 0) {
+ | $rightRow = null;
+ | } else if (comp < 0) {
+ | if (!$matches.isEmpty()) {
+ | ${matchedKeyVars.map(_.code).mkString("\n")}
+ | return true;
+ | }
+ | $leftRow = null;
+ | } else {
+ | $matches.add($rightRow.copy());
+ | $rightRow = null;;
+ | }
+ | } while ($leftRow != null);
+ | }
+ | return false; // unreachable
+ |}
+ """.stripMargin)
+
+ (leftRow, matches)
+ }
+
+ /**
+ * Creates variables for left part of result row.
+ *
+ * In order to defer the access after condition and also only access once in the loop,
+ * the variables should be declared separately from accessing the columns, we can't use the
+ * codegen of BoundReference here.
+ */
+ private def createLeftVars(ctx: CodegenContext, leftRow: String): Seq[ExprCode] = {
+ ctx.INPUT_ROW = leftRow
+ left.output.zipWithIndex.map { case (a, i) =>
+ val value = ctx.freshName("value")
+ val valueCode = ctx.getValue(leftRow, a.dataType, i.toString)
+ // declare it as class member, so we can access the column before or in the loop.
+ ctx.addMutableState(ctx.javaType(a.dataType), value, "")
+ if (a.nullable) {
+ val isNull = ctx.freshName("isNull")
+ ctx.addMutableState("boolean", isNull, "")
+ val code =
+ s"""
+ |$isNull = $leftRow.isNullAt($i);
+ |$value = $isNull ? ${ctx.defaultValue(a.dataType)} : ($valueCode);
+ """.stripMargin
+ ExprCode(code, isNull, value)
+ } else {
+ ExprCode(s"$value = $valueCode;", "false", value)
+ }
+ }
+ }
+
+ /**
+ * Creates the variables for right part of result row, using BoundReference, since the right
+ * part are accessed inside the loop.
+ */
+ private def createRightVar(ctx: CodegenContext, rightRow: String): Seq[ExprCode] = {
+ ctx.INPUT_ROW = rightRow
+ right.output.zipWithIndex.map { case (a, i) =>
+ BoundReference(i, a.dataType, a.nullable).gen(ctx)
+ }
+ }
+
+ /**
+ * Splits variables based on whether it's used by condition or not, returns the code to create
+ * these variables before the condition and after the condition.
+ *
+ * Only a few columns are used by condition, then we can skip the accessing of those columns
+ * that are not used by condition also filtered out by condition.
+ */
+ private def splitVarsByCondition(
+ attributes: Seq[Attribute],
+ variables: Seq[ExprCode]): (String, String) = {
+ if (condition.isDefined) {
+ val condRefs = condition.get.references
+ val (used, notUsed) = attributes.zip(variables).partition{ case (a, ev) =>
+ condRefs.contains(a)
+ }
+ val beforeCond = used.map(_._2.code).mkString("\n")
+ val afterCond = notUsed.map(_._2.code).mkString("\n")
+ (beforeCond, afterCond)
+ } else {
+ (variables.map(_.code).mkString("\n"), "")
+ }
+ }
+
+ override def doProduce(ctx: CodegenContext): String = {
+ val leftInput = ctx.freshName("leftInput")
+ ctx.addMutableState("scala.collection.Iterator", leftInput, s"$leftInput = inputs[0];")
+ val rightInput = ctx.freshName("rightInput")
+ ctx.addMutableState("scala.collection.Iterator", rightInput, s"$rightInput = inputs[1];")
+
+ val (leftRow, matches) = genScanner(ctx)
+
+ // Create variables for row from both sides.
+ val leftVars = createLeftVars(ctx, leftRow)
+ val rightRow = ctx.freshName("rightRow")
+ val rightVars = createRightVar(ctx, rightRow)
+ val resultVars = leftVars ++ rightVars
+
+ // Check condition
+ ctx.currentVars = resultVars
+ val cond = if (condition.isDefined) {
+ BindReferences.bindReference(condition.get, output).gen(ctx)
+ } else {
+ ExprCode("", "false", "true")
+ }
+ // Split the code of creating variables based on whether it's used by condition or not.
+ val loaded = ctx.freshName("loaded")
+ val (leftBefore, leftAfter) = splitVarsByCondition(left.output, leftVars)
+ val (rightBefore, rightAfter) = splitVarsByCondition(right.output, rightVars)
+
+
+ val size = ctx.freshName("size")
+ val i = ctx.freshName("i")
+ val numOutput = metricTerm(ctx, "numOutputRows")
+ s"""
+ |while (findNextInnerJoinRows($leftInput, $rightInput)) {
+ | int $size = $matches.size();
+ | boolean $loaded = false;
+ | $leftBefore
+ | for (int $i = 0; $i < $size; $i ++) {
+ | InternalRow $rightRow = (InternalRow) $matches.get($i);
+ | $rightBefore
+ | ${cond.code}
+ | if (${cond.isNull} || !${cond.value}) continue;
+ | if (!$loaded) {
+ | $loaded = true;
+ | $leftAfter
+ | }
+ | $rightAfter
+ | $numOutput.add(1);
+ | ${consume(ctx, resultVars)}
+ | }
+ | if (shouldStop()) return;
--- End diff --
For another patch. I think doing the shouldStop() check here is weird. I think the rule should be the last operator in the pipeline is responsible for dealing with this.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13375] [SQL] generate sort merge join
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-185654399
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/51477/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13373] [SQL] generate sort merge join
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/11248#discussion_r53865666
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/execution/BenchmarkWholeStageCodegen.scala ---
@@ -38,6 +38,7 @@ import org.apache.spark.util.Benchmark
class BenchmarkWholeStageCodegen extends SparkFunSuite {
lazy val conf = new SparkConf().setMaster("local[1]").setAppName("benchmark")
.set("spark.sql.shuffle.partitions", "1")
+ .set("spark.sql.autoBroadcastJoinThreshold", "0")
--- End diff --
We will use BroadcastHint when test with Broadcast join, it's better to disable it for all.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13375] [SQL] generate sort merge join
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-185665301
**[Test build #51478 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/51478/consoleFull)** for PR 11248 at commit [`b3669f2`](https://github.com/apache/spark/commit/b3669f293fd904b716e2dc919375319519c0894f).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* `public abstract class BufferedRowIterator `
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org