[CELEBORN-2032][FOLLOWUP] Disable replica preference for skewed partitions without map range - #3798
Closed
buska88 wants to merge 1 commit into
Closed
[CELEBORN-2032][FOLLOWUP] Disable replica preference for skewed partitions without map range#3798buska88 wants to merge 1 commit into
buska88 wants to merge 1 commit into
Conversation
…tions without map range
Author
|
Hello,please take a look at this PR @RexXiong @SteNicholas |
Member
|
@buska88, please bind your email to your github account. |
Author
done |
SteNicholas
pushed a commit
that referenced
this pull request
Aug 12, 2026
…tions without map range ### What changes were proposed in this pull request? CELEBORN-2032 introduced attempt-based primary/replica switching (`preferReplicaRead = context.attemptNumber % 2 == 1`) in `CelebornShuffleReader` so that odd-numbered task attempts prefer reading the replica `PartitionLocation` instead of the primary, improving fault tolerance across retries/speculative execution. This PR disables that replica preference specifically when `celeborn.client.adaptive.optimizeSkewedPartitionRead.enabled` is on and the partition is being read as a skewed partition without map range (`splitSkewPartitionWithoutMapRange`). In that mode, all attempts for the same skewed partition will now consistently read the primary (or previously-resolved) locations instead of alternating between primary and replica. A unit test (`CelebornPartitionUtilSuiteJ#testSkewPartitionSplitDiffersBetweenPrimaryAndReplicaChunkOffsets`) is added to demonstrate the root cause directly against `CelebornPartitionUtil#splitSkewedPartitionLocations`. ### Why are the changes needed? When `celeborn.client.adaptive.optimizeSkewedPartitionRead.enabled=true`, a skewed reduce partition is not read by map-id range. Instead, Celeborn treats all `PartitionLocation`s of that partition as one logical byte stream and splits it into `subPartitionSize` sub-partitions purely by byte offset (`CelebornPartitionUtil#splitSkewedPartitionLocations`). For a given `subPartitionIndex`, this method computes a `chunkRange` (physical chunk index interval) by walking the `chunkOffsets` of whichever `PartitionLocation` objects are passed in. The primary and its replica are flushed independently by two different Workers. Even though they hold logically identical data and share the same `uniqueId`, their physical `chunkOffsets` (the byte positions at which each flush produced a new chunk) are **not guaranteed to be identical**. Because of CELEBORN-2032, whether a task attempt reads the primary or the replica depends on `attemptNumber % 2`. So: - Attempt 0 (first run) reads the primary and resolves `chunkRange` from the primary's chunk offsets. - Attempt 1 (retry / speculative execution) reads the replica and resolves `chunkRange` from the replica's (possibly different) chunk offsets. For the exact same logical `subPartitionIndex`, this can produce two **different physical byte ranges**, e.g. primary resolves to chunk range `[2, 3]` while replica resolves to `[3, 3]` (dropping chunk 2 entirely). The two attempts then read different bytes for what should be the identical logical sub-partition, so their computed byte-count/CRC diverge and fail `SkewHandlingWithoutMapRangeValidator`, surfacing as:org.apache.celeborn.common.exception.CelebornIOException: AQE Partition <n> failed validation check ... Mismatch in metadata for the same chunk range on retry <img width="1711" height="605" alt="image" src="https://github.com/user-attachments/assets/b4bff79e-7287-4bb5-8ba1-f2bf7cf21f20" /> ### Does this PR resolve a correctness bug? - [x] Yes ### Does this PR introduce _any_ user-facing change? - [ ] Yes ### How was this patch tested? - Added `CelebornPartitionUtilSuiteJ#testSkewPartitionSplitDiffersBetweenPrimaryAndReplicaChunkOffsets`, which constructs a primary and a replica `PartitionLocation` sharing the same `uniqueId` but with different `chunkOffsets` (simulating independent flush behavior), and asserts that `CelebornPartitionUtil#splitSkewedPartitionLocations` resolves different chunk ranges (`[2, 3]` vs `[3, 3]`) for the identical `subPartitionIndex`, proving the root cause of the non-idempotent read. - Ran the full `CelebornPartitionUtilSuiteJ` (5 tests) and `CelebornShuffleReaderSuite` (10 tests) in `client-spark/spark-3` — all passed. Closes #3798 from buska88/celeborn-2032-fix. Authored-by: lijianfu03 <lijianfu03@meituan.com> Signed-off-by: 子懿 <programgeek@163.com> (cherry picked from commit 87aae3f) Signed-off-by: 子懿 <programgeek@163.com>
Member
|
Merged to main(v1.0.0) and branch-0.7(v0.7.1). |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
CELEBORN-2032 introduced attempt-based primary/replica switching (
preferReplicaRead = context.attemptNumber % 2 == 1) inCelebornShuffleReaderso that odd-numbered task attempts prefer reading the replicaPartitionLocationinstead of the primary, improving fault tolerance across retries/speculative execution.This PR disables that replica preference specifically when
celeborn.client.adaptive.optimizeSkewedPartitionRead.enabledis on and the partition is being read as a skewed partition without map range (splitSkewPartitionWithoutMapRange). In that mode, all attempts for the same skewed partition will now consistently read the primary (or previously-resolved) locations instead of alternating between primary and replica.A unit test (
CelebornPartitionUtilSuiteJ#testSkewPartitionSplitDiffersBetweenPrimaryAndReplicaChunkOffsets) is added to demonstrate the root cause directly againstCelebornPartitionUtil#splitSkewedPartitionLocations.Why are the changes needed?
When
celeborn.client.adaptive.optimizeSkewedPartitionRead.enabled=true, a skewed reduce partition is not read by map-id range. Instead, Celeborn treats allPartitionLocations of that partition as one logical byte stream and splits it intosubPartitionSizesub-partitions purely by byte offset (CelebornPartitionUtil#splitSkewedPartitionLocations). For a givensubPartitionIndex, this method computes achunkRange(physical chunk index interval) by walking thechunkOffsetsof whicheverPartitionLocationobjects are passed in.The primary and its replica are flushed independently by two different Workers. Even though they hold logically identical data and share the same
uniqueId, their physicalchunkOffsets(the byte positions at which each flush produced a new chunk) are not guaranteed to be identical.Because of CELEBORN-2032, whether a task attempt reads the primary or the replica depends on
attemptNumber % 2. So:chunkRangefrom the primary's chunk offsets.chunkRangefrom the replica's (possibly different) chunk offsets.For the exact same logical
subPartitionIndex, this can produce two different physical byte ranges, e.g. primary resolves to chunk range[2, 3]while replica resolves to[3, 3](dropping chunk 2 entirely). The two attempts then read different bytes for what should be the identical logical sub-partition, so their computed byte-count/CRC diverge and failSkewHandlingWithoutMapRangeValidator, surfacing as:org.apache.celeborn.common.exception.CelebornIOException: AQE Partition failed validation check ... Mismatch in metadata for the same chunk range on retryDoes this PR resolve a correctness bug?
Does this PR introduce any user-facing change?
How was this patch tested?
CelebornPartitionUtilSuiteJ#testSkewPartitionSplitDiffersBetweenPrimaryAndReplicaChunkOffsets, which constructs a primary and a replicaPartitionLocationsharing the sameuniqueIdbut with differentchunkOffsets(simulating independent flush behavior), and asserts thatCelebornPartitionUtil#splitSkewedPartitionLocationsresolves different chunk ranges ([2, 3]vs[3, 3]) for the identicalsubPartitionIndex, proving the root cause of the non-idempotent read.CelebornPartitionUtilSuiteJ(5 tests) andCelebornShuffleReaderSuite(10 tests) inclient-spark/spark-3— all passed.