[SPARK-22746][SQL] Avoid the generation of useless mutable states by SortMergeJoin #19937

kiszk · 2017-12-09T15:32:29Z

What changes were proposed in this pull request?

This PR reduce the number of global mutable variables in generated code of SortMergeJoin.

Before this PR, global mutable variables are used to extend lifetime of variables in the nested loop. This can be achieved by declaring variable at the outer most loop level where the variables are used.
In the following example, smj_value8, smj_value8, and smj_value9 are declared as local variable at lines 145-147 in With this PR.

This PR fixes potential assertion error by #19865. Without this PR, a global mutable variable is potentially passed to arguments in generated code of split function.

Without this PR

/* 010 */   int smj_value8;
/* 011 */   boolean smj_value8;
/* 012 */   int smj_value9;
..
/* 143 */   protected void processNext() throws java.io.IOException {
/* 144 */     while (findNextInnerJoinRows(smj_leftInput, smj_rightInput)) {
/* 145 */       boolean smj_loaded = false;
/* 146 */       smj_isNull6 = smj_leftRow.isNullAt(1);
/* 147 */       smj_value9 = smj_isNull6 ? -1 : (smj_leftRow.getInt(1));
/* 148 */       scala.collection.Iterator<UnsafeRow> smj_iterator = smj_matches.generateIterator();
/* 149 */       while (smj_iterator.hasNext()) {
/* 150 */         InternalRow smj_rightRow1 = (InternalRow) smj_iterator.next();
/* 151 */         boolean smj_isNull8 = smj_rightRow1.isNullAt(1);
/* 152 */         int smj_value11 = smj_isNull8 ? -1 : (smj_rightRow1.getInt(1));
/* 153 */
/* 154 */         boolean smj_value12 = (smj_isNull6 && smj_isNull8) ||
/* 155 */         (!smj_isNull6 && !smj_isNull8 && smj_value9 == smj_value11);
/* 156 */         if (false || !smj_value12) continue;
/* 157 */         if (!smj_loaded) {
/* 158 */           smj_loaded = true;
/* 159 */           smj_value8 = smj_leftRow.getInt(0);
/* 160 */         }
/* 161 */         int smj_value10 = smj_rightRow1.getInt(0);
/* 162 */         smj_numOutputRows.add(1);
/* 163 */
/* 164 */         smj_rowWriter.zeroOutNullBytes();
/* 165 */
/* 166 */         smj_rowWriter.write(0, smj_value8);
/* 167 */
/* 168 */         if (smj_isNull6) {
/* 169 */           smj_rowWriter.setNullAt(1);
/* 170 */         } else {
/* 171 */           smj_rowWriter.write(1, smj_value9);
/* 172 */         }
/* 173 */
/* 174 */         smj_rowWriter.write(2, smj_value10);
/* 175 */
/* 176 */         if (smj_isNull8) {
/* 177 */           smj_rowWriter.setNullAt(3);
/* 178 */         } else {
/* 179 */           smj_rowWriter.write(3, smj_value11);
/* 180 */         }
/* 181 */         append(smj_result.copy());
/* 182 */
/* 183 */       }
/* 184 */       if (shouldStop()) return;
/* 185 */     }
/* 186 */   }

With this PR

/* 143 */   protected void processNext() throws java.io.IOException {
/* 144 */     while (findNextInnerJoinRows(smj_leftInput, smj_rightInput)) {
/* 145 */       int smj_value8 = -1;
/* 146 */       boolean smj_isNull6 = false;
/* 147 */       int smj_value9 = -1;
/* 148 */       boolean smj_loaded = false;
/* 149 */       smj_isNull6 = smj_leftRow.isNullAt(1);
/* 150 */       smj_value9 = smj_isNull6 ? -1 : (smj_leftRow.getInt(1));
/* 151 */       scala.collection.Iterator<UnsafeRow> smj_iterator = smj_matches.generateIterator();
/* 152 */       while (smj_iterator.hasNext()) {
/* 153 */         InternalRow smj_rightRow1 = (InternalRow) smj_iterator.next();
/* 154 */         boolean smj_isNull8 = smj_rightRow1.isNullAt(1);
/* 155 */         int smj_value11 = smj_isNull8 ? -1 : (smj_rightRow1.getInt(1));
/* 156 */
/* 157 */         boolean smj_value12 = (smj_isNull6 && smj_isNull8) ||
/* 158 */         (!smj_isNull6 && !smj_isNull8 && smj_value9 == smj_value11);
/* 159 */         if (false || !smj_value12) continue;
/* 160 */         if (!smj_loaded) {
/* 161 */           smj_loaded = true;
/* 162 */           smj_value8 = smj_leftRow.getInt(0);
/* 163 */         }
/* 164 */         int smj_value10 = smj_rightRow1.getInt(0);
/* 165 */         smj_numOutputRows.add(1);
/* 166 */
/* 167 */         smj_rowWriter.zeroOutNullBytes();
/* 168 */
/* 169 */         smj_rowWriter.write(0, smj_value8);
/* 170 */
/* 171 */         if (smj_isNull6) {
/* 172 */           smj_rowWriter.setNullAt(1);
/* 173 */         } else {
/* 174 */           smj_rowWriter.write(1, smj_value9);
/* 175 */         }
/* 176 */
/* 177 */         smj_rowWriter.write(2, smj_value10);
/* 178 */
/* 179 */         if (smj_isNull8) {
/* 180 */           smj_rowWriter.setNullAt(3);
/* 181 */         } else {
/* 182 */           smj_rowWriter.write(3, smj_value11);
/* 183 */         }
/* 184 */         append(smj_result.copy());
/* 185 */
/* 186 */       }
/* 187 */       if (shouldStop()) return;
/* 188 */     }
/* 189 */   }

How was this patch tested?

Existing test cases

mgaido91 · 2017-12-09T16:06:16Z

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoinExec.scala


    s"""
       |while (findNextInnerJoinRows($leftInput, $rightInput)) {
+       |  ${leftVarDecl.mkString("\n")}


can't we define them before the loop and reuse them?

It could be. Would it be possible to let us know advantages compare to the current method?
I think that to shorten lifetime of variables (i.e. current approach) makes generated code more readable.

Since I have no strong preference, I would like to hear opinions from others.

I think that the advantage would be to reuse them, avoiding creating and destroying them at every loop.

Since they are local variable, it takes almost no cost in native code. If they are on CPU registers, there is no cost. If they are in stack frame, it is up to one instruction to increase or decrease stack frame size.

WDYT? Did you see huge overhead to create and destroy local variables?

We would appreciate it if you would past the kernel code what you are thinking about.
This is because we would like to measure overhead which you are taking care of.

Here is my benchmark result. Is there any your result?

$ cat Loop.java public class Loop { private static int _i; private static boolean _b; public static void main(String[] args){ Loop l = new Loop(); long s, e; long c1 = 10000000000L; long c2 = 100000L; int N = 10; // warmup for (int i = 0; i < 1000000; i++) { l.inner(10000, 100); l.outer(10000, 100); } for (int n = 0; n < N; n++) { s = System.nanoTime(); l.inner(c1, c2); e = System.nanoTime(); System.out.println("inner(ms): " + (e-s) / 1000000); } for (int n = 0; n < N; n++) { s = System.nanoTime(); l.outer(c1, c2); e = System.nanoTime(); System.out.println("outer(ms): " + (e-s) / 1000000); } } public void inner(long c1, long c2) { for (long i = 0; i < c1; i++) { int v1 = 1; boolean v2 = true; for (long j = 0; i < c2; i++) { _i = v1; _b = v2; } } } public void outer(long c1, long c2) { int v1 = 1; boolean v2 = true; for (long i = 0; i < c1; i++) { for (long j = 0; i < c2; i++) { _i = v1; _b = v2; } } } } $ java -version openjdk version "1.8.0_131" OpenJDK Runtime Environment (build 1.8.0_131-8u131-b11-2ubuntu1.16.04.3-b11) OpenJDK 64-Bit Server VM (build 25.131-b11, mixed mode) $ java Loop inner(ms): 2779 inner(ms): 2779 inner(ms): 2779 inner(ms): 2779 inner(ms): 2779 inner(ms): 2779 inner(ms): 2779 inner(ms): 2779 inner(ms): 2779 inner(ms): 2779 outer(ms): 2779 outer(ms): 2779 outer(ms): 2779 outer(ms): 2779 outer(ms): 2779 outer(ms): 2779 outer(ms): 2779 outer(ms): 2779 outer(ms): 2779 outer(ms): 2779

If no performance difference, we prefer to the simpler codes.

Since these local variables are only needed inside the loop, I feel the current code is more readable. Performance is not a concern here as the overhead is very low or none.

@kiszk thanks for the test. Then I agree this is the best option, thanks.

SparkQA · 2017-12-09T18:22:37Z

Test build #84685 has finished for PR 19937 at commit 6d48924.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

kiszk · 2017-12-11T01:18:51Z

@cloud-fan @viirya Could you please review this?

gatorsmile · 2017-12-11T04:57:27Z

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoinExec.scala

   * codegen of BoundReference here.
   */
-  private def createLeftVars(ctx: CodegenContext, leftRow: String): Seq[ExprCode] = {
+  private def createLeftVars(ctx: CodegenContext, leftRow: String): (Seq[ExprCode], Seq[String]) = {


We need to update the function description.

gatorsmile · 2017-12-11T05:02:24Z

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoinExec.scala

+          s"""
+             |boolean $isNull = false;
+             |${ctx.javaType(a.dataType)} $value = ${ctx.defaultValue(a.dataType)};
+           """.stripMargin)


Could you assign it first

val leftVarsDecl = ... (ExprCode(code, isNull, value), leftVarsDecl)

gatorsmile · 2017-12-11T05:02:34Z

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoinExec.scala

      } else {
-        ExprCode(s"$value = $valueCode;", "false", value)
+        (ExprCode(s"$value = $valueCode;", "false", value),
+          s"""${ctx.javaType(a.dataType)} $value = ${ctx.defaultValue(a.dataType)};""")


The same here.

cloud-fan · 2017-12-11T07:49:27Z

LGTM except one code style comment

viirya · 2017-12-11T08:14:31Z

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoinExec.scala

             |$value = $isNull ? ${ctx.defaultValue(a.dataType)} : ($valueCode);
           """.stripMargin
-        ExprCode(code, isNull, value)
+        (ExprCode(code, isNull, value),


Why separate variable declaration and code to two parts? Previously it is because it uses global variables. Now since we use local variables, I think we can simply do:

val code = s""" |boolean $isNull = $leftRow.isNullAt($i); |${ctx.javaType(a.dataType)} $value = $isNull ? ${ctx.defaultValue(a.dataType)} : ($valueCode); """.stripMargin

Don't we?

good catch! Makes sense to me.

Good question. If we declare variable types only at code, it may lead to compilation error injanino.

Around here, the code that you pointed out may go to $leftAfter in $condCheck that exists in a if-then block at the inner most of the loop in the generated code, by using leftVar. The variable in leftVars is also referred to at ${consume(ctx, leftVars ++ rightVars)}.

In the following example, If we declare variable types only at code, we will drop lines 145 and will declare int for smj_value8 at lines 162. Since smj_value8 is refered at line 169, the generated code would cause compilation error.

WDYT?

/* 143 */ protected void processNext() throws java.io.IOException { /* 144 */ while (findNextInnerJoinRows(smj_leftInput, smj_rightInput)) { /* 145 */ int smj_value8 = -1; /* 146 */ boolean smj_isNull6 = false; /* 147 */ int smj_value9 = -1; /* 148 */ boolean smj_loaded = false; /* 149 */ smj_isNull6 = smj_leftRow.isNullAt(1); /* 150 */ smj_value9 = smj_isNull6 ? -1 : (smj_leftRow.getInt(1)); /* 151 */ scala.collection.Iterator<UnsafeRow> smj_iterator = smj_matches.generateIterator(); /* 152 */ while (smj_iterator.hasNext()) { ... /* 160 */ if (!smj_loaded) { /* 161 */ smj_loaded = true; /* 162 */ smj_value8 = smj_leftRow.getInt(0); /* 163 */ } /* 164 */ int smj_value10 = smj_rightRow1.getInt(0); /* 165 */ smj_numOutputRows.add(1); /* 166 */ /* 167 */ smj_rowWriter.zeroOutNullBytes(); /* 168 */ /* 169 */ smj_rowWriter.write(0, smj_value8); ... /* 185 */ /* 186 */ } /* 187 */ if (shouldStop()) return; /* 188 */ } /* 189 */ }

Yea, I see, I just ran the test and found it too. Looks like puts the declaration at top is simplest.

viirya · 2017-12-11T08:47:12Z

LGTM

mgaido91 · 2017-12-11T08:50:45Z

LGTM too

SparkQA · 2017-12-11T11:29:43Z

Test build #84703 has finished for PR 19937 at commit 9faf0a2.

This patch fails SparkR unit tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2017-12-11T11:42:53Z

retest this please

SparkQA · 2017-12-11T14:41:42Z

Test build #84705 has finished for PR 19937 at commit 9faf0a2.

This patch fails SparkR unit tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2017-12-11T16:21:18Z

retest this please

SparkQA · 2017-12-11T18:56:48Z

Test build #84718 has finished for PR 19937 at commit 9faf0a2.

This patch fails SparkR unit tests.
This patch merges cleanly.
This patch adds no public classes.

kiszk · 2017-12-11T18:59:13Z

Jenkins, retest this please

gatorsmile · 2017-12-11T21:14:12Z

The test failure is not related to this PR. I have reverted the PR that caused the CRAN checking failure.

gatorsmile · 2017-12-11T21:15:34Z

LGTM

Thanks! Merged to master

SparkQA · 2017-12-11T21:43:10Z

Test build #84723 has finished for PR 19937 at commit 9faf0a2.

This patch fails SparkR unit tests.
This patch merges cleanly.
This patch adds no public classes.

kiszk · 2017-12-12T02:54:51Z

Thank you for pointing out the root cause of this failure.

initial commit

6d48924

kiszk mentioned this pull request Dec 9, 2017

[SPARK-22668][SQL] Ensure no global variables in arguments of method split by CodegenContext.splitExpressions() #19865

Closed

mgaido91 reviewed Dec 9, 2017

View reviewed changes

gatorsmile reviewed Dec 11, 2017

View reviewed changes

viirya reviewed Dec 11, 2017

View reviewed changes

address review comments

9faf0a2

asfgit closed this in c235b5f Dec 11, 2017

kiszk mentioned this pull request Dec 19, 2017

[SPARK-22668][SQL] Ensure no global variables in arguments of method split by CodegenContext.splitExpressions() #20021

Closed

[SPARK-22746][SQL] Avoid the generation of useless mutable states by SortMergeJoin #19937

[SPARK-22746][SQL] Avoid the generation of useless mutable states by SortMergeJoin #19937

Uh oh!

Conversation

kiszk commented Dec 9, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kiszk Dec 9, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kiszk Dec 9, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kiszk Dec 10, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Dec 9, 2017

Uh oh!

kiszk commented Dec 11, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cloud-fan commented Dec 11, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kiszk Dec 11, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

viirya commented Dec 11, 2017

Uh oh!

mgaido91 commented Dec 11, 2017

Uh oh!

SparkQA commented Dec 11, 2017

Uh oh!

cloud-fan commented Dec 11, 2017

Uh oh!

SparkQA commented Dec 11, 2017

Uh oh!

cloud-fan commented Dec 11, 2017

Uh oh!

SparkQA commented Dec 11, 2017

Uh oh!

kiszk commented Dec 11, 2017

Uh oh!

gatorsmile commented Dec 11, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

gatorsmile commented Dec 11, 2017

Uh oh!

kiszk commented Dec 9, 2017 •

edited

Loading

kiszk Dec 9, 2017 •

edited

Loading

kiszk Dec 9, 2017 •

edited

Loading

kiszk Dec 10, 2017 •

edited

Loading

kiszk Dec 11, 2017 •

edited

Loading

gatorsmile commented Dec 11, 2017 •

edited

Loading