[SPARK-22103] Move HashAggregateExec parent consume to a separate function in codegen #19324

juliuszsompolski · 2017-09-22T15:40:06Z

What changes were proposed in this pull request?

HashAggregateExec codegen uses two paths for fast hash table and a generic one.
It generates code paths for iterating over both, and both code paths generate the consume code of the parent operator, resulting in that code being expanded twice.
This leads to a long generated function that might be an issue for the compiler (see e.g. SPARK-21603).
I propose to remove the double expansion by generating the consume code in a helper function that can just be called from both iterating loops.

An issue with separating the consume code to a helper function was that a number of places relied and assumed on being in the scope of an outside produce loop and e.g. use continue to jump out.
I replaced such code flows with nested scopes. It is code that should be handled the same by compiler, while getting rid of depending on assumptions that are outside of the consume's own scope.

How was this patch tested?

Existing test coverage.

juliuszsompolski · 2017-09-22T15:42:00Z

@hvanhovell @gatorsmile @cloud-fan @rednaxelafx

SparkQA · 2017-09-22T15:42:25Z

Test build #82088 has started for PR 19324 at commit ca64368.

juliuszsompolski · 2017-09-22T15:45:27Z

sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala

    val doAggFuncName = ctx.addNewFunction(doAgg,
      s"""
-        ${generateGenerateCode}


this is a tangent fix: this generated code for the hash map was piggy-backed here together with the doAggregateWithKeys function, and it could become inaccessible from the top function if the function gets generated in a nested class (after #18075)

juliuszsompolski · 2017-09-22T15:46:26Z

sql/core/src/main/scala/org/apache/spark/sql/execution/WholeStageCodegenExec.scala

@@ -329,6 +332,15 @@ case class WholeStageCodegenExec(child: SparkPlan) extends UnaryExecNode with Co
  def doCodeGen(): (CodegenContext, CodeAndComment) = {
    val ctx = new CodegenContext
    val code = child.asInstanceOf[CodegenSupport].produce(ctx, this)
+
+    // main next function.
+    ctx.addNewFunction("processNext",


tangent fix: add processNext() with addNewFunction, so that it is also taken into account by #18810

juliuszsompolski · 2017-09-22T16:02:29Z

sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala

    } else if (modes.contains(Partial) || modes.contains(PartialMerge)) {
-      // This should be the last operator in a stage, we should output UnsafeRow directly


tangent fix: The partial aggregation doesn't necessarily have to be the last operator in the stage. E.g. if the shuffle requirement between the partial/final aggregation was already satisfied, or between 2. and 3. in planAggregateWithOneDistinct. Outputting the UnsafeRow through UnsafeRowJoiner was unnecessary then.

juliuszsompolski · 2017-09-22T16:19:53Z

@viirya This is related to #18931, as it also separates out the consume function. Maybe it would be enough to do similar splits into functions in the codegen of some operators that are materialization points (sort, joins) to keep the function length in check?
Splitting out on every consume takes away some of compiler's opportunities to optimize, like e.g. delaying evaluation of some projection (which you mentioned in your PR).
Removing the use of continue also simplifies not needing to handle it in your PR.

viirya · 2017-09-24T10:45:21Z

@juliuszsompolski Thanks for pinging me.

#18931 is an attempt to separate the consume function as it can as possible. With long chain of any operators, you can have a long consume function and fail JIT, this is the one reason it tries to split into functions at the root of codegen support, instead of in few operators individually. I'd avoid to duplicate the separate logic in all operators, IMO.

For the explicit delaying evaluation of projection, currently the strategy I take is not going to split it. I guess that you mean the evaluation that can be delayed by the compiler, I personally think it should not be an observable impact under the whole-stage codegen framework. The reason is those evaluation are actually needed and can't be avoided in most of (if not all) cases. From the benchmark we can see there is no negative impact even in the cases where no long consume function exists.

Yeah, I think the simplifies for the use of continue is a good thing. Personally I'd like to have this part merged first individually and so I can simplify #18931.

gatorsmile · 2017-09-24T18:57:42Z

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

@@ -328,10 +325,11 @@ case class BroadcastHashJoinExec(
         |  UnsafeRow $matched = $matches != null && $matches.hasNext() ?
         |    (UnsafeRow) $matches.next() : null;
         |  ${checkCondition.trim}


nvm. This is for outer join. The same name but different value.

gatorsmile · 2017-09-24T19:01:35Z

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

@@ -186,8 +186,7 @@ case class BroadcastHashJoinExec(
   */
  private def getJoinCondition(
      ctx: CodegenContext,
-      input: Seq[ExprCode],
-      anti: Boolean = false): (String, String, Seq[ExprCode]) = {


I like this change.

so we never set it to true?

We used to in https://github.com/apache/spark/pull/19324/files#diff-4455c05ddcdb096c36d9e0bd326dfe12L389, we don't anymore.

nvm, I saw the refactor in codegenAnti, cool!

gatorsmile · 2017-09-24T19:03:14Z

LGTM

gatorsmile · 2017-09-25T16:50:45Z

retest this please

SparkQA · 2017-09-25T19:36:24Z

Test build #82153 has finished for PR 19324 at commit ca64368.

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

gatorsmile · 2017-09-25T19:49:57Z

Merged to master

cloud-fan · 2017-09-26T09:26:53Z

...atalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala

+   * Add extra source code to the outermost generated class.
+   * @param code verbatim source code to be added.
+   */
+  def addExtraCode(code: String): Unit = {


I'd call it addInnerClass, as ideally you can't add arbitrary code to outer class.

+1 Although it doesn't prevent you going to add functions, but we have addNewFunction for it. So we'd better claim that this is just for inner class.

cloud-fan · 2017-09-26T11:44:25Z

sql/core/src/main/scala/org/apache/spark/sql/execution/WholeStageCodegenExec.scala

   *   # code to evaluate the predicate expression, result is isNull1 and value2
-   *   if (isNull1 || !value2) continue;
-   *   # call consume(), which will call parent.doConsume()
+   *   if (!isNull1 && value2) {


this may lead to deeply nested code, but I don't have a better idea for now.

in reality the filter code generates a do { } while(false) with continue inside to jump out, just like it did before. There's appropriate comment to it there.
I didn't want to complicate this example here, so changing the "will generate" to "could generate" is intentional to kind of show that it could, but not necessarily will :-)

cloud-fan · 2017-09-26T12:00:53Z

sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala

-           |   ${consume(ctx, Seq.empty, {generateRow.value})}
+           |   ${generateKeyRow.code}
+           |   ${generateBufferRow.code}
+           |   $outputFunc(${generateKeyRow.value}, ${generateBufferRow.value});


we didn't call outputCode before, are you fixing a potential bug?

generateRow.code was doing the job of outputCode before - i.e. putting all expected output into one UnsafeRow, from which the parent can consume it.

cloud-fan · 2017-09-26T12:05:15Z

sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala

+      // resultExpressions are Attributes of groupingExpressions and aggregateBufferAttributes.
+      assert(resultExpressions.forall(_.isInstanceOf[Attribute]))
+      assert(resultExpressions.length ==
+        groupingExpressions.length + aggregateBufferAttributes.length)


why we don't have these 2 requirements for the modes.contains(Final) || modes.contains(Complete) branch?

Final/Complete aggregations can have arbitrary projections in their resultExpressions, while partial aggregations are always constructed with only the grouping keys and aggregate expressions. The code that was here before with the UnsafeRowJoiner was using this assumption, so now I put it into assertion.

cloud-fan · 2017-09-26T12:06:57Z

sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala

       """
-
    } else {
      // generate result based on grouping key


we only go to this branch when aggregateExpressions is empty, is that possible?

Yes, e.g. for aggregation coming from Distinct.

cloud-fan · 2017-09-26T12:15:07Z

sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala

@@ -201,11 +201,14 @@ case class FilterExec(condition: Expression, child: SparkPlan)
      ev
    }

+    // Note: wrap in "do { } while(false);", so the generated checks can jump out with "continue;"


this is tricky, how hard it is to fix all places that use continue?

ah i see, you are trying to avoid generating deeply nested if-else branches.

genPredicate and generated ~50 lines above would have to be rewritten to now use continue. As you pointed in a previous comment, that would potentially lead to very nested scopes. Shouldn't be a problem for the compiler; for code generation the genPredicate would have to maintain these scopes and where to end them - i.e. wherever it not places a continue, it would have to open a nested scope, and then it would have to be closed in a correct place.

cloud-fan · 2017-09-26T12:18:18Z

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

-         |$numOutput.add(1);
-         |${consume(ctx, resultVars)}
+         |if ($matched != null) {
+         |  $checkCondition {


cloud-fan · 2017-09-26T12:26:25Z

a late LGTM :)

viirya · 2017-09-26T13:41:16Z

...atalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala

+   */
+  def addExtraCode(code: String): Unit = {
+    extraCode.append(code)
+    classSize(outerClassName) += code.length


The classSize is mainly used to deal with the limit of number of named constants. So I think we don't need to add extra code size into it, if we only add inner class?

Move HashAggregateExec parent consume to a separate function in codegen

ca64368

juliuszsompolski commented Sep 22, 2017

View reviewed changes

gatorsmile reviewed Sep 24, 2017

View reviewed changes

asfgit closed this in 038b185 Sep 25, 2017

viirya mentioned this pull request Sep 26, 2017

[SPARK-21717][SQL] Decouple consume functions of physical operators in whole-stage codegen #18931

Closed

cloud-fan reviewed Sep 26, 2017

View reviewed changes

viirya reviewed Sep 26, 2017

View reviewed changes

juliuszsompolski mentioned this pull request Sep 26, 2017

[SPARK-22103][FOLLOWUP] Rename addExtraCode to addInnerClass #19353

Closed

		} else if (modes.contains(Partial) \|\| modes.contains(PartialMerge)) {
		// This should be the last operator in a stage, we should output UnsafeRow directly

[SPARK-22103] Move HashAggregateExec parent consume to a separate function in codegen #19324

[SPARK-22103] Move HashAggregateExec parent consume to a separate function in codegen #19324

Uh oh!

Conversation

juliuszsompolski commented Sep 22, 2017

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

juliuszsompolski commented Sep 22, 2017

Uh oh!

SparkQA commented Sep 22, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

juliuszsompolski Sep 22, 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!

juliuszsompolski commented Sep 22, 2017

Uh oh!

viirya commented Sep 24, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

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!

Choose a reason for hiding this comment

Uh oh!

gatorsmile commented Sep 24, 2017

Uh oh!

gatorsmile commented Sep 25, 2017

Uh oh!

SparkQA commented Sep 25, 2017

Uh oh!

gatorsmile commented Sep 25, 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!

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!

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!

cloud-fan commented Sep 26, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

juliuszsompolski Sep 22, 2017 •

edited

Loading

viirya commented Sep 24, 2017 •

edited

Loading