[CELEBORN-2032][FOLLOWUP] Disable replica preference for skewed partitions without map range

[buska88]

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 PartitionLocations 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 failed validation check ... Mismatch in metadata for the same chunk range on retry

Does this PR resolve a correctness bug?

• 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.

[buska88]

Hello，please take a look at this PR @RexXiong @SteNicholas

[SteNicholas]

@buska88, please bind your email to your github account.

[buska88]

@buska88, please bind your email to your github account.

done

[RexXiong]

LGTM

[SteNicholas]

LGTM.

[SteNicholas]

Merged to main(v1.0.0) and branch-0.7(v0.7.1).