Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 2 commits into from
Closed

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

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6d48924
initial commit
kiszk Dec 9, 2017
9faf0a2
address review comments
kiszk Dec 11, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 17 additions & 10 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoinExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -507,32 +507,38 @@ case class SortMergeJoinExec(
}

/**
* Creates variables for left part of result row.
* Creates variables and declarations 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] = {
private def createLeftVars(ctx: CodegenContext, leftRow: String): (Seq[ExprCode], Seq[String]) = {
Copy link
Member

@gatorsmile gatorsmile Dec 11, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We need to update the function description.

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)
val javaType = ctx.javaType(a.dataType)
val defaultValue = ctx.defaultValue(a.dataType)
if (a.nullable) {
val isNull = ctx.freshName("isNull")
ctx.addMutableState(ctx.JAVA_BOOLEAN, isNull)
val code =
s"""
|$isNull = $leftRow.isNullAt($i);
|$value = $isNull ? ${ctx.defaultValue(a.dataType)} : ($valueCode);
|$value = $isNull ? $defaultValue : ($valueCode);
""".stripMargin
ExprCode(code, isNull, value)
val leftVarsDecl =
s"""
|boolean $isNull = false;
|$javaType $value = $defaultValue;
""".stripMargin
(ExprCode(code, isNull, value), leftVarsDecl)
} else {
ExprCode(s"$value = $valueCode;", "false", value)
val code = s"$value = $valueCode;"
val leftVarsDecl = s"""$javaType $value = $defaultValue;"""
(ExprCode(code, "false", value), leftVarsDecl)
}
}
}.unzip
}

/**
Expand Down Expand Up @@ -580,7 +586,7 @@ case class SortMergeJoinExec(
val (leftRow, matches) = genScanner(ctx)

// Create variables for row from both sides.
val leftVars = createLeftVars(ctx, leftRow)
val (leftVars, leftVarDecl) = createLeftVars(ctx, leftRow)
val rightRow = ctx.freshName("rightRow")
val rightVars = createRightVar(ctx, rightRow)

Expand Down Expand Up @@ -617,6 +623,7 @@ case class SortMergeJoinExec(

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

@mgaido91 mgaido91 Dec 9, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Copy link
Member Author

@kiszk kiszk Dec 9, 2017 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Member Author

@kiszk kiszk Dec 9, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Copy link
Contributor

@mgaido91 mgaido91 Dec 9, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Copy link
Member Author

@kiszk kiszk Dec 9, 2017 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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?

Copy link
Member Author

@kiszk kiszk Dec 10, 2017 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Member Author

@kiszk kiszk Dec 10, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Copy link
Member

@gatorsmile gatorsmile Dec 11, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Copy link
Contributor

@cloud-fan cloud-fan Dec 11, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Contributor

@mgaido91 mgaido91 Dec 11, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

| ${beforeLoop.trim}
| scala.collection.Iterator<UnsafeRow> $iterator = $matches.generateIterator();
| while ($iterator.hasNext()) {
Expand Down