ODIn/spark-instrumented-optimizer
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

[SPARK-36389][CORE][SHUFFLE] Revert the change that accepts negative mapId in ShuffleBlockId

Browse source 
### What changes were proposed in this pull request?
With SPARK-32922, we added a change that ShuffleBlockId can have a negative mapId. This was to support push-based shuffle where -1 as mapId indicated a push-merged block. However with SPARK-32923, a different type of BlockId was introduced - ShuffleMergedId, but reverting the change to ShuffleBlockId was missed.

### Why are the changes needed?
This reverts the changes to `ShuffleBlockId` which will never have a negative mapId.

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

### How was this patch tested?
Modified the unit test to verify the newly added ShuffleMergedBlockId.

Closes #33616 from otterc/SPARK-36389.

Authored-by: Chandni Singh <singh.chandni@gmail.com>
Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>
(cherry picked from commit 2712343a27)
Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>
This commit is contained in:
Chandni Singh 2021-08-02 23:35:32 -07:00 committed by Dongjoon Hyun
parent 71e4c56974
commit 369781a331
2 changed files with 6 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
core/src/main/scala/org/apache/spark/storage/BlockId.scala
View file

@ -191,7 +191,7 @@ class UnrecognizedBlockId(name: String)
@DeveloperApi
object BlockId {
val RDD = "rdd_([0-9]+)_([0-9]+)".r
val SHUFFLE = "shuffle_([0-9]+)_(-?[0-9]+)_([0-9]+)".r
val SHUFFLE = "shuffle_([0-9]+)_([0-9]+)_([0-9]+)".r
val SHUFFLE_BATCH = "shuffle_([0-9]+)_([0-9]+)_([0-9]+)_([0-9]+)".r
val SHUFFLE_DATA = "shuffle_([0-9]+)_([0-9]+)_([0-9]+).data".r
val SHUFFLE_INDEX = "shuffle_([0-9]+)_([0-9]+)_([0-9]+).index".r

10
core/src/test/scala/org/apache/spark/storage/BlockIdSuite.scala
View file

@ -227,13 +227,13 @@ class BlockIdSuite extends SparkFunSuite {
}
test("merged shuffle id") {
val id = ShuffleBlockId(1, -1, 0)
assertSame(id, ShuffleBlockId(1, -1, 0))
assertDifferent(id, ShuffleBlockId(1, 1, 1))
assert(id.name === "shuffle_1_-1_0")
val id = ShuffleMergedBlockId(1, 2, 0)
assertSame(id, ShuffleMergedBlockId(1, 2, 0))
assertDifferent(id, ShuffleMergedBlockId(1, 3, 1))
assert(id.name === "shuffleMerged_1_2_0")
assert(id.asRDDId === None)
assert(id.shuffleId === 1)
assert(id.mapId === -1)
assert(id.shuffleMergeId === 2)
assert(id.reduceId === 0)
assertSame(id, BlockId(id.toString))
}