[SPARK-30027][SQL] Support codegen for aggregate filters in HashAggregateExec #27019

maropu · 2019-12-26T13:35:43Z

What changes were proposed in this pull request?

This pr intends to support code generation for HashAggregateExec with filters.

Quick benchmark results:

$ ./bin/spark-shell --master=local[1] --conf spark.driver.memory=8g --conf spark.sql.shuffle.partitions=1 -v

scala> spark.range(100000000).selectExpr("id % 3 as k1", "id % 5 as k2", "rand() as v1", "rand() as v2").write.saveAsTable("t")
scala> sql("SELECT k1, k2, AVG(v1) FILTER (WHERE v2 > 0.5) FROM t GROUP BY k1, k2").write.format("noop").mode("overwrite").save()

>> Before this PR
Elapsed time: 16.170697619s 

>> After this PR
Elapsed time: 6.7825313s

The query above is compiled into code below;

...
/* 285 */   private void agg_doAggregate_avg_0(boolean agg_exprIsNull_2_0, org.apache.spark.sql.catalyst.InternalRow agg_unsafeRowAggBuffer_0, double agg_expr_2_0) throws java.io.IOException {
/* 286 */     // evaluate aggregate function for avg
/* 287 */     boolean agg_isNull_10 = true;
/* 288 */     double agg_value_12 = -1.0;
/* 289 */     boolean agg_isNull_11 = agg_unsafeRowAggBuffer_0.isNullAt(0);
/* 290 */     double agg_value_13 = agg_isNull_11 ?
/* 291 */     -1.0 : (agg_unsafeRowAggBuffer_0.getDouble(0));
/* 292 */     if (!agg_isNull_11) {
/* 293 */       agg_agg_isNull_12_0 = true;
/* 294 */       double agg_value_14 = -1.0;
/* 295 */       do {
/* 296 */         if (!agg_exprIsNull_2_0) {
/* 297 */           agg_agg_isNull_12_0 = false;
/* 298 */           agg_value_14 = agg_expr_2_0;
/* 299 */           continue;
/* 300 */         }
/* 301 */
/* 302 */         if (!false) {
/* 303 */           agg_agg_isNull_12_0 = false;
/* 304 */           agg_value_14 = 0.0D;
/* 305 */           continue;
/* 306 */         }
/* 307 */
/* 308 */       } while (false);
/* 309 */
/* 310 */       agg_isNull_10 = false; // resultCode could change nullability.
/* 311 */
/* 312 */       agg_value_12 = agg_value_13 + agg_value_14;
/* 313 */
/* 314 */     }
/* 315 */     boolean agg_isNull_15 = false;
/* 316 */     long agg_value_17 = -1L;
/* 317 */     if (!false && agg_exprIsNull_2_0) {
/* 318 */       boolean agg_isNull_18 = agg_unsafeRowAggBuffer_0.isNullAt(1);
/* 319 */       long agg_value_20 = agg_isNull_18 ?
/* 320 */       -1L : (agg_unsafeRowAggBuffer_0.getLong(1));
/* 321 */       agg_isNull_15 = agg_isNull_18;
/* 322 */       agg_value_17 = agg_value_20;
/* 323 */     } else {
/* 324 */       boolean agg_isNull_19 = true;
/* 325 */       long agg_value_21 = -1L;
/* 326 */       boolean agg_isNull_20 = agg_unsafeRowAggBuffer_0.isNullAt(1);
/* 327 */       long agg_value_22 = agg_isNull_20 ?
/* 328 */       -1L : (agg_unsafeRowAggBuffer_0.getLong(1));
/* 329 */       if (!agg_isNull_20) {
/* 330 */         agg_isNull_19 = false; // resultCode could change nullability.
/* 331 */
/* 332 */         agg_value_21 = agg_value_22 + 1L;
/* 333 */
/* 334 */       }
/* 335 */       agg_isNull_15 = agg_isNull_19;
/* 336 */       agg_value_17 = agg_value_21;
/* 337 */     }
/* 338 */     // update unsafe row buffer
/* 339 */     if (!agg_isNull_10) {
/* 340 */       agg_unsafeRowAggBuffer_0.setDouble(0, agg_value_12);
/* 341 */     } else {
/* 342 */       agg_unsafeRowAggBuffer_0.setNullAt(0);
/* 343 */     }
/* 344 */
/* 345 */     if (!agg_isNull_15) {
/* 346 */       agg_unsafeRowAggBuffer_0.setLong(1, agg_value_17);
/* 347 */     } else {
/* 348 */       agg_unsafeRowAggBuffer_0.setNullAt(1);
/* 349 */     }
/* 350 */   }
...

Why are the changes needed?

For high performance.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Existing tests.

SparkQA · 2019-12-26T17:32:54Z

Test build #115814 has finished for PR 27019 at commit 0f75cab.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
trait GeneratePredicateHelper extends PredicateHelper
case class FilterExec(condition: Expression, child: SparkPlan)

HeartSaVioR · 2019-12-27T02:20:20Z

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

@@ -153,9 +153,7 @@ case class HashAggregateExec(

  override def supportCodegen: Boolean = {
    // ImperativeAggregate and filter predicate are not supported right now


nit: Let's also update the comment here as well.

ok, thanks.

SparkQA · 2019-12-28T05:54:52Z

Test build #115878 has finished for PR 27019 at commit e449d3c.

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

cloud-fan · 2020-01-02T15:15:12Z

cc @rednaxelafx as well

beliefer · 2020-01-07T07:42:38Z

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

@@ -91,60 +91,39 @@ case class ProjectExec(projectList: Seq[NamedExpression], child: SparkPlan)
  }
 }

-
-/** Physical plan for Filter. */
-case class FilterExec(condition: Expression, child: SparkPlan)


@maropu I don't understand why we need change FilterExec ?

Just for sharing code to process predicates between aggregates and filters.

maropu · 2020-01-10T07:34:16Z

retest this please

SparkQA · 2020-01-10T08:05:01Z

Test build #116458 has finished for PR 27019 at commit e449d3c.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

maropu · 2020-01-10T08:08:30Z

retest this please

SparkQA · 2020-01-10T12:26:26Z

Test build #116467 has finished for PR 27019 at commit e449d3c.

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

maropu · 2020-01-15T07:34:36Z

retest this please

SparkQA · 2020-01-15T08:05:02Z

Test build #116759 has finished for PR 27019 at commit e449d3c.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

maropu · 2020-01-15T11:20:29Z

retest this please

SparkQA · 2020-01-15T15:38:36Z

Test build #116780 has finished for PR 27019 at commit e449d3c.

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

kiszk · 2020-01-19T01:20:29Z

Would it be possible to add benchmark result?

kiszk · 2020-01-19T01:21:54Z

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

+             |  ${generatePredicateCode(ctx, condition, inputAttrs, input)}
+             |  $aggCode
+             |} while(false);
+         """.stripMargin


nit: indentation problem?

kiszk · 2020-01-19T01:22:48Z

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

+               |  ${generatePredicateCode(ctx, condition, inputAttrs, input)}
+               |  $aggCode
+               |} while(false);
+           """.stripMargin


kiszk · 2020-01-19T01:22:58Z

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

+                   |  ${generatePredicateCode(ctx, condition, inputAttrs, input)}
+                   |  $aggCode
+                   |} while(false);
+               """.stripMargin


maropu · 2020-01-20T04:23:51Z

@kiszk I added the performance numbers in the PR description. I think the codegen for hash-aggregates can have performance gains in most queries. But, aggregate filters (recently merged in the master) forcibly disable the codegen. So, I think this fix has a good effect on performance.

SparkQA · 2020-01-20T06:44:46Z

Test build #117048 has finished for PR 27019 at commit c064d58.

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

kiszk · 2020-01-20T07:55:02Z

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

@@ -375,16 +373,30 @@ case class HashAggregateExec(
       """.stripMargin
    }

-    val codeToEvalAggFunc = if (conf.codegenSplitAggregateFunc &&
+    val codeToEvalAggFunc = {


Can we refactor this and that to use one common function?

Yea, I'll update.

SparkQA · 2020-01-21T05:11:59Z

Test build #117150 has finished for PR 27019 at commit ba48342.

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

maropu · 2020-01-21T05:17:43Z

retest this please

SparkQA · 2020-01-21T06:55:40Z

Test build #117144 has finished for PR 27019 at commit 5a2114f.

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

kiszk · 2020-01-21T07:52:51Z

cc @rednaxelafx @cloud-fan

SparkQA · 2020-01-21T08:05:02Z

Test build #117154 has finished for PR 27019 at commit ba48342.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

kiszk · 2020-01-21T09:19:43Z

retest this please

SparkQA · 2020-01-21T13:41:21Z

Test build #117174 has finished for PR 27019 at commit ba48342.

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

maropu · 2020-08-01T01:03:20Z

retest this please

SparkQA · 2020-08-01T05:22:01Z

Test build #126907 has finished for PR 27019 at commit 464fbaa.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
trait GeneratePredicateHelper extends PredicateHelper
case class FilterExec(condition: Expression, child: SparkPlan)

HyukjinKwon · 2020-08-26T04:51:19Z

retest this please

SparkQA · 2020-08-26T07:05:02Z

Test build #127912 has finished for PR 27019 at commit 464fbaa.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds the following public classes (experimental):
trait GeneratePredicateHelper extends PredicateHelper
case class FilterExec(condition: Expression, child: SparkPlan)

SparkQA · 2020-08-27T03:50:34Z

Test build #127941 has finished for PR 27019 at commit a43aa25.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
trait GeneratePredicateHelper extends PredicateHelper
case class FilterExec(condition: Expression, child: SparkPlan)

SparkQA · 2020-09-10T12:54:37Z

Test build #128517 has finished for PR 27019 at commit 9b1aea0.

This patch fails to build.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2020-09-10T13:02:39Z

Test build #128519 has finished for PR 27019 at commit 26ce9b2.

This patch fails to build.
This patch merges cleanly.
This patch adds the following public classes (experimental):
trait GeneratePredicateHelper extends PredicateHelper
case class FilterExec(condition: Expression, child: SparkPlan)

SparkQA · 2020-09-10T21:29:05Z

Test build #128525 has finished for PR 27019 at commit 009fe4a.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
trait GeneratePredicateHelper extends PredicateHelper
case class FilterExec(condition: Expression, child: SparkPlan)

github-actions · 2020-12-20T00:53:15Z

We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!

HeartSaVioR · 2020-12-21T00:05:10Z

@maropu Looks like this PR is not properly tracked. I've removed the Stale tag assuming you want to move forward, but wanted to check again, and curious how you handle this.

maropu · 2020-12-21T01:38:48Z

Thanks for taking care of it, @HeartSaVioR. If there are no reviewer who is against this PR and one has no more comment, I will merge this for the v3.2.0 release in a few weeks. cc: @cloud-fan @viirya @dongjoon-hyun

SparkQA · 2020-12-21T02:23:25Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/37719/

SparkQA · 2020-12-21T02:28:02Z

Test build #133120 has finished for PR 27019 at commit 86d89ba.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
trait GeneratePredicateHelper extends PredicateHelper
case class FilterExec(condition: Expression, child: SparkPlan)

SparkQA · 2020-12-21T02:53:46Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/37719/

maropu · 2020-12-22T01:23:52Z

retest this please

SparkQA · 2020-12-22T02:19:48Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/37786/

SparkQA · 2020-12-22T02:24:21Z

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/37786/

SparkQA · 2020-12-22T05:59:05Z

Test build #133188 has finished for PR 27019 at commit 86d89ba.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
trait GeneratePredicateHelper extends PredicateHelper
case class FilterExec(condition: Expression, child: SparkPlan)

dongjoon-hyun

+1, LGTM. Sorry for the delay and thank you, @maropu .
Merged to master for Apache Spark 3.2.0.

Merry Christmas and Happy New Year!

maropu · 2020-12-25T00:04:40Z

Thanks for your review, Dongjoon ! Yea, you, too. Happy Merry Christmas!

HeartSaVioR reviewed Dec 27, 2019

View reviewed changes

beliefer reviewed Jan 7, 2020

View reviewed changes

kiszk reviewed Jan 19, 2020

View reviewed changes

kiszk reviewed Jan 20, 2020

View reviewed changes

maropu force-pushed the AggregateFilterCodegen branch from 5a2114f to ba48342 Compare January 21, 2020 03:25

kiszk approved these changes Jan 21, 2020

View reviewed changes

maropu force-pushed the AggregateFilterCodegen branch from ba48342 to 2796407 Compare January 26, 2020 11:30

maropu force-pushed the AggregateFilterCodegen branch from 464fbaa to a43aa25 Compare August 26, 2020 23:21

maropu force-pushed the AggregateFilterCodegen branch 3 times, most recently from 9b1aea0 to 26ce9b2 Compare September 10, 2020 12:01

maropu force-pushed the AggregateFilterCodegen branch from 26ce9b2 to 009fe4a Compare September 10, 2020 13:18

github-actions bot added the Stale label Dec 20, 2020

HeartSaVioR removed the Stale label Dec 21, 2020

Support codegen for agg filters in HashAggregateExec

86d89ba

maropu force-pushed the AggregateFilterCodegen branch from 009fe4a to 86d89ba Compare December 21, 2020 01:40

dongjoon-hyun approved these changes Dec 24, 2020

View reviewed changes

dongjoon-hyun closed this in 65a9ac2 Dec 24, 2020

		@@ -153,9 +153,7 @@ case class HashAggregateExec(

		override def supportCodegen: Boolean = {
		// ImperativeAggregate and filter predicate are not supported right now

[SPARK-30027][SQL] Support codegen for aggregate filters in HashAggregateExec #27019

[SPARK-30027][SQL] Support codegen for aggregate filters in HashAggregateExec #27019

Uh oh!

Conversation

maropu commented Dec 26, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Uh oh!

SparkQA commented Dec 26, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Dec 28, 2019

Uh oh!

cloud-fan commented Jan 2, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

maropu commented Jan 10, 2020

Uh oh!

SparkQA commented Jan 10, 2020

Uh oh!

maropu commented Jan 10, 2020

Uh oh!

SparkQA commented Jan 10, 2020

Uh oh!

maropu commented Jan 15, 2020

Uh oh!

SparkQA commented Jan 15, 2020

Uh oh!

maropu commented Jan 15, 2020

Uh oh!

SparkQA commented Jan 15, 2020

Uh oh!

kiszk commented Jan 19, 2020

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!

maropu commented Jan 20, 2020

Uh oh!

SparkQA commented Jan 20, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Jan 21, 2020

Uh oh!

maropu commented Jan 21, 2020

Uh oh!

SparkQA commented Jan 21, 2020

Uh oh!

kiszk commented Jan 21, 2020

Uh oh!

SparkQA commented Jan 21, 2020

Uh oh!

kiszk commented Jan 21, 2020

Uh oh!

SparkQA commented Jan 21, 2020

Uh oh!

maropu commented Aug 1, 2020

Uh oh!

SparkQA commented Aug 1, 2020

Uh oh!

HyukjinKwon commented Aug 26, 2020

Uh oh!

SparkQA commented Aug 26, 2020

Uh oh!

SparkQA commented Aug 27, 2020

maropu commented Dec 26, 2019 •

edited

Loading

maropu commented Dec 21, 2020 •

edited

Loading