[Improvement][AQE] Avoid calling getShuffleResult multiple times #190

leixm · 2022-08-27T04:45:21Z

###What changes were proposed in this pull request?
For issue #136 , When we use AQE, we may call shuffleWriteClient.getShuffleResult multiple times. But if both partition 1 and partition 2 are on the server A, we call getShuffleResult(partition 1) to get data form server A, and then we call getShuffleResult(partition 2) to get data form server A, it's not necassray. We can get getShuffleResult(partition 1, partition 2) instead.

###Why are the changes needed?
Improve getShuffleResult

###Does this PR introduce any user-facing change?
No

###How was this patch tested?
Added UT

leixm · 2022-08-27T04:48:16Z

@jerqi can you help review this pr plz?

codecov-commenter · 2022-08-27T05:02:00Z

Codecov Report

Attention: Patch coverage is 42.57426% with 58 lines in your changes missing coverage. Please review.

Project coverage is 58.99%. Comparing base (5eb2758) to head (08c858f).
Report is 922 commits behind head on master.

Files with missing lines	Patch %	Lines
...pache/uniffle/server/ShuffleServerGrpcService.java	0.00%	29 Missing ⚠️
...he/uniffle/client/impl/ShuffleWriteClientImpl.java	0.00%	28 Missing ⚠️
.../java/org/apache/uniffle/common/util/RssUtils.java	95.45%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master     #190      +/-   ##
============================================
+ Coverage     58.40%   58.99%   +0.58%     
- Complexity     1273     1325      +52     
============================================
  Files           158      160       +2     
  Lines          8446     8699     +253     
  Branches        784      815      +31     
============================================
+ Hits           4933     5132     +199     
- Misses         3260     3303      +43     
- Partials        253      264      +11

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

jerqi · 2022-08-27T07:22:20Z

Is it a compatible feature? If this is a breaking change, we should dicuss about it in the mail list first. But it seems to become a compatible feature if you adjust your implement.

leixm · 2022-08-27T09:16:08Z

I think it's a compatible feature, could you give me more detail? @jerqi

leixm · 2022-08-27T09:18:11Z

Or we should keep the original implementation getShuffleResult(partitionId), and then add a new implementation getShuffleResult(startPartition, endPartition)?

jerqi · 2022-08-27T09:21:17Z

A compatible feature means that old version server can use the new version server, the new version client can use the old version server. You change the protobuf field name, it will cause incompatibility in my thought.

leixm · 2022-08-27T09:24:39Z

Thank you for your reminder, i will make it compatible.

leixm · 2022-08-28T13:23:50Z

@jerqi can you help review this pr plz?

leixm · 2022-08-28T13:24:50Z

org.apache.uniffle.common.security.HadoopSecurityContextTest test failed cause by Address already in use， It seems to be caused by running multiple tests at the same time

jerqi · 2022-08-28T13:56:49Z

@zuston Could you help me solve the problem of the flaky kerberos test?

proto/src/main/proto/Rss.proto

zuston · 2022-08-28T14:08:50Z

@zuston Could you help me solve the problem of the flaky kerberos test?

OK. Let me solve this tomorrow.

jerqi · 2022-08-28T14:24:21Z

I think we should test this with some SQL query in integration test.

leixm · 2022-08-28T14:28:38Z

You're right , I will add some test in integration test.

jerqi · 2022-08-29T04:03:15Z

Flaky test was fixed by #191 .Thanks @zuston

leixm · 2022-08-30T08:10:18Z

I think we should test this with some SQL query in integration test.

Already added integration test, Can you help review? Thank you @jerqi

jerqi · 2022-08-30T10:52:18Z

If the new version client access the old version server, what will happen? Could we fallback to use old way to access the old version server when there are exceptions?

integration-test/spark3/src/test/java/org/apache/uniffle/test/MockRssShuffleManager.java

leixm · 2022-08-30T11:18:25Z

If the new version client access the old version server, what will happen? Could we fallback to use old way to access the old version server when there are exceptions?

I think the new version of the client does not need to be compatible with the old version of the server.

jerqi · 2022-08-30T11:31:53Z

If the new version client access the old version server, what will happen? Could we fallback to use old way to access the old version server when there are exceptions?

I think the new version of the client does not need to be compatible with the old version of the server.

OK, I make a mistake. Maybe we just guarantee that old client can access the new server.

jerqi · 2022-08-30T11:40:23Z

proto/src/main/proto/Rss.proto

+  int32 shuffleId = 2;
+  repeated int32 partitions = 3;
+}
+


Should we add a new message for response?

common/src/main/java/org/apache/uniffle/common/util/RssUtils.java

integration-test/spark3/src/test/java/org/apache/uniffle/test/MockRssShuffleManager.java

jerqi · 2022-08-30T12:05:08Z

common/src/main/java/org/apache/uniffle/common/util/RssUtils.java

    return hostName.replaceAll("[\\.-]", "_");
  }
+
+  public static Map<Integer, Roaring64NavigableMap> shuffleBitmapToPartitionBitmap(


Why do we need startPartition and endPartition?

Because the partition may not have a block, but we need to fill it to avoid NPE when used, See

for (int partitionId = startPartition; partitionId <= endPartition; partitionId++) { result.putIfAbsent(partitionId, Roaring64NavigableMap.bitmapOf()); }

Because the partition may not have a block, but we need to fill it to avoid NPE when used, See

for (int partitionId = startPartition; partitionId <= endPartition; partitionId++) { result.putIfAbsent(partitionId, Roaring64NavigableMap.bitmapOf()); }

A little weird.

Maybe we should have a better name.

This comment isn't resolved.

jerqi · 2022-08-30T12:21:09Z

common/src/main/java/org/apache/uniffle/common/util/RssUtils.java

+  public static Map<Integer, Roaring64NavigableMap> shuffleBitmapToPartitionBitmap(
+      Roaring64NavigableMap shuffleBitmap, int startPartition, int endPartition) {
+    Map<Integer, Roaring64NavigableMap> result = Maps.newHashMap();
+    for (int partitionId = startPartition; partitionId <= endPartition; partitionId++) {


Why do we need <= instead of <?

We think we need <= here.

We think we need <= here.

Sorry, my mistake. We need < here

jerqi · 2022-08-30T12:42:39Z

Could you provide some data about benchmark or performance test?

leixm · 2022-08-30T14:10:14Z

Could you provide some data about benchmark or performance test?

Ok, I will do some performance tests to try to prove that the performance will improve in some extreme scenarios of AQE

leixm · 2022-09-07T12:39:51Z

Environment

Shuffle Server Num : 5
Shuffle Write: 48G
Configuration: --conf spark.sql.shuffle.partitions=5000 --conf spark.sql.adaptive.enabled=true --conf spark.sql.adaptive.shuffle.targetPostShuffleInputSize=64MB

We measure the performance of get_shuffle_result by the following metrics:

get_shuffle_result_times: The number of calls of the get_shuffle_result interface
get_shuffle_result_cost: Time consumption of get_shuffle_result interface
get_shuffle_result_for_multi_part_times:The number of calls of the get_shuffle_result_for_multi_part interface
get_shuffle_result_for_multi_part_cost: Time consumption of get_shuffle_result_for_multi_part interface

Test Results

Before issue_136

serverId	get_shuffle_result_times	get_shuffle_result_cost(ms)
Server1	1000	157614
Server2	1000	426897
Server3	1000	269488
Server4	1000	906758
Server5	1001	123217
sum	5001	1883974

After issue_136

serverId	get_shuffle_result_for_multi_part_times	get_shuffle_result_for_multi_part_cost(ms)
Server1	833	870720
Server2	833	260865
Server3	834	333202
Server4	833	90277
Server5	835	94113
sum	4168	1649177

Summarize

The number of interface requests is reduced by 16%, and the total time is reduced by 12.5%. If we assign consecutive partitions to a server, the improvement will be more obvious.

jerqi · 2022-09-07T12:46:27Z

Environment

Shuffle Server Num : 5 Shuffle Write: 48G Configuration: --conf spark.sql.shuffle.partitions=5000 --conf spark.sql.adaptive.enabled=true --conf spark.sql.adaptive.shuffle.targetPostShuffleInputSize=64MB

We measure the performance of get_shuffle_result by the following metrics:

get_shuffle_result_times: The number of calls of the get_shuffle_result interface

get_shuffle_result_cost: Time consumption of get_shuffle_result interface

get_shuffle_result_for_multi_part_times:The number of calls of the get_shuffle_result_for_multi_part interface

get_shuffle_result_for_multi_part_cost: Time consumption of get_shuffle_result_for_multi_part interface

Test Results

Before issue_136

serverId get_shuffle_result_times get_shuffle_result_cost(ms)
Server1 1000 157614
Server2 1000 426897
Server3 1000 269488
Server4 1000 906758
Server5 1001 123217
sum 5001 1883974
After issue_136

serverId get_shuffle_result_for_multi_part_times get_shuffle_result_for_multi_part_cost(ms)
Server1 833 870720
Server2 833 260865
Server3 834 333202
Server4 833 90277
Server5 835 94113
sum 4168 1649177

Summarize

The number of interface requests is reduced by 16%, and the total time is reduced by 12.5%. If we assign consecutive partitions to a server, the improvement will be more obvious.

What's your test case? TPC-DS?

leixm · 2022-09-07T12:50:06Z

No, this is our production task.

jerqi · 2022-09-07T14:01:29Z

Please resolve the comments left.

…reversePartitionToServers

leixm · 2022-09-08T03:40:25Z

Can you help review please? @jerqi

jerqi · 2022-09-08T03:44:08Z

common/src/main/java/org/apache/uniffle/common/util/RssUtils.java

    return hostName.replaceAll("[\\.-]", "_");
  }
+
+  public static Map<Integer, Roaring64NavigableMap> shuffleBitmapToPartitionBitmap(


This comment isn't resolved.

...tion-test/spark3/src/test/java/org/apache/uniffle/test/GetShuffleReportForMultiPartTest.java

common/src/test/java/org/apache/uniffle/common/util/RssUtilsTest.java

server/src/main/java/org/apache/uniffle/server/ShuffleTaskManager.java

jerqi

There are some comments that is not resolved.

sorry, i miss the unresolved comments.

leixm · 2022-09-08T14:05:34Z

Can you help review please? @jerqi

jerqi

LGTM, thanks @leixm @zuston

leixm · 2022-09-08T14:33:54Z

@jerqi @zuston Thank you very much, good brother.

jerqi · 2022-10-26T11:30:31Z

Environment

Shuffle Server Num : 5 Shuffle Write: 48G Configuration: --conf spark.sql.shuffle.partitions=5000 --conf spark.sql.adaptive.enabled=true --conf spark.sql.adaptive.shuffle.targetPostShuffleInputSize=64MB

We measure the performance of get_shuffle_result by the following metrics:

get_shuffle_result_times: The number of calls of the get_shuffle_result interface

get_shuffle_result_cost: Time consumption of get_shuffle_result interface

get_shuffle_result_for_multi_part_times:The number of calls of the get_shuffle_result_for_multi_part interface

get_shuffle_result_for_multi_part_cost: Time consumption of get_shuffle_result_for_multi_part interface

Test Results

Before issue_136

serverId get_shuffle_result_times get_shuffle_result_cost(ms)
Server1 1000 157614
Server2 1000 426897
Server3 1000 269488
Server4 1000 906758
Server5 1001 123217
sum 5001 1883974
After issue_136

serverId get_shuffle_result_for_multi_part_times get_shuffle_result_for_multi_part_cost(ms)
Server1 833 870720
Server2 833 260865
Server3 834 333202
Server4 833 90277
Server5 835 94113
sum 4168 1649177

Summarize

The number of interface requests is reduced by 16%, and the total time is reduced by 12.5%. If we assign consecutive partitions to a server, the improvement will be more obvious.

Could you raise a new issue that Coordinator support to assign consecutive partitions to a server?

leixm · 2022-10-26T11:43:42Z

Environment

Shuffle Server Num : 5 Shuffle Write: 48G Configuration: --conf spark.sql.shuffle.partitions=5000 --conf spark.sql.adaptive.enabled=true --conf spark.sql.adaptive.shuffle.targetPostShuffleInputSize=64MB
We measure the performance of get_shuffle_result by the following metrics:

get_shuffle_result_times: The number of calls of the get_shuffle_result interface

get_shuffle_result_cost: Time consumption of get_shuffle_result interface

get_shuffle_result_for_multi_part_times:The number of calls of the get_shuffle_result_for_multi_part interface

get_shuffle_result_for_multi_part_cost: Time consumption of get_shuffle_result_for_multi_part interface

Test Results

Before issue_136
serverId get_shuffle_result_times get_shuffle_result_cost(ms)
Server1 1000 157614
Server2 1000 426897
Server3 1000 269488
Server4 1000 906758
Server5 1001 123217
sum 5001 1883974
After issue_136
serverId get_shuffle_result_for_multi_part_times get_shuffle_result_for_multi_part_cost(ms)
Server1 833 870720
Server2 833 260865
Server3 834 333202
Server4 833 90277
Server5 835 94113
sum 4168 1649177

Summarize

The number of interface requests is reduced by 16%, and the total time is reduced by 12.5%. If we assign consecutive partitions to a server, the improvement will be more obvious.

Could you raise a new issue that Coordinator support to assign consecutive partitions to a server?

No problem, I'll follow up on this.

leixm force-pushed the issue_136 branch from 9a18532 to a6bb086 Compare August 28, 2022 09:35

jerqi reviewed Aug 28, 2022

View reviewed changes

proto/src/main/proto/Rss.proto Outdated Show resolved Hide resolved

[Improvement][AQE] Avoid calling getShuffleResult multiple times

8ce943d

leixm force-pushed the issue_136 branch from c9f56e1 to 8ce943d Compare August 30, 2022 06:25

leixianming and others added 2 commits August 30, 2022 14:40

Fix indentation

49f96e4

Merge branch 'apache:master' into issue_136

f286f81

jerqi reviewed Aug 30, 2022

View reviewed changes

integration-test/spark3/src/test/java/org/apache/uniffle/test/MockRssShuffleManager.java Outdated Show resolved Hide resolved

jerqi reviewed Aug 30, 2022

View reviewed changes

Move MockRssShuffleManager to GetShuffleReportForMultiPartTest & Fix …

e21c505

…reversePartitionToServers

jerqi reviewed Sep 8, 2022

View reviewed changes

leixianming added 5 commits September 8, 2022 11:59

Fix

d23d8ef

Fix

9af4adf

Fix

e51e6e5

Fix wrong use of endPartition.

c668113

Fix UT

a17df24

jerqi reviewed Sep 8, 2022

View reviewed changes

server/src/main/java/org/apache/uniffle/server/ShuffleTaskManager.java Show resolved Hide resolved

Remove redundant or.

ec0e360

jerqi previously approved these changes Sep 8, 2022

View reviewed changes

leixianming added 2 commits September 8, 2022 18:09

Fix shuffleBitmapToPartitionBitmap.

85e9400

Add GetShuffleResultForMultiPartResponse.

08c858f

jerqi approved these changes Sep 8, 2022

View reviewed changes

jerqi merged commit 6aa4379 into apache:master Sep 8, 2022

jerqi mentioned this pull request Sep 9, 2022

[Improvement][Aqe] Avoid calling getShuffleResult multiple times #136

Closed

jerqi mentioned this pull request Oct 10, 2022

[Bug] No data can be read from ShuffleServer when the reduce task number exceeds 1024 #257

Closed

3 tasks

leixm mentioned this pull request Nov 11, 2022

[ISSUE-135][FOLLOWUP][Improvement][AQE] Assign adjacent partitions to the same ShuffleServer #307

Merged

[Improvement][AQE] Avoid calling getShuffleResult multiple times #190

[Improvement][AQE] Avoid calling getShuffleResult multiple times #190

Uh oh!

Conversation

leixm commented Aug 27, 2022

Uh oh!

leixm commented Aug 27, 2022

Uh oh!

codecov-commenter commented Aug 27, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

jerqi commented Aug 27, 2022

Uh oh!

leixm commented Aug 27, 2022

Uh oh!

leixm commented Aug 27, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

jerqi commented Aug 27, 2022

Uh oh!

leixm commented Aug 27, 2022

Uh oh!

leixm commented Aug 28, 2022

Uh oh!

leixm commented Aug 28, 2022

Uh oh!

jerqi commented Aug 28, 2022

Uh oh!

Uh oh!

zuston commented Aug 28, 2022

Uh oh!

jerqi commented Aug 28, 2022

Uh oh!

leixm commented Aug 28, 2022

Uh oh!

jerqi commented Aug 29, 2022

Uh oh!

leixm commented Aug 30, 2022

Uh oh!

jerqi commented Aug 30, 2022

Uh oh!

Uh oh!

leixm commented Aug 30, 2022

Uh oh!

jerqi commented Aug 30, 2022

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jerqi Sep 8, 2022 • 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!

jerqi commented Aug 30, 2022

Uh oh!

leixm commented Aug 30, 2022

Uh oh!

leixm commented Sep 7, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Environment

codecov-commenter commented Aug 27, 2022 •

edited

Loading

leixm commented Aug 27, 2022 •

edited

Loading

jerqi Sep 8, 2022 •

edited

Loading

leixm commented Sep 7, 2022 •

edited

Loading

jerqi left a comment •

edited

Loading