gaoyajun02 commented on code in PR #58008:
URL: https://github.com/apache/spark/pull/58008#discussion_r3851637481
##########
core/src/main/scala/org/apache/spark/storage/PushBasedFetchHelper.scala:
##########
@@ -301,33 +289,19 @@ private class PushBasedFetchHelper(
}
/**
- * Check whether a push-merged block contains data from stale (duplicate)
task attempts.
- * When speculation is enabled, multiple attempts for the same map output
may both push data
- * to the merger. The merger may include data from both attempts in the same
merged block,
- * but the driver only tracks one as the canonical MapStatus. We detect this
by checking
- * if any stale pushed map index appears in the server-side chunkBitmaps.
+ * Check whether a shuffle chunk contains any stale pushed map index,
enabling
+ * chunk-granularity fallback: only chunks that actually contain stale data
fall back to their
+ * original blocks, while the remaining chunks of the same merged block are
read normally.
*
- * @param shuffleBlockId ShuffleMergedBlockId to be checked
- * @param address BlockManagerId of push-based shuffle service
- * @param chunkBitmaps Chunks bitmap from push-based shuffle service side
- * @return false if any stale-marked mapIndex is present in this block
(forcing fallback),
- * true otherwise
+ * @param blockId ShuffleBlockChunkId to be checked
+ * @return true if this chunk contains any stale-marked mapIndex (forcing a
fallback for this
+ * chunk only), false otherwise
*/
- private[this] def checkStaleMapIdInMergedBlock(
- shuffleBlockId: ShuffleMergedBlockId,
- address: BlockManagerId,
- chunkBitmaps: Array[RoaringBitmap]): Boolean = {
- val staleMapIndexes =
- mapOutputTracker.getStaleMapIndexes(shuffleBlockId.shuffleId)
- if (staleMapIndexes.isEmpty) return true
- val mergedBlockBitmap = new RoaringBitmap()
- chunkBitmaps.foreach(mergedBlockBitmap.or)
- val hasStale = staleMapIndexes.exists(id => mergedBlockBitmap.contains(id))
- if (hasStale) {
- logWarning(s"Found stale pushed map indexes in merged block
$shuffleBlockId from" +
- s" ${address.host}:${address.port}, falling back to fetch the original
blocks")
+ private[spark] def isStaleChunk(blockId: ShuffleBlockChunkId): Boolean = {
+ val staleMapIndexes =
mapOutputTracker.getStaleMapIndexes(blockId.shuffleId)
Review Comment:
Done. Added `intersectsStaleMapIndexes(shuffleId, bitmap)` to
`MapOutputTracker` (trait + master + worker) and `ShuffleStatus`, which
inspects the published stale snapshot via iterator + `bitmap.contains` without
returning a defensive copy. `isStaleChunk` now delegates to it, dropping the
O(C * S) copy on the reducer fetch path.
`getStaleMapIndexes` is kept for the driver-side propagation serialization
path, where a copy is still appropriate.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
