Implement hash partitioned aggregation #320

Dandandan · 2021-05-11T21:26:01Z

Which issue does this PR close?

Closes #27

Rationale for this change

A more scalable hash aggregate that works well for group by expressions that have high cardinality int he output and allows to scale better with the number of cpu cores.
The algorithm changes the steps from:

partial hash aggregate -> merge (to 1 partition) -> full hash aggregate

to

partial hash aggregate -> repartition on group by expressions -> hash aggregate (on partitions)

This is the same as what Spark is doing.

This mostly has an effect on group by with higher cardinality, but no substantial effect on lower cardinality (as partial result is already small).
For Ballista this would also be required I think @andygrove - currently every partition is merged into one which can be problematic (and slow).

For example, TPC-H query 3 benefits quite a bit:
Master:

Query 3 avg time: 71.82 ms

PR:

Query 3 avg time: 49.20 ms

The db-benchmark group by queries improve by quite a bit:

Master:

q1 took 34 ms
q2 took 454 ms
q3 took 3284 ms
q4 took 42 ms
q5 took 2924 ms
q7 took 2843 ms

PR

q1 took 33 ms
q2 took 369 ms
q3 took 1875 ms
q4 took 46 ms
q5 took 1756 ms
q7 took 1686 ms

What changes are included in this PR?

Are there any user-facing changes?

codecov-commenter · 2021-05-11T22:42:41Z

Codecov Report

Merging #320 (2fc12eb) into master (1702d6c) will decrease coverage by 0.00%.
The diff coverage is 76.47%.

@@            Coverage Diff             @@
##           master     apache/arrow-datafusion#320      +/-   ##
==========================================
- Coverage   75.72%   75.71%   -0.01%     
==========================================
  Files         143      143              
  Lines       23832    23881      +49     
==========================================
+ Hits        18046    18081      +35     
- Misses       5786     5800      +14

Impacted Files	Coverage Δ
...ta/rust/core/src/serde/physical_plan/from_proto.rs	`47.39% <0.00%> (-0.42%)`	⬇️
...ista/rust/core/src/serde/physical_plan/to_proto.rs	`50.62% <0.00%> (-0.32%)`	⬇️
ballista/rust/core/src/utils.rs	`30.43% <ø> (ø)`
datafusion/src/physical_plan/hash_join.rs	`85.57% <0.00%> (-0.83%)`	⬇️
datafusion/src/physical_plan/mod.rs	`84.70% <ø> (ø)`
...atafusion/src/physical_plan/unicode_expressions.rs	`90.37% <ø> (ø)`
datafusion/tests/sql.rs	`99.88% <ø> (ø)`
datafusion/src/physical_plan/planner.rs	`80.62% <86.36%> (+0.62%)`	⬆️
ballista/rust/scheduler/src/planner.rs	`69.46% <100.00%> (ø)`
ballista/rust/scheduler/src/test_utils.rs	`100.00% <100.00%> (ø)`
... and 7 more

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 1702d6c...2fc12eb. Read the comment docs.

ballista/rust/scheduler/src/test_utils.rs

Dandandan · 2021-05-12T05:40:42Z

datafusion/src/physical_plan/planner.rs

-                    input_schema,
-                )?))
+                // TODO: dictionary type not yet supported in Hash Repartition
+                let contains_dict = groups


Will create an issue for this

alamb

This looks really cool @Dandandan. I suggest you add a test, if possible, that shows the plan with the repartition exec operation in order to prevent someone accidentally turning off this optimization during a refactor

datafusion/src/execution/context.rs

alamb · 2021-05-12T10:28:42Z

datafusion/src/physical_plan/planner.rs

-                    input_schema,
-                )?))
+                // TODO: dictionary type not yet supported in Hash Repartition
+                let contains_dict = groups


datafusion/src/physical_plan/hash_aggregate.rs

datafusion/src/physical_plan/hash_join.rs

Co-authored-by: Andrew Lamb <andrew@nerdnetworks.org>

… into agg_partition

Dandandan · 2021-05-12T17:46:37Z

This looks really cool @Dandandan. I suggest you add a test, if possible, that shows the plan with the repartition exec operation in order to prevent someone accidentally turning off this optimization during a refactor

Good idea! Added a test for this

alamb

I think it is looking great. Thanks @Dandandan

alamb · 2021-05-12T21:14:54Z

datafusion/src/physical_plan/hash_aggregate.rs

@@ -202,6 +209,9 @@ impl ExecutionPlan for HashAggregateExec {
    fn required_child_distribution(&self) -> Distribution {
        match &self.mode {
            AggregateMode::Partial => Distribution::UnspecifiedDistribution,
+            AggregateMode::FinalPartitioned => Distribution::HashPartitioned(
+                self.group_expr.iter().map(|x| x.0.clone()).collect(),


jorgecarleitao

I went through this and it looks great. Thanks a lot @Dandandan , great work. 💯

from my side.

We may want to give some time in case someone else would like to go through this before merging.

❤️

Dandandan · 2021-05-14T14:36:46Z

@andygrove maybe? :)

andygrove · 2021-05-14T15:30:10Z

@andygrove maybe? :)

I will make time this weekend to review this and take it for a spin!

Dandandan · 2021-05-14T15:33:16Z

Awesome, thanks @andygrove !

I have also some nice followup this weekend for more performance improvements for hash aggregates :D

andygrove

LGTM. I tested this out locally and confirmed that performance is much better for unpartitioned data, and about the same for partitioned data.

andygrove · 2021-05-15T13:50:17Z

@Dandandan Looks like there is a conflict that needs fixing

andygrove · 2021-05-15T16:36:05Z

I filed apache/datafusion-ballista#23 for implementing this optimization in Ballista

Dandandan · 2021-05-15T19:03:42Z

Somehow coverage run seems to fail... But doesn't show what is failing

alamb · 2021-05-16T10:05:03Z

The https://github.com/apache/arrow-datafusion/pull/320/checks?check_run_id=2591763548 shows this buried in the logs (not at the end, annoyingly):

failures:

---- physical_plan::planner::tests::hash_agg_group_by_partitioned stdout ----
thread 'physical_plan::planner::tests::hash_agg_group_by_partitioned' panicked at 'assertion failed: formatted.contains(\"FinalPartitioned\")', datafusion/src/physical_plan/planner.rs:1051:9
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace


failures:
    physical_plan::planner::tests::hash_agg_group_by_partitioned

Dandandan · 2021-05-16T19:10:54Z

thanks @alamb will merge it now when it's green

Dandandan · 2021-05-16T19:33:20Z

Thanks all 🎉

Dandandan added 6 commits May 11, 2021 23:24

Implement hash partitioned aggregation

c90bfc9

Ballista

3dc2cf4

Make configurable and use configured concurrency

381df9c

WIP

9dd926f

Add some hash types

268eb1e

Fmt

358b21b

Dandandan changed the title ~~WIP: Implement hash partitioned aggregation~~ Implement hash partitioned aggregation May 11, 2021

Disable repartition aggregations in ballista

6824f1b

Dandandan added 2 commits May 12, 2021 00:56

fmt

207dbf4

Clippy, ballista

6ea59c1

Dandandan requested review from alamb and andygrove May 11, 2021 23:28

Fix test

a0db9ed

andygrove reviewed May 12, 2021

View reviewed changes

ballista/rust/scheduler/src/test_utils.rs Show resolved Hide resolved

Revert test ode

edf9396

Dandandan requested a review from andygrove May 12, 2021 05:39

Dandandan commented May 12, 2021

View reviewed changes

alamb approved these changes May 12, 2021

View reviewed changes

Dandandan and others added 4 commits May 12, 2021 12:55

Update datafusion/src/physical_plan/hash_aggregate.rs

f258008

Co-authored-by: Andrew Lamb <andrew@nerdnetworks.org>

Add info about required child partitioning

96b5a96

Merge branch 'agg_partition' of github.com:Dandandan/arrow-datafusion…

6c9093b

… into agg_partition

Add test

e13e345

alamb approved these changes May 12, 2021

View reviewed changes

jorgecarleitao approved these changes May 12, 2021

View reviewed changes

Dandandan mentioned this pull request May 14, 2021

Speed up create_batch_from_map #339

Merged

andygrove approved these changes May 15, 2021

View reviewed changes

Solve merge conflict

8f4522d

andygrove mentioned this pull request May 19, 2022

Implement hash partitioned aggregation in Ballista apache/datafusion-ballista#23

Open

Test fix

48c8b93

Set concurrency

2fc12eb

Dandandan merged commit ed92673 into apache:master May 16, 2021

houqp added datafusion Changes in the datafusion crate enhancement New feature or request performance labels Jul 29, 2021

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Implement hash partitioned aggregation #320

Implement hash partitioned aggregation #320

Dandandan commented May 11, 2021 •

edited

Loading

codecov-commenter commented May 11, 2021 •

edited

Loading

Dandandan May 12, 2021

alamb May 12, 2021

alamb left a comment

alamb May 12, 2021

Dandandan commented May 12, 2021

alamb left a comment

alamb May 12, 2021

jorgecarleitao left a comment

Dandandan commented May 14, 2021

andygrove commented May 14, 2021

Dandandan commented May 14, 2021

andygrove left a comment

andygrove commented May 15, 2021

andygrove commented May 15, 2021

Dandandan commented May 15, 2021 •

edited

Loading

alamb commented May 16, 2021

Dandandan commented May 16, 2021

Dandandan commented May 16, 2021

Implement hash partitioned aggregation #320

Implement hash partitioned aggregation #320

Conversation

Dandandan commented May 11, 2021 • edited Loading

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Are there any user-facing changes?

codecov-commenter commented May 11, 2021 • edited Loading

Codecov Report

Dandandan May 12, 2021

Choose a reason for hiding this comment

alamb May 12, 2021

Choose a reason for hiding this comment

alamb left a comment

Choose a reason for hiding this comment

alamb May 12, 2021

Choose a reason for hiding this comment

Dandandan commented May 12, 2021

alamb left a comment

Choose a reason for hiding this comment

alamb May 12, 2021

Choose a reason for hiding this comment

jorgecarleitao left a comment

Choose a reason for hiding this comment

Dandandan commented May 14, 2021

andygrove commented May 14, 2021

Dandandan commented May 14, 2021

andygrove left a comment

Choose a reason for hiding this comment

andygrove commented May 15, 2021

andygrove commented May 15, 2021

Dandandan commented May 15, 2021 • edited Loading

alamb commented May 16, 2021

Dandandan commented May 16, 2021

Dandandan commented May 16, 2021

Dandandan commented May 11, 2021 •

edited

Loading

codecov-commenter commented May 11, 2021 •

edited

Loading

Dandandan commented May 15, 2021 •

edited

Loading