buska88 opened a new pull request, #3798:
URL: https://github.com/apache/celeborn/pull/3798
…tions without map range
<!--
Thanks for sending a pull request! Here are some tips for you:
- Make sure the PR title start w/ a JIRA ticket, e.g. '[CELEBORN-XXXX]
Your PR title ...'.
- Be sure to keep the PR description updated to reflect all changes.
- Please write your PR title to summarize what this PR proposes.
- If possible, provide a concise example to reproduce the issue for a
faster review.
-->
### 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?
<!-- Check if yes. The `correctness` label will be added/removed
automatically. -->
- [x] Yes
### Does this PR introduce _any_ user-facing change?
<!-- Check if yes. -->
- [ ] 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.
--
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]
