diff --git a/.github/workflows/pr_build_linux.yml b/.github/workflows/pr_build_linux.yml
index a9c184a20b..9c6c4881c3 100644
--- a/.github/workflows/pr_build_linux.yml
+++ b/.github/workflows/pr_build_linux.yml
@@ -507,6 +507,7 @@ jobs:
               org.apache.spark.sql.comet.execution.shuffle.CometCelebornShuffleReaderSuite
               org.apache.spark.sql.comet.execution.shuffle.CometCelebornShufflePlanningSuite
               org.apache.spark.sql.comet.execution.shuffle.CometNativeShuffleInputRDDSuite
+              org.apache.spark.sql.comet.execution.shuffle.CometNativeRoundRobinRetrySuite
               org.apache.spark.sql.comet.execution.shuffle.CometDiskBlockWriterSuite
               org.apache.comet.exec.CometShuffleEncryptionSuite
               org.apache.comet.exec.CometShuffleManagerSuite
diff --git a/.github/workflows/pr_build_macos.yml b/.github/workflows/pr_build_macos.yml
index b47ed5a46f..b5d7f8816e 100644
--- a/.github/workflows/pr_build_macos.yml
+++ b/.github/workflows/pr_build_macos.yml
@@ -155,6 +155,7 @@ jobs:
               org.apache.spark.sql.comet.execution.shuffle.CometCelebornShuffleReaderSuite
               org.apache.spark.sql.comet.execution.shuffle.CometCelebornShufflePlanningSuite
               org.apache.spark.sql.comet.execution.shuffle.CometNativeShuffleInputRDDSuite
+              org.apache.spark.sql.comet.execution.shuffle.CometNativeRoundRobinRetrySuite
               org.apache.spark.sql.comet.execution.shuffle.CometDiskBlockWriterSuite
               org.apache.comet.exec.CometShuffleEncryptionSuite
               org.apache.comet.exec.CometShuffleManagerSuite
diff --git a/docs/source/contributor-guide/native_shuffle.md b/docs/source/contributor-guide/native_shuffle.md
index 8d84aa2f6b..e040c87a77 100644
--- a/docs/source/contributor-guide/native_shuffle.md
+++ b/docs/source/contributor-guide/native_shuffle.md
@@ -49,7 +49,6 @@ Native shuffle (`CometExchange`) is selected when all of the following condition
    columnar output. Row-based Spark operators require JVM shuffle.
 
 3. **Supported partitioning type**: Native shuffle supports:
-
    - `HashPartitioning`
    - `RangePartitioning`
    - `SinglePartition`
@@ -152,7 +151,6 @@ The native shuffle implementation is its own workspace crate, `datafusion-comet-
 1. **Plan construction**: `CometNativeShuffleWriter` builds a protobuf operator tree with a
    `ShuffleWriter` operator at the root and `childNativeOp` as its child. `childNativeOp` takes
    one of two shapes:
-
    - The child plan's `nativeOp` directly, when `CometShuffleExchangeExec`'s child is a
      `CometNativeExec` subtree. The upstream operators run inside the same `CometExecIterator`
      as the writer, with no JVM-to-native batch boundary between them.
@@ -165,7 +163,6 @@ The native shuffle implementation is its own workspace crate, `datafusion-comet-
 2. **Native execution**: A single `CometExecIterator` per partition runs the unified plan.
 
 3. **Partitioning**: `ShuffleWriterExec` receives batches and routes to the appropriate partitioner:
-
    - `MultiPartitionShuffleRepartitioner`: For hash/range/round-robin partitioning
    - `SinglePartitionShufflePartitioner`: For single partition (simpler path)
 
@@ -173,7 +170,6 @@ The native shuffle implementation is its own workspace crate, `datafusion-comet-
    exceeds the threshold, partitions spill to temporary files.
 
 5. **Encoding**: `ShuffleBlockWriter` encodes each partition's data as compressed Arrow IPC:
-
    - Writes compression type header
    - Writes field count header
    - Writes compressed IPC stream
@@ -203,7 +199,6 @@ read time. See [Direct Read](#direct-read-shufflescan) below for how the choice
 1. `CometBlockStoreShuffleReader` fetches shuffle blocks via `ShuffleBlockFetcherIterator`.
 
 2. For each block, `NativeBatchDecoderIterator`:
-
    - Reads the 8-byte compressed length header
    - Reads the 8-byte field count header
    - Reads the compressed IPC data
@@ -305,7 +300,11 @@ batch is written as a single block that may exceed the batch size.
 
 ### Round Robin Partitioning
 
-Comet implements round robin partitioning using hash-based assignment for determinism:
+`CometPartitioning::RoundRobin` carries a `RoundRobinStrategy` that decides how rows reach output
+partitions. The default is `HashAll`; `WholeBatch` is opt-in through
+`spark.comet.shuffle.native.partitioning.roundrobin.batchGranular`.
+
+#### `HashAll`: hash-based assignment (default)
 
 1. Computes a Murmur3 hash of columns (using seed 42)
 2. Assigns partitions directly using the hash: `partition_id = hash % num_partitions`
@@ -321,6 +320,71 @@ which Arrow's layout does not reproduce, unsorted output can land in different p
 Spark's. Sorted output is identical. That difference is why
 `spark.comet.shuffle.native.partitioning.roundrobin.enabled` defaults to `false`.
 
+`spark.comet.shuffle.native.partitioning.roundrobin.maxHashColumns` caps how many leading columns
+are hashed. `0`, the default, hashes all of them.
+
+#### `WholeBatch`: positional assignment
+
+Hashing every column of every row dominates the shuffle write on wide nested schemas, because
+`create_murmur3_hashes` recurses into every struct child per row and the resulting row-level
+scatter forces `interleave_record_batch` to walk every column and child again on flush.
+`WholeBatch` assigns each incoming `RecordBatch` whole to `counter % num_partitions`.
+`partition_row_indices` is left unmaterialized and the flush clones the source batch instead of
+interleaving it, which removes both the per-row hash and the per-row `interleave_record_batch`
+gather. Those two are what dominate the shuffle write on a wide nested schema.
+
+Two caveats on the cost, because the path is not free per batch. It still writes and rescans
+`partition_starts` across all `num_partitions` slots, and it still appends one `(batch, row)` pair
+per row to the target partition's index list, which is charged against the spill reservation and
+scanned again by `PartitionedBatchIterator` to recognise the whole-batch case. The clone fast path
+also only fires when a flush chunk lines up exactly with one buffered batch, so a partition holding
+several batches shorter than `batch_size` falls back to interleaving.
+
+The counter starts at the Spark map partition id, carried on the strategy as
+`RoundRobinStrategy::WholeBatch { start_partition }` and filled in by
+`PhysicalPlanner::create_partitioning` from the planner's partition. It cannot come from
+`ShuffleWriterExec::execute`, because `jni_api` runs every native root plan with partition 0 (one
+Comet execution per Spark task). The map partition id is the right seed on both counts: it is
+distinct across mappers, so a task emitting fewer batches than there are output partitions does not
+leave the tail empty stage-wide, and it is a pure function of the partition, so a re-executed task
+reproduces its placement. Spark's own round robin seeds `XORShiftRandom(partitionId)` for the same
+two reasons.
+
+Distribution is even at batch granularity rather than row granularity: fewer batches than
+partitions leaves partitions empty, and unequal batch sizes give unequal partitions.
+
+#### Retry safety under `WholeBatch`
+
+Positional assignment is not a function of the rows, so it is only reproducible when the upstream
+operator replays the same batches in the same order and with the same framing. Re-executing one map
+task against differently framed input writes a different partitioning of the same rows, and once
+any consumer has fetched the output that attempt replaces, the reduce side silently gets some rows
+twice and others not at all. Spark faces the same problem with its own round robin and answers it
+in two ways, both of which Comet's native path mirrors:
+
+- **Declaring the risk.** Spark wraps a round-robin repartition in a `MapPartitionsRDD` with
+  `isOrderSensitive = true`, which reports `INDETERMINATE` whenever its parent is `UNORDERED`. The
+  `DAGScheduler` then rolls the whole stage back rather than re-running a single task, and aborts
+  the job outright when a result stage has already consumed output. The native path has no
+  `MapPartitionsRDD` to carry the flag, so `CometNativeShuffleInputRDD.getOutputDeterministicLevel`
+  applies the same rule directly. A determinate parent such as a plain scan stays determinate and
+  keeps cheap per-task retry; anything below another exchange is unordered, because reduce tasks
+  see shuffle blocks in arrival order, and goes indeterminate.
+
+- **Refusing the retry.** Rollback is only as sound as the parent's determinism level being an
+  accurate description of what the upstream operator replays. While the strategy is opt-in and
+  unproven, `spark.comet.shuffle.native.partitioning.roundrobin.batchGranular.failOnRetry`
+  (default `true`) makes `CometNativeShuffleWriter` refuse to run at all when `TaskContext` reports
+  a task attempt after the first or a re-submitted stage attempt. Both counters are needed:
+  `TaskSetManager.executorLost` re-enqueues a dead executor's map tasks inside the current task set
+  with a fresh task attempt, while a fetch failure against that executor's output resubmits the
+  stage instead. Spark has no API for failing an application from inside a task, so this throws an
+  ordinary exception; because the condition only gets more true on each attempt, the task set
+  aborts after `spark.task.maxFailures` and takes the job with it. Setting the config to `false`
+  leaves the indeterminate declaration above as the only defence.
+
+Neither applies to `HashAll`, whose output is a pure function of the rows it sees.
+
 ## Memory Management
 
 Native shuffle uses DataFusion's memory management with spilling support:
@@ -370,21 +434,23 @@ independently compressed, allowing parallel decompression during reads.
 
 ## Configuration
 
-| Config                                                              | Default | Description                                                   |
-| ------------------------------------------------------------------- | ------- | ------------------------------------------------------------- |
-| `spark.comet.shuffle.enabled`                                       | `true`  | Enable Comet shuffle                                          |
-| `spark.comet.shuffle.mode`                                          | `auto`  | Shuffle mode: `native`, `jvm`, or `auto`                      |
-| `spark.comet.shuffle.directRead.enabled`                            | `true`  | Decode shuffle blocks in native code, bypassing Arrow FFI     |
-| `spark.comet.shuffle.compression.codec`                             | `lz4`   | Compression codec                                             |
-| `spark.comet.shuffle.compression.zstd.level`                        | `1`     | Zstd compression level                                        |
-| `spark.comet.shuffle.native.writeBufferSize`                        | `1MB`   | Write buffer size                                             |
-| `spark.comet.shuffle.native.maxBufferBytes`                         | `0`     | Fixed spill threshold. `0` disables it, leaving pool pressure |
-| `spark.comet.shuffle.native.partitioning.hash.enabled`              | `true`  | Allow `HashPartitioning` on the native path                   |
-| `spark.comet.shuffle.native.partitioning.hash.nested.enabled`       | `false` | Allow struct and array hash keys, and map keys on Spark 4.0+  |
-| `spark.comet.shuffle.native.partitioning.range.enabled`             | `true`  | Allow `RangePartitioning` on the native path                  |
-| `spark.comet.shuffle.native.partitioning.roundrobin.enabled`        | `false` | Allow `RoundRobinPartitioning` on the native path             |
-| `spark.comet.shuffle.native.partitioning.roundrobin.maxHashColumns` | `0`     | Columns to hash for round robin. `0` hashes all of them       |
-| `spark.comet.shuffle.jvm.batchSize`                                 | `8192`  | Target rows per batch                                         |
+| Config                                                                         | Default | Description                                                                |
+| ------------------------------------------------------------------------------ | ------- | -------------------------------------------------------------------------- |
+| `spark.comet.shuffle.enabled`                                                  | `true`  | Enable Comet shuffle                                                       |
+| `spark.comet.shuffle.mode`                                                     | `auto`  | Shuffle mode: `native`, `jvm`, or `auto`                                   |
+| `spark.comet.shuffle.directRead.enabled`                                       | `true`  | Decode shuffle blocks in native code, bypassing Arrow FFI                  |
+| `spark.comet.shuffle.compression.codec`                                        | `lz4`   | Compression codec                                                          |
+| `spark.comet.shuffle.compression.zstd.level`                                   | `1`     | Zstd compression level                                                     |
+| `spark.comet.shuffle.native.writeBufferSize`                                   | `1MB`   | Write buffer size                                                          |
+| `spark.comet.shuffle.native.maxBufferBytes`                                    | `0`     | Fixed spill threshold. `0` disables it, leaving pool pressure              |
+| `spark.comet.shuffle.native.partitioning.hash.enabled`                         | `true`  | Allow `HashPartitioning` on the native path                                |
+| `spark.comet.shuffle.native.partitioning.hash.nested.enabled`                  | `false` | Allow struct and array hash keys, and map keys on Spark 4.0+               |
+| `spark.comet.shuffle.native.partitioning.range.enabled`                        | `true`  | Allow `RangePartitioning` on the native path                               |
+| `spark.comet.shuffle.native.partitioning.roundrobin.enabled`                   | `false` | Allow `RoundRobinPartitioning` on the native path                          |
+| `spark.comet.shuffle.native.partitioning.roundrobin.maxHashColumns`            | `0`     | Columns to hash for round robin. `0` hashes all of them                    |
+| `spark.comet.shuffle.native.partitioning.roundrobin.batchGranular`             | `false` | Assign each batch whole to one partition instead of hashing rows           |
+| `spark.comet.shuffle.native.partitioning.roundrobin.batchGranular.failOnRetry` | `true`  | Fail a retried positional round-robin map task instead of rewriting output |
+| `spark.comet.shuffle.jvm.batchSize`                                            | `8192`  | Target rows per batch                                                      |
 
 ## Comparison with JVM Shuffle
 
diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs
index 8f030da455..e569bcf061 100644
--- a/native/core/src/execution/planner.rs
+++ b/native/core/src/execution/planner.rs
@@ -98,7 +98,7 @@ use datafusion_comet_spark_expr::{
 use iceberg::expr::Bind;
 
 use crate::execution::operators::ExecutionError::GeneralError;
-use crate::execution::shuffle::{CometPartitioning, CompressionCodec};
+use crate::execution::shuffle::{CometPartitioning, CompressionCodec, RoundRobinStrategy};
 use crate::execution::spark_plan::SparkPlan;
 use crate::parquet::objectstore::s3_blob_fs_support::normalize_object_store_url;
 use crate::parquet::parquet_support::prepare_object_store_with_configs;
@@ -3652,15 +3652,22 @@ impl PhysicalPlanner {
             }
             PartitioningStruct::SinglePartition(_) => Ok(CometPartitioning::SinglePartition),
             PartitioningStruct::RoundRobinPartition(rr_partition) => {
-                // Treat negative max_hash_columns as 0 (no limit)
-                let max_hash_columns = if rr_partition.max_hash_columns <= 0 {
-                    0
+                let strategy = if rr_partition.batch_granular {
+                    // The Spark map partition id, not the DataFusion one: the root plan is
+                    // always executed with partition 0 (one Comet execution per Spark task),
+                    // so `ShuffleWriterExec::execute` cannot supply it. See
+                    // `RoundRobinStrategy::WholeBatch` for why it has to be this value.
+                    RoundRobinStrategy::WholeBatch {
+                        start_partition: self.partition.max(0) as usize,
+                    }
                 } else {
-                    rr_partition.max_hash_columns as usize
+                    // Treat negative max_hash_columns as 0 (no limit).
+                    let max_hash_columns = rr_partition.max_hash_columns.max(0) as usize;
+                    RoundRobinStrategy::HashAll { max_hash_columns }
                 };
                 Ok(CometPartitioning::RoundRobin(
                     rr_partition.num_partitions as usize,
-                    max_hash_columns,
+                    strategy,
                 ))
             }
         }
diff --git a/native/proto/src/proto/partitioning.proto b/native/proto/src/proto/partitioning.proto
index e70b8264f0..9aa6acc322 100644
--- a/native/proto/src/proto/partitioning.proto
+++ b/native/proto/src/proto/partitioning.proto
@@ -57,4 +57,8 @@ message RoundRobinPartition {
   int32 num_partitions = 1;
   // Maximum number of columns to hash. 0 means no limit (hash all columns).
   int32 max_hash_columns = 2;
+  // When true, assign each incoming RecordBatch as a whole to one output partition,
+  // skipping per-row hashing entirely. Retry-safe when the upstream operator
+  // emits deterministic batches under retry.
+  bool batch_granular = 3;
 }
diff --git a/native/shuffle/README.md b/native/shuffle/README.md
index 0f53604fa3..2a47f643ff 100644
--- a/native/shuffle/README.md
+++ b/native/shuffle/README.md
@@ -41,21 +41,21 @@ cargo run --release --features shuffle-bench --bin shuffle_bench -- \
 
 ### Options
 
-| Option                | Default                    | Description                                            |
-| --------------------- | -------------------------- | ------------------------------------------------------ |
-| `--input`             | _(required)_               | Path to a Parquet file or directory of Parquet files   |
-| `--partitions`        | `200`                      | Number of output shuffle partitions                    |
-| `--partitioning`      | `hash`                     | Partitioning scheme: `hash`, `single`, `round-robin`   |
-| `--hash-columns`      | `0`                        | Comma-separated column indices to hash on (e.g. `0,3`) |
-| `--codec`             | `lz4`                      | Compression codec: `none`, `lz4`, `zstd`, `snappy`     |
-| `--zstd-level`        | `1`                        | Zstd compression level (1–22)                          |
-| `--batch-size`        | `8192`                     | Batch size for reading Parquet data                    |
-| `--memory-limit`      | _(none)_                   | Memory limit in bytes; triggers spilling when exceeded |
-| `--write-buffer-size` | `1048576`                  | Write buffer size in bytes                             |
-| `--limit`             | `0`                        | Limit rows processed per iteration (0 = no limit)      |
-| `--iterations`        | `1`                        | Number of timed iterations                             |
-| `--warmup`            | `0`                        | Number of warmup iterations before timing              |
-| `--output-dir`        | `/tmp/comet_shuffle_bench` | Directory for temporary shuffle output files           |
+| Option                | Default                    | Description                                                                     |
+| --------------------- | -------------------------- | ------------------------------------------------------------------------------- |
+| `--input`             | _(required)_               | Path to a Parquet file or directory of Parquet files                            |
+| `--partitions`        | `200`                      | Number of output shuffle partitions                                             |
+| `--partitioning`      | `hash`                     | Partitioning scheme: `hash`, `single`, `round-robin`, `round-robin-whole-batch` |
+| `--hash-columns`      | `0`                        | Comma-separated column indices to hash on (e.g. `0,3`)                          |
+| `--codec`             | `lz4`                      | Compression codec: `none`, `lz4`, `zstd`, `snappy`                              |
+| `--zstd-level`        | `1`                        | Zstd compression level (1–22)                                                   |
+| `--batch-size`        | `8192`                     | Batch size for reading Parquet data                                             |
+| `--memory-limit`      | _(none)_                   | Memory limit in bytes; triggers spilling when exceeded                          |
+| `--write-buffer-size` | `1048576`                  | Write buffer size in bytes                                                      |
+| `--limit`             | `0`                        | Limit rows processed per iteration (0 = no limit)                               |
+| `--iterations`        | `1`                        | Number of timed iterations                                                      |
+| `--warmup`            | `0`                        | Number of warmup iterations before timing                                       |
+| `--output-dir`        | `/tmp/comet_shuffle_bench` | Directory for temporary shuffle output files                                    |
 
 ### Profiling with flamegraph
 
diff --git a/native/shuffle/benches/shuffle_writer.rs b/native/shuffle/benches/shuffle_writer.rs
index cc94375436..5fd0ebb945 100644
--- a/native/shuffle/benches/shuffle_writer.rs
+++ b/native/shuffle/benches/shuffle_writer.rs
@@ -19,18 +19,21 @@ use arrow::array::builder::{Date32Builder, Decimal128Builder, Int32Builder};
 use arrow::array::{builder::StringBuilder, Array, Int32Array, RecordBatch};
 use arrow::datatypes::{DataType, Field, Schema};
 use arrow::row::{RowConverter, SortField};
-use criterion::{criterion_group, criterion_main, Criterion};
+use criterion::{criterion_group, criterion_main, BatchSize, Criterion};
 use datafusion::datasource::memory::MemorySourceConfig;
 use datafusion::datasource::source::DataSourceExec;
+use datafusion::execution::runtime_env::RuntimeEnvBuilder;
 use datafusion::physical_expr::expressions::{col, Column};
 use datafusion::physical_expr::{LexOrdering, PhysicalSortExpr};
-use datafusion::physical_plan::metrics::Time;
+use datafusion::physical_plan::metrics::{ExecutionPlanMetricsSet, Time};
 use datafusion::{
     physical_plan::{common::collect, ExecutionPlan},
     prelude::SessionContext,
 };
 use datafusion_comet_shuffle::{
-    CometPartitioning, CompressionCodec, ShuffleBlockWriter, ShuffleCodecContext, ShuffleWriterExec,
+    CometPartitioning, CompressionCodec, LocalPartitionWriter, MultiPartitionShuffleRepartitioner,
+    PartitionOffsets, RoundRobinStrategy, ShuffleBlockWriter, ShuffleCodecContext,
+    ShufflePartitioner, ShufflePartitionerMetrics, ShuffleWriterExec,
 };
 use itertools::Itertools;
 use std::io::Cursor;
@@ -77,13 +80,12 @@ fn criterion_benchmark(c: &mut Criterion) {
                 let exec = create_shuffle_writer_exec(
                     compression_codec.clone(),
                     CometPartitioning::Hash(vec![Arc::new(Column::new("a", 0))], 16),
-                    8192,
-                    10,
+                    create_batches(8192, 10),
                 );
+                let rt = Runtime::new().unwrap();
                 b.iter(|| {
                     let task_ctx = ctx.task_ctx();
                     let stream = exec.execute(0, task_ctx).unwrap();
-                    let rt = Runtime::new().unwrap();
                     rt.block_on(collect(stream)).unwrap();
                 });
             },
@@ -130,13 +132,12 @@ fn criterion_benchmark(c: &mut Criterion) {
                 let exec = create_shuffle_writer_exec(
                     compression_codec.clone(),
                     partitioning.clone(),
-                    8192,
-                    10,
+                    create_batches(8192, 10),
                 );
+                let rt = Runtime::new().unwrap();
                 b.iter(|| {
                     let task_ctx = ctx.task_ctx();
                     let stream = exec.execute(0, task_ctx).unwrap();
-                    let rt = Runtime::new().unwrap();
                     rt.block_on(collect(stream)).unwrap();
                 });
             },
@@ -161,18 +162,117 @@ fn criterion_benchmark(c: &mut Criterion) {
                 let exec = create_shuffle_writer_exec(
                     CompressionCodec::None,
                     CometPartitioning::SinglePartition,
-                    rows_per_batch,
-                    num_batches,
+                    create_batches(rows_per_batch, num_batches),
                 );
+                let rt = Runtime::new().unwrap();
                 b.iter(|| {
                     let task_ctx = ctx.task_ctx();
                     let stream = exec.execute(0, task_ctx).unwrap();
-                    let rt = Runtime::new().unwrap();
                     rt.block_on(collect(stream)).unwrap();
                 });
             },
         );
     }
+
+    // RoundRobin on a wide nested schema: compares the row-level hash-all-columns strategy
+    // against the whole-batch strategy. The nested schema is where the win is largest:
+    // `create_murmur3_hashes` recurses into every Struct child per row on the row-level path,
+    // while the whole-batch path does one mod per batch and never inspects row contents.
+    // 40 top-level struct columns of shallow depth reflect the real-world shape where this
+    // optimization matters (event log workload was ~194 nested columns).
+    let num_partitions = 50usize;
+    let wide_batches: Vec<RecordBatch> = (0..8).map(|_| nested_schema_batch(8192, 40, 2)).collect();
+    let wide_schema = wide_batches[0].schema();
+    let round_robin_strategies = [
+        ("hash_all_columns", RoundRobinStrategy::default()),
+        (
+            "whole_batch",
+            RoundRobinStrategy::WholeBatch { start_partition: 0 },
+        ),
+    ];
+    for (label, strategy) in &round_robin_strategies {
+        group.bench_function(
+            format!("shuffle_writer: RoundRobin nested schema (strategy={label})"),
+            |b| {
+                let ctx = SessionContext::new();
+                let rt = Runtime::new().unwrap();
+                let exec = create_shuffle_writer_exec(
+                    CompressionCodec::None,
+                    CometPartitioning::RoundRobin(num_partitions, strategy.clone()),
+                    wide_batches.clone(),
+                );
+                b.iter(|| {
+                    let task_ctx = ctx.task_ctx();
+                    let stream = exec.execute(0, task_ctx).unwrap();
+                    rt.block_on(collect(stream)).unwrap();
+                });
+            },
+        );
+    }
+
+    // Partitioning-only microbench. `insert_batch` runs partition assignment + index
+    // buffering but never flushes to disk, so the two RoundRobin strategies can be compared
+    // without the IPC encode + disk write cost that dominates the end-to-end bench above.
+    //
+    // The repartitioner is built in the setup closure, not the timed one: `try_new` sizes the
+    // row scratch from the strategy, so timing it would charge `hash_all_columns` ~64 KB of
+    // per-iteration allocation that `whole_batch` does not pay and bias the comparison.
+    for (label, strategy) in &round_robin_strategies {
+        group.bench_function(
+            format!("partitioning_only: RoundRobin nested schema (strategy={label})"),
+            |b| {
+                let rt = Runtime::new().unwrap();
+                let batches = wide_batches.clone();
+                let runtime_env = Arc::new(
+                    RuntimeEnvBuilder::new()
+                        .with_memory_limit(1024 * 1024 * 1024, 1.0)
+                        .build()
+                        .unwrap(),
+                );
+                let dir = tempfile::tempdir().unwrap();
+                let data_path = dir.path().join("data.out").to_str().unwrap().to_string();
+                b.iter_batched(
+                    || {
+                        let block_writer = ShuffleBlockWriter::try_new(
+                            wide_schema.as_ref(),
+                            CompressionCodec::None,
+                        )
+                        .unwrap();
+                        let writer = LocalPartitionWriter::try_new(
+                            data_path.clone(),
+                            Arc::new(PartitionOffsets::default()),
+                            block_writer,
+                            num_partitions,
+                            8192,
+                            1024 * 1024,
+                            Arc::clone(&runtime_env),
+                        )
+                        .unwrap();
+                        MultiPartitionShuffleRepartitioner::try_new(
+                            0,
+                            writer,
+                            CometPartitioning::RoundRobin(num_partitions, strategy.clone()),
+                            ShufflePartitionerMetrics::new(&ExecutionPlanMetricsSet::new(), 0),
+                            Arc::clone(&runtime_env),
+                            8192,
+                            false,
+                            None,
+                        )
+                        .unwrap()
+                    },
+                    |mut repartitioner| {
+                        rt.block_on(async {
+                            for batch in &batches {
+                                repartitioner.insert_batch(batch.clone()).await.unwrap();
+                            }
+                        });
+                    },
+                    BatchSize::LargeInput,
+                );
+            },
+        );
+    }
+
     group.finish();
 
     // High partition counts stress the per-partition write path (one short-lived
@@ -189,13 +289,12 @@ fn criterion_benchmark(c: &mut Criterion) {
                 let exec = create_shuffle_writer_exec(
                     CompressionCodec::None,
                     CometPartitioning::Hash(vec![Arc::new(Column::new("a", 0))], num_partitions),
-                    8192,
-                    10,
+                    create_batches(8192, 10),
                 );
+                let rt = Runtime::new().unwrap();
                 b.iter(|| {
                     let task_ctx = ctx.task_ctx();
                     let stream = exec.execute(0, task_ctx).unwrap();
-                    let rt = Runtime::new().unwrap();
                     rt.block_on(collect(stream)).unwrap();
                 });
             },
@@ -207,10 +306,8 @@ fn criterion_benchmark(c: &mut Criterion) {
 fn create_shuffle_writer_exec(
     compression_codec: CompressionCodec,
     partitioning: CometPartitioning,
-    rows_per_batch: usize,
-    num_batches: usize,
+    batches: Vec<RecordBatch>,
 ) -> ShuffleWriterExec {
-    let batches = create_batches(rows_per_batch, num_batches);
     let schema = batches[0].schema();
     let partitions = &[batches];
     ShuffleWriterExec::try_new(
@@ -278,7 +375,7 @@ fn schema_encoding_benchmark(c: &mut Criterion) {
 
     for (name, batch) in [
         ("flat", flat_schema_batch(8192)),
-        ("nested", nested_schema_batch(8192)),
+        ("nested", nested_schema_batch(8192, 4, 6)),
     ] {
         let writer =
             ShuffleBlockWriter::try_new(batch.schema().as_ref(), CompressionCodec::None).unwrap();
@@ -307,7 +404,7 @@ fn ipc_context_reuse_benchmark(c: &mut Criterion) {
         for (name, batch) in [
             ("mixed", create_batch(rows, true)),
             ("flat", flat_schema_batch(rows)),
-            ("nested", nested_schema_batch(rows)),
+            ("nested", nested_schema_batch(rows, 4, 6)),
         ] {
             for codec in [
                 CompressionCodec::None,
@@ -370,11 +467,8 @@ fn flat_schema_batch(num_rows: usize) -> RecordBatch {
     RecordBatch::try_new(schema, columns).unwrap()
 }
 
-/// A schema of several deeply nested struct columns.
-fn nested_schema_batch(num_rows: usize) -> RecordBatch {
-    let num_cols = 4;
-    let depth = 6;
-
+/// A schema of `num_cols` struct columns, each nested `depth` levels deep.
+fn nested_schema_batch(num_rows: usize, num_cols: usize, depth: usize) -> RecordBatch {
     let mut fields: Vec<Field> = Vec::with_capacity(num_cols);
     let mut columns: Vec<Arc<dyn Array>> = Vec::with_capacity(num_cols);
     for col in 0..num_cols {
diff --git a/native/shuffle/src/bin/shuffle_bench.rs b/native/shuffle/src/bin/shuffle_bench.rs
index 051e18fbdd..6c7feaf8c7 100644
--- a/native/shuffle/src/bin/shuffle_bench.rs
+++ b/native/shuffle/src/bin/shuffle_bench.rs
@@ -45,7 +45,9 @@ use datafusion::physical_plan::common::collect;
 use datafusion::physical_plan::metrics::{MetricValue, MetricsSet};
 use datafusion::physical_plan::ExecutionPlan;
 use datafusion::prelude::{ParquetReadOptions, SessionContext};
-use datafusion_comet_shuffle::{CometPartitioning, CompressionCodec, ShuffleWriterExec};
+use datafusion_comet_shuffle::{
+    CometPartitioning, CompressionCodec, RoundRobinStrategy, ShuffleWriterExec,
+};
 use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
 use std::fs;
 use std::path::{Path, PathBuf};
@@ -70,7 +72,7 @@ struct Args {
     #[arg(long, default_value_t = 200)]
     partitions: usize,
 
-    /// Partitioning scheme: hash, single, round-robin
+    /// Partitioning scheme: hash, single, round-robin, round-robin-whole-batch
     #[arg(long, default_value = "hash")]
     partitioning: String,
 
@@ -583,7 +585,15 @@ fn build_partitioning(
 ) -> CometPartitioning {
     match scheme {
         "single" => CometPartitioning::SinglePartition,
-        "round-robin" => CometPartitioning::RoundRobin(num_partitions, 0),
+        "round-robin" => {
+            CometPartitioning::RoundRobin(num_partitions, RoundRobinStrategy::default())
+        }
+        "round-robin-whole-batch" => CometPartitioning::RoundRobin(
+            num_partitions,
+            // Standalone bench, so there is no Spark map partition id to carry. One mapper
+            // means the start offset makes no difference to the work measured.
+            RoundRobinStrategy::WholeBatch { start_partition: 0 },
+        ),
         "hash" => {
             let exprs: Vec<Arc<dyn datafusion::physical_expr::PhysicalExpr>> = hash_col_indices
                 .iter()
diff --git a/native/shuffle/src/comet_partitioning.rs b/native/shuffle/src/comet_partitioning.rs
index 15912e6481..16d6565495 100644
--- a/native/shuffle/src/comet_partitioning.rs
+++ b/native/shuffle/src/comet_partitioning.rs
@@ -19,6 +19,38 @@ use arrow::row::{OwnedRow, RowConverter};
 use datafusion::physical_expr::{LexOrdering, PhysicalExpr};
 use std::sync::Arc;
 
+/// How Comet's RoundRobin partitioning assigns rows to output partitions.
+#[derive(Debug, Clone)]
+pub enum RoundRobinStrategy {
+    /// Hash every row across up to `max_hash_columns` columns (0 means no limit)
+    /// and assign rows individually. Deterministic under retry regardless of upstream
+    /// order preservation, but pays the per-row hash cost.
+    HashAll { max_hash_columns: usize },
+    /// Assign each incoming RecordBatch as a whole to one output partition, chosen by a
+    /// per-task batch counter. Skips per-row hashing entirely. Retry-safe only when the
+    /// upstream operator emits the same batches in the same order under retry.
+    ///
+    /// `start_partition` seeds that counter, so mapper i sends its k-th batch to
+    /// `(i + k) % num_partitions`. It must be the Spark map partition id, for two reasons
+    /// that pull in different directions and are both required:
+    ///
+    ///   * distinct across mappers, or every task starts at partition 0 and a task emitting
+    ///     fewer batches than there are output partitions leaves the tail empty stage-wide;
+    ///   * a pure function of the map partition, so a re-executed task reproduces its
+    ///     placement. Spark's own round robin seeds `XORShiftRandom(partitionId)` for the
+    ///     same pair of reasons.
+    WholeBatch { start_partition: usize },
+}
+
+impl Default for RoundRobinStrategy {
+    /// Hash every column, which is what Comet's round robin did before `WholeBatch` existed.
+    fn default() -> Self {
+        Self::HashAll {
+            max_hash_columns: 0,
+        }
+    }
+}
+
 /// Partitioning scheme for distributing rows across shuffle output partitions.
 #[derive(Debug, Clone)]
 pub enum CometPartitioning {
@@ -32,10 +64,9 @@ pub enum CometPartitioning {
     /// Rows for comparing to 4) OwnedRows that represent the boundaries of each partition, used with
     /// LexOrdering to bin each value in the RecordBatch to a partition.
     RangePartitioning(LexOrdering, usize, Arc<RowConverter>, Vec<OwnedRow>),
-    /// Round robin partitioning. Distributes rows across partitions by sorting them by hash
-    /// (computed from columns) and then assigning partitions sequentially. Args are:
-    /// 1) number of partitions, 2) max columns to hash (0 means no limit).
-    RoundRobin(usize, usize),
+    /// Round robin partitioning. Args are the number of partitions and the strategy that
+    /// decides how rows are assigned. See [`RoundRobinStrategy`] for the trade-offs.
+    RoundRobin(usize, RoundRobinStrategy),
 }
 
 impl CometPartitioning {
diff --git a/native/shuffle/src/lib.rs b/native/shuffle/src/lib.rs
index 1158a2b1e2..1844ce2da8 100644
--- a/native/shuffle/src/lib.rs
+++ b/native/shuffle/src/lib.rs
@@ -32,9 +32,18 @@ pub mod spark_unsafe;
 pub(crate) mod writers;
 
 pub use codec_context::ShuffleCodecContext;
-pub use comet_partitioning::CometPartitioning;
+pub use comet_partitioning::{CometPartitioning, RoundRobinStrategy};
 pub use ipc::{read_ipc_compressed, read_ipc_compressed_validated, reset_schema_cache};
 pub use remote_schema::{decode_remote_shuffle_batch, validate_remote_schema};
 pub use schema_align::SchemaAlignExec;
 pub use shuffle_writer::{PartitionOffsets, ShuffleWriterDestination, ShuffleWriterExec};
 pub use writers::{CompressionCodec, ShuffleBlockWriter};
+
+// Bench-only re-exports. `#[doc(hidden)]` marks these as not part of the stable API contract.
+// Consumed by the criterion benches under `native/shuffle/benches/`.
+#[doc(hidden)]
+pub use metrics::ShufflePartitionerMetrics;
+#[doc(hidden)]
+pub use partitioners::{MultiPartitionShuffleRepartitioner, ShufflePartitioner};
+#[doc(hidden)]
+pub use writers::LocalPartitionWriter;
diff --git a/native/shuffle/src/metrics.rs b/native/shuffle/src/metrics.rs
index 855bb95011..10d8106329 100644
--- a/native/shuffle/src/metrics.rs
+++ b/native/shuffle/src/metrics.rs
@@ -20,7 +20,7 @@ use datafusion::physical_plan::metrics::{
 };
 
 /// Execution metrics for a shuffle partition operation.
-pub(crate) struct ShufflePartitionerMetrics {
+pub struct ShufflePartitionerMetrics {
     /// metrics
     pub(crate) baseline: BaselineMetrics,
 
@@ -55,7 +55,7 @@ pub(crate) struct ShufflePartitionerMetrics {
 }
 
 impl ShufflePartitionerMetrics {
-    pub(crate) fn new(metrics: &ExecutionPlanMetricsSet, partition: usize) -> Self {
+    pub fn new(metrics: &ExecutionPlanMetricsSet, partition: usize) -> Self {
         Self {
             baseline: BaselineMetrics::new(metrics, partition),
             repart_time: MetricBuilder::new(metrics).subset_time("repart_time", partition),
diff --git a/native/shuffle/src/partitioners/mod.rs b/native/shuffle/src/partitioners/mod.rs
index 8f4239a820..46d51603ad 100644
--- a/native/shuffle/src/partitioners/mod.rs
+++ b/native/shuffle/src/partitioners/mod.rs
@@ -22,6 +22,6 @@ mod single_partition;
 mod traits;
 
 pub(crate) use empty_schema::EmptySchemaShufflePartitioner;
-pub(crate) use multi_partition::MultiPartitionShuffleRepartitioner;
+pub use multi_partition::MultiPartitionShuffleRepartitioner;
 pub(crate) use single_partition::SinglePartitionShufflePartitioner;
-pub(crate) use traits::ShufflePartitioner;
+pub use traits::ShufflePartitioner;
diff --git a/native/shuffle/src/partitioners/multi_partition.rs b/native/shuffle/src/partitioners/multi_partition.rs
index 37f67eacbd..7028bff9f9 100644
--- a/native/shuffle/src/partitioners/multi_partition.rs
+++ b/native/shuffle/src/partitioners/multi_partition.rs
@@ -19,8 +19,8 @@ use crate::metrics::ShufflePartitionerMetrics;
 use crate::partitioners::partitioned_batch_iterator::PartitionedBatchesProducer;
 use crate::partitioners::ShufflePartitioner;
 use crate::writers::PartitionWriter;
-use crate::{comet_partitioning, CometPartitioning};
-use arrow::array::{Array, ArrayData, ArrayRef, RecordBatch};
+use crate::{comet_partitioning, CometPartitioning, RoundRobinStrategy};
+use arrow::array::{Array, ArrayData, RecordBatch};
 use datafusion::common::utils::proxy::VecAllocExt;
 use datafusion::common::{DataFusionError, HashSet};
 use datafusion::execution::memory_pool::{MemoryConsumer, MemoryReservation};
@@ -99,7 +99,7 @@ impl ScratchSpace {
 }
 
 /// A partitioner that uses a hash function to partition data into multiple partitions
-pub(crate) struct MultiPartitionShuffleRepartitioner<T: PartitionWriter> {
+pub struct MultiPartitionShuffleRepartitioner<T: PartitionWriter> {
     buffered_batches: Vec<RecordBatch>,
     partition_indices: Vec<Vec<(u32, u32)>>,
     partition_writer: T,
@@ -122,6 +122,10 @@ pub(crate) struct MultiPartitionShuffleRepartitioner<T: PartitionWriter> {
     /// allocation once rather than once per slice that references it. Cleared whenever the
     /// buffered batches drain (spill / shuffle_write). See `count_new_buffers`.
     pinned_buffers: HashSet<usize>,
+    /// Batch counter for [`RoundRobinStrategy::WholeBatch`], seeded from that variant's
+    /// `start_partition` and incremented once per input batch slice that reaches
+    /// `partitioning_batch`. Unused by other strategies.
+    round_robin_batch_seq: usize,
 }
 
 /// Sum of the capacities of the backing buffers reachable from `batch` whose start address is
@@ -172,7 +176,7 @@ fn count_new_buffers(batch: &RecordBatch, seen: &mut HashSet<usize>) -> usize {
 
 impl<T: PartitionWriter> MultiPartitionShuffleRepartitioner<T> {
     #[allow(clippy::too_many_arguments)]
-    pub(crate) fn try_new(
+    pub fn try_new(
         partition: usize,
         partition_writer: T,
         partitioning: CometPartitioning,
@@ -191,17 +195,39 @@ impl<T: PartitionWriter> MultiPartitionShuffleRepartitioner<T> {
         // Vectors in the scratch space will be filled with valid values before being used, this
         // initialization code is simply initializing the vectors to the desired size.
         // The initial values are not used.
+        //
+        // Whole-batch RoundRobin is the one strategy that never inspects rows. It needs neither
+        // `partition_ids` nor `partition_row_indices` (~64 KB of dead scratch per task), and it
+        // is the only one that needs a seeded batch counter. Read both off the partitioning here,
+        // before it is moved into the struct below.
+        let whole_batch_start = match &partitioning {
+            CometPartitioning::RoundRobin(
+                _,
+                RoundRobinStrategy::WholeBatch { start_partition },
+            ) => Some(*start_partition),
+            _ => None,
+        };
+        let needs_row_scratch = whole_batch_start.is_none();
         let scratch = ScratchSpace {
-            hashes_buf: match partitioning {
-                // Allocate hashes_buf for hash and round robin partitioning.
-                // Round robin hashes all columns to achieve even, deterministic distribution.
-                CometPartitioning::Hash(_, _) | CometPartitioning::RoundRobin(_, _) => {
+            hashes_buf: match &partitioning {
+                // Allocate hashes_buf for hash and hash-all-columns round robin partitioning.
+                // Whole-batch round robin does no per-row hashing.
+                CometPartitioning::Hash(_, _)
+                | CometPartitioning::RoundRobin(_, RoundRobinStrategy::HashAll { .. }) => {
                     vec![0; batch_size]
                 }
                 _ => vec![],
             },
-            partition_ids: vec![0; batch_size],
-            partition_row_indices: vec![0; batch_size],
+            partition_ids: if needs_row_scratch {
+                vec![0; batch_size]
+            } else {
+                vec![]
+            },
+            partition_row_indices: if needs_row_scratch {
+                vec![0; batch_size]
+            } else {
+                vec![]
+            },
             partition_starts: vec![0; num_output_partitions + 1],
         };
 
@@ -221,6 +247,9 @@ impl<T: PartitionWriter> MultiPartitionShuffleRepartitioner<T> {
             max_buffer_bytes,
             tracing_enabled,
             pinned_buffers: HashSet::new(),
+            // `partition` is the DataFusion partition and is always 0 here, so the seed comes
+            // from the strategy instead. See `RoundRobinStrategy::WholeBatch`.
+            round_robin_batch_seq: whole_batch_start.unwrap_or(0),
         })
     }
 
@@ -290,7 +319,7 @@ impl<T: PartitionWriter> MultiPartitionShuffleRepartitioner<T> {
 
                 self.buffer_partitioned_batch_may_spill(
                     input,
-                    partition_row_indices,
+                    Some(partition_row_indices),
                     partition_starts,
                 )
                 .await?;
@@ -343,68 +372,79 @@ impl<T: PartitionWriter> MultiPartitionShuffleRepartitioner<T> {
 
                 self.buffer_partitioned_batch_may_spill(
                     input,
-                    partition_row_indices,
+                    Some(partition_row_indices),
                     partition_starts,
                 )
                 .await?;
                 self.scratch = scratch;
             }
-            CometPartitioning::RoundRobin(num_output_partitions, max_hash_columns) => {
-                // Comet implements "round robin" as hash partitioning on columns.
-                // This achieves the same goal as Spark's round robin (even distribution
-                // without semantic grouping) while being deterministic for fault tolerance.
-                //
-                // Note: This produces different partition assignments than Spark's round robin,
-                // which sorts by UnsafeRow binary representation before assigning partitions.
-                // However, both approaches provide even distribution and determinism.
+            CometPartitioning::RoundRobin(num_output_partitions, strategy) => {
                 let mut scratch = std::mem::take(&mut self.scratch);
-                let (partition_starts, partition_row_indices): (&Vec<u32>, &Vec<u32>) = {
+                let num_rows = input.num_rows();
+                let partition_row_indices: Option<&[u32]> = {
                     let mut timer = self.metrics.repart_time.timer();
 
-                    let num_rows = input.num_rows();
+                    let indices = match strategy {
+                        RoundRobinStrategy::WholeBatch { .. } => {
+                            let target_idx = self.round_robin_batch_seq % *num_output_partitions;
+                            self.round_robin_batch_seq = self.round_robin_batch_seq.wrapping_add(1);
+
+                            // `target_idx` owns the whole `[0..num_rows)` range and every other
+                            // partition an empty one. `partition_row_indices` stays
+                            // unmaterialized, so `None` below tells the spill path to read that
+                            // range as an identity mapping.
+                            //
+                            // Overwrite in place rather than `clear` + `resize`: the vector is
+                            // always `num_output_partitions + 1` long (set in `try_new`, kept by
+                            // `map_partition_ids_to_starts_and_indices`), and `fill` lowers to a
+                            // memset where `resize` goes through `extend_with`.
+                            let partition_starts = &mut scratch.partition_starts;
+                            partition_starts[..=target_idx].fill(0);
+                            partition_starts[target_idx + 1..].fill(num_rows as u32);
+                            None
+                        }
+                        RoundRobinStrategy::HashAll { max_hash_columns } => {
+                            // Hash-partition rows into pmod(hash, N). This produces different
+                            // partition assignments than Spark's round robin (which sorts by
+                            // UnsafeRow binary representation before assigning partitions), but
+                            // both approaches provide even distribution and determinism.
+                            //
+                            // max_hash_columns of 0 means no limit (hash all columns). Negative
+                            // values are normalized to 0 in the planner.
+                            let num_columns_to_hash = if *max_hash_columns == 0 {
+                                input.num_columns()
+                            } else {
+                                (*max_hash_columns).min(input.num_columns())
+                            };
+                            let columns_to_hash = &input.columns()[..num_columns_to_hash];
+
+                            // Use identical seed as Spark hash partitioning.
+                            let hashes_buf = &mut scratch.hashes_buf[..num_rows];
+                            hashes_buf.fill(42_u32);
+                            create_murmur3_hashes(columns_to_hash, hashes_buf)?;
+
+                            let partition_ids = &mut scratch.partition_ids[..num_rows];
+                            hashes_buf.iter().enumerate().for_each(|(idx, hash)| {
+                                partition_ids[idx] =
+                                    comet_partitioning::pmod(*hash, *num_output_partitions) as u32;
+                            });
 
-                    // Collect columns for hashing, respecting max_hash_columns limit
-                    // max_hash_columns of 0 means no limit (hash all columns)
-                    // Negative values are normalized to 0 in the planner
-                    let num_columns_to_hash = if *max_hash_columns == 0 {
-                        input.num_columns()
-                    } else {
-                        (*max_hash_columns).min(input.num_columns())
+                            scratch.map_partition_ids_to_starts_and_indices(
+                                *num_output_partitions,
+                                num_rows,
+                            );
+                            Some(scratch.partition_row_indices.as_slice())
+                        }
                     };
-                    let columns_to_hash: Vec<ArrayRef> = (0..num_columns_to_hash)
-                        .map(|i| Arc::clone(input.column(i)))
-                        .collect();
-
-                    // Use identical seed as Spark hash partitioning.
-                    let hashes_buf = &mut scratch.hashes_buf[..num_rows];
-                    hashes_buf.fill(42_u32);
-
-                    // Compute hash for selected columns
-                    create_murmur3_hashes(&columns_to_hash, hashes_buf)?;
-
-                    // Assign partition IDs based on hash (same as hash partitioning)
-                    let partition_ids = &mut scratch.partition_ids[..num_rows];
-                    hashes_buf.iter().enumerate().for_each(|(idx, hash)| {
-                        partition_ids[idx] =
-                            comet_partitioning::pmod(*hash, *num_output_partitions) as u32;
-                    });
-
-                    // We now have partition ids for every input row, map that to partition starts
-                    // and partition indices to eventually write these rows to partition buffers.
-                    scratch
-                        .map_partition_ids_to_starts_and_indices(*num_output_partitions, num_rows);
 
                     timer.stop();
-                    Ok::<(&Vec<u32>, &Vec<u32>), DataFusionError>((
-                        &scratch.partition_starts,
-                        &scratch.partition_row_indices,
-                    ))
-                }?;
+                    indices
+                };
 
                 self.buffer_partitioned_batch_may_spill(
                     input,
                     partition_row_indices,
-                    partition_starts,
+                    &scratch.partition_starts,
                 )
                 .await?;
                 self.scratch = scratch;
@@ -423,7 +463,7 @@ impl<T: PartitionWriter> MultiPartitionShuffleRepartitioner<T> {
     async fn buffer_partitioned_batch_may_spill(
         &mut self,
         input: RecordBatch,
-        partition_row_indices: &[u32],
+        partition_row_indices: Option<&[u32]>,
         partition_starts: &[u32],
     ) -> datafusion::common::Result<()> {
         // Charge both the reservation and the data_size metric for the buffers this batch newly
@@ -434,26 +474,29 @@ impl<T: PartitionWriter> MultiPartitionShuffleRepartitioner<T> {
         let buffered_partition_idx = self.buffered_batches.len() as u32;
         self.buffered_batches.push(input);
 
-        // partition_starts conceptually slices partition_row_indices into smaller slices,
-        // each slice contains the indices of rows in input that will go into the corresponding
-        // partition. The following loop iterates over the slices and put the row indices into
-        // the indices array of the corresponding partition.
+        // `partition_starts` slices the input's rows into per-partition ranges: partition K
+        // owns rows `partition_starts[K]..partition_starts[K + 1]`. When `partition_row_indices`
+        // is `Some(indices)`, `indices[start..end]` are the source-row ids in the input for
+        // partition K (see the hash arms). When it is `None`, the source rows are the identity
+        // range `start..end` itself (whole-batch RoundRobin — one target partition owns all
+        // input rows, and the mapping is trivial so we do not materialize it).
         for (partition_id, (&start, &end)) in partition_starts
             .iter()
             .tuple_windows()
             .enumerate()
             .filter(|(_, (start, end))| start < end)
         {
-            let row_indices = &partition_row_indices[start as usize..end as usize];
-
-            // Put row indices for the current partition into the indices array of that partition.
-            // This indices array will be used for calling interleave_record_batch to produce
-            // shuffled batches.
             let indices = &mut self.partition_indices[partition_id];
             let before_size = indices.allocated_size();
-            indices.reserve(row_indices.len());
-            for row_idx in row_indices {
-                indices.push((buffered_partition_idx, *row_idx));
+            match partition_row_indices {
+                Some(row_indices) => indices.extend(
+                    row_indices[start as usize..end as usize]
+                        .iter()
+                        .map(|&row_idx| (buffered_partition_idx, row_idx)),
+                ),
+                None => {
+                    indices.extend((start..end).map(|row_idx| (buffered_partition_idx, row_idx)))
+                }
             }
             let after_size = indices.allocated_size();
             mem_growth += after_size.saturating_sub(before_size);
@@ -625,7 +668,7 @@ impl<T: PartitionWriter> Debug for MultiPartitionShuffleRepartitioner<T> {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use arrow::array::Int64Array;
+    use arrow::array::{ArrayRef, Int64Array};
     use datafusion::physical_plan::metrics::ExecutionPlanMetricsSet;
 
     #[derive(Default)]
@@ -683,7 +726,12 @@ mod tests {
         let mut repartitioner = MultiPartitionShuffleRepartitioner::try_new(
             0,
             FailingPartitionWriter::default(),
-            CometPartitioning::RoundRobin(2, 1),
+            CometPartitioning::RoundRobin(
+                2,
+                RoundRobinStrategy::HashAll {
+                    max_hash_columns: 1,
+                },
+            ),
             ShufflePartitionerMetrics::new(&metrics_set, 0),
             Arc::clone(&runtime),
             64,
@@ -788,7 +836,7 @@ mod tests {
         let mut repartitioner = MultiPartitionShuffleRepartitioner::try_new(
             0,
             FailingPartitionWriter::default(),
-            CometPartitioning::RoundRobin(2, 0),
+            CometPartitioning::RoundRobin(2, RoundRobinStrategy::default()),
             ShufflePartitionerMetrics::new(&metrics_set, 0),
             Arc::clone(&runtime),
             2,
@@ -798,7 +846,7 @@ mod tests {
         .unwrap();
 
         repartitioner
-            .buffer_partitioned_batch_may_spill(batch.slice(0, 2), &[0, 1], &[0, 1, 2])
+            .buffer_partitioned_batch_may_spill(batch.slice(0, 2), Some(&[0, 1]), &[0, 1, 2])
             .await
             .unwrap();
         assert_eq!(repartitioner.reservation.size(), 0);
@@ -816,7 +864,7 @@ mod tests {
 
         repartitioner.max_buffer_bytes = None;
         repartitioner
-            .buffer_partitioned_batch_may_spill(batch.slice(2, 2), &[0, 1], &[0, 1, 2])
+            .buffer_partitioned_batch_may_spill(batch.slice(2, 2), Some(&[0, 1]), &[0, 1, 2])
             .await
             .unwrap();
         let reservation_before_failure = repartitioner.reservation.size();
@@ -912,7 +960,7 @@ mod tests {
                 let mut repartitioner = MultiPartitionShuffleRepartitioner::try_new(
                     0,
                     FailingPartitionWriter::default(),
-                    CometPartitioning::RoundRobin(2, 0),
+                    CometPartitioning::RoundRobin(2, RoundRobinStrategy::default()),
                     ShufflePartitionerMetrics::new(&metrics_set, 0),
                     Arc::clone(&runtime),
                     64,
@@ -922,7 +970,7 @@ mod tests {
                 .unwrap();
                 let row_indices = (0..64).collect::<Vec<_>>();
                 repartitioner
-                    .buffer_partitioned_batch_may_spill(batch, &row_indices, &[0, 32, 64])
+                    .buffer_partitioned_batch_may_spill(batch, Some(&row_indices), &[0, 32, 64])
                     .await
                     .unwrap();
                 // The first reservation includes 512 bytes of input and 512 bytes of indices.
@@ -932,7 +980,11 @@ mod tests {
                 repartitioner.partition_writer.fail = fail;
                 repartitioner.partition_writer.consume_before_failure = consume_before_failure;
                 let result = repartitioner
-                    .buffer_partitioned_batch_may_spill(next_batch, &row_indices, &[0, 32, 64])
+                    .buffer_partitioned_batch_may_spill(
+                        next_batch,
+                        Some(&row_indices),
+                        &[0, 32, 64],
+                    )
                     .await;
                 assert_eq!(result.is_err(), fail);
                 // The indices grow by 512 bytes. Only independent input adds another 512.
diff --git a/native/shuffle/src/partitioners/partitioned_batch_iterator.rs b/native/shuffle/src/partitioners/partitioned_batch_iterator.rs
index 424fa827d5..8385563e74 100644
--- a/native/shuffle/src/partitioners/partitioned_batch_iterator.rs
+++ b/native/shuffle/src/partitioners/partitioned_batch_iterator.rs
@@ -111,6 +111,27 @@ impl<'a> PartitionedBatchIterator<'a> {
             interleave_time,
         }
     }
+
+    /// Returns the source batch when `indices[self.pos..indices_end]` selects every row of a
+    /// single source batch in natural order, so a clone is equivalent to interleaving them.
+    fn whole_source_batch(&self, indices_end: usize) -> Option<RecordBatch> {
+        let indices = &self.indices[self.pos..indices_end];
+        let &(first_batch, _) = indices.first()?;
+        let source = self.record_batches[first_batch as usize];
+        // O(1) reject, which is all a chunk that cannot take this path pays.
+        if indices.len() != source.num_rows() {
+            return None;
+        }
+        // Given the length check, `b == first_batch && r == i` for every element implies the
+        // rest: one source batch, starting at row 0, ending at its last row. `WholeBatch`
+        // guarantees it by construction, but `HashAll` can land a chunk here by chance, so the
+        // scan has to run in release builds too. It short-circuits on the first scattered pair.
+        indices
+            .iter()
+            .enumerate()
+            .all(|(i, &(b, r))| b == first_batch && r as usize == i)
+            .then(|| source.clone())
+    }
 }
 
 impl Iterator for PartitionedBatchIterator<'_> {
@@ -122,6 +143,14 @@ impl Iterator for PartitionedBatchIterator<'_> {
         }
 
         let indices_end = std::cmp::min(self.pos + self.batch_size, self.indices.len());
+
+        // Generic, so `HashAll` benefits opportunistically, but this is the path that makes
+        // `RoundRobinStrategy::WholeBatch` cheap.
+        if let Some(batch) = self.whole_source_batch(indices_end) {
+            self.pos = indices_end;
+            return Some(Ok(batch));
+        }
+
         self.chunk_scratch.clear();
         self.chunk_scratch.extend(
             self.indices[self.pos..indices_end]
diff --git a/native/shuffle/src/partitioners/traits.rs b/native/shuffle/src/partitioners/traits.rs
index 904c7d0088..e89fc3dbf6 100644
--- a/native/shuffle/src/partitioners/traits.rs
+++ b/native/shuffle/src/partitioners/traits.rs
@@ -19,7 +19,7 @@ use arrow::record_batch::RecordBatch;
 use datafusion::common::Result;
 
 #[async_trait::async_trait]
-pub(crate) trait ShufflePartitioner: Send {
+pub trait ShufflePartitioner: Send {
     /// Insert a batch into the partitioner
     async fn insert_batch(&mut self, batch: RecordBatch) -> Result<()>;
     /// Write the buffered shuffle data to the partition writer
diff --git a/native/shuffle/src/rss_execution_tests.rs b/native/shuffle/src/rss_execution_tests.rs
index e9269edc9e..f18409d0ba 100644
--- a/native/shuffle/src/rss_execution_tests.rs
+++ b/native/shuffle/src/rss_execution_tests.rs
@@ -16,7 +16,7 @@
 // under the License.
 
 use crate::{
-    read_ipc_compressed, CometPartitioning, CompressionCodec, PartitionOffsets,
+    read_ipc_compressed, CometPartitioning, CompressionCodec, PartitionOffsets, RoundRobinStrategy,
     ShuffleWriterDestination, ShuffleWriterExec,
 };
 use arrow::array::{Array, Int32Array, RecordBatch, RecordBatchOptions};
@@ -231,7 +231,7 @@ fn rss_multi_partition_supports_hash_range_and_round_robin() {
     for partitioning in [
         CometPartitioning::Hash(vec![expression], 4),
         CometPartitioning::RangePartitioning(ordering, 4, Arc::new(converter), boundaries),
-        CometPartitioning::RoundRobin(4, 0),
+        CometPartitioning::RoundRobin(4, RoundRobinStrategy::default()),
     ] {
         let pusher = Arc::new(RecordingPusher::default());
         let execution = rss_execution(
@@ -271,7 +271,7 @@ fn rss_empty_schema_preserves_row_counts_in_partition_zero() {
     let execution = rss_execution(
         vec![batch.clone(), batch],
         schema,
-        CometPartitioning::RoundRobin(4, 0),
+        CometPartitioning::RoundRobin(4, RoundRobinStrategy::default()),
         pusher.clone(),
         CompressionCodec::None,
         1024 * 1024,
@@ -299,7 +299,7 @@ fn rss_empty_schema_without_rows_does_not_push_frames() {
     let execution = rss_execution(
         vec![batch],
         schema,
-        CometPartitioning::RoundRobin(4, 0),
+        CometPartitioning::RoundRobin(4, RoundRobinStrategy::default()),
         pusher.clone(),
         CompressionCodec::None,
         1024 * 1024,
diff --git a/native/shuffle/src/shuffle_writer.rs b/native/shuffle/src/shuffle_writer.rs
index ca88509a81..f0f6f05164 100644
--- a/native/shuffle/src/shuffle_writer.rs
+++ b/native/shuffle/src/shuffle_writer.rs
@@ -440,7 +440,7 @@ fn contextualize_shuffle_error(error: DataFusionError, phase: &str) -> DataFusio
 #[cfg(test)]
 mod test {
     use super::*;
-    use crate::{read_ipc_compressed, ShuffleBlockWriter, ShuffleCodecContext};
+    use crate::{read_ipc_compressed, RoundRobinStrategy, ShuffleBlockWriter, ShuffleCodecContext};
     use arrow::array::{Array, Int64Array, StringArray, StringBuilder};
     use arrow::datatypes::{DataType, Field, Schema};
     use arrow::record_batch::RecordBatch;
@@ -1143,7 +1143,7 @@ mod test {
                 Arc::new(row_converter),
                 owned_rows,
             ),
-            CometPartitioning::RoundRobin(num_partitions, 0),
+            CometPartitioning::RoundRobin(num_partitions, RoundRobinStrategy::default()),
         ] {
             let batches = (0..num_batches).map(|_| batch.clone()).collect::<Vec<_>>();
 
@@ -1211,7 +1211,7 @@ mod test {
                 Arc::new(DataSourceExec::new(Arc::new(
                     MemorySourceConfig::try_new(partitions, batch.schema(), None).unwrap(),
                 ))),
-                CometPartitioning::RoundRobin(num_partitions, 0),
+                CometPartitioning::RoundRobin(num_partitions, RoundRobinStrategy::default()),
                 CompressionCodec::Zstd(1),
                 data_file.clone(),
                 false,
@@ -1276,6 +1276,275 @@ mod test {
         let _ = fs::remove_file("/tmp/rr_index_1.out");
     }
 
+    #[test]
+    #[cfg_attr(miri, ignore)]
+    fn test_round_robin_batch_granular_retry_deterministic() {
+        // A retry that sees the same batches in the same order must produce byte-identical
+        // shuffle output. This is the contract the batch-granular path relies on in place
+        // of per-row content hashing.
+        use arrow::array::{Float64Array, Int64Array, StringArray, StructArray};
+        use std::fs;
+        use std::io::Read;
+
+        let num_rows = 1024;
+        let num_batches = 6;
+        let num_partitions = 8;
+
+        // A schema with a nested struct to exercise the code path this optimization targets.
+        let leaf_dt = DataType::Struct(
+            vec![
+                Field::new("a", DataType::Int64, false),
+                Field::new("b", DataType::Utf8, false),
+                Field::new("c", DataType::Float64, false),
+            ]
+            .into(),
+        );
+        let wrapped_dt = DataType::Struct(vec![Field::new("leaf", leaf_dt.clone(), false)].into());
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("scalar", DataType::Int64, false),
+            Field::new("nested", wrapped_dt.clone(), false),
+        ]));
+
+        let make_batch = |seed: i64| {
+            let scalar = Int64Array::from_iter_values((0..num_rows as i64).map(|i| i + seed));
+            let leaf = StructArray::from(vec![
+                (
+                    Arc::new(Field::new("a", DataType::Int64, false)),
+                    Arc::new(Int64Array::from_iter_values(
+                        (0..num_rows as i64).map(|i| i * 3 + seed),
+                    )) as Arc<dyn Array>,
+                ),
+                (
+                    Arc::new(Field::new("b", DataType::Utf8, false)),
+                    Arc::new(StringArray::from_iter_values(
+                        (0..num_rows).map(|i| format!("v{i}")),
+                    )) as Arc<dyn Array>,
+                ),
+                (
+                    Arc::new(Field::new("c", DataType::Float64, false)),
+                    Arc::new(Float64Array::from_iter_values(
+                        (0..num_rows).map(|i| (i as f64) * 0.5),
+                    )) as Arc<dyn Array>,
+                ),
+            ]);
+            let nested = StructArray::from(vec![(
+                Arc::new(Field::new("leaf", leaf_dt.clone(), false)),
+                Arc::new(leaf) as Arc<dyn Array>,
+            )]);
+            RecordBatch::try_new(
+                Arc::clone(&schema),
+                vec![Arc::new(scalar), Arc::new(nested)],
+            )
+            .unwrap()
+        };
+        let batches: Vec<RecordBatch> = (0..num_batches as i64).map(make_batch).collect();
+
+        let dir = tempfile::tempdir().unwrap();
+        let run = |tag: &str| -> (Vec<u8>, Vec<i64>) {
+            let data_file = dir.path().join(format!("{tag}_data.out"));
+            let partitions = std::slice::from_ref(&batches);
+            let exec = ShuffleWriterExec::try_new(
+                Arc::new(DataSourceExec::new(Arc::new(
+                    MemorySourceConfig::try_new(partitions, Arc::clone(&schema), None).unwrap(),
+                ))),
+                CometPartitioning::RoundRobin(
+                    num_partitions,
+                    RoundRobinStrategy::WholeBatch { start_partition: 0 },
+                ),
+                CompressionCodec::Zstd(1),
+                data_file.to_str().unwrap().to_string(),
+                false,
+                1024 * 1024,
+                None,
+            )
+            .unwrap();
+
+            let ctx = SessionContext::new_with_config_rt(
+                SessionConfig::new(),
+                Arc::new(
+                    RuntimeEnvBuilder::new()
+                        .with_memory_limit(10 * 1024 * 1024, 1.0)
+                        .build()
+                        .unwrap(),
+                ),
+            );
+            let stream = exec.execute(0, ctx.task_ctx()).unwrap();
+            Runtime::new().unwrap().block_on(collect(stream)).unwrap();
+
+            let mut data = Vec::new();
+            fs::File::open(&data_file)
+                .unwrap()
+                .read_to_end(&mut data)
+                .unwrap();
+            let offsets = exec
+                .partition_offsets()
+                .expect("local destination publishes offsets")
+                .get()
+                .expect("writer published its partition offsets")
+                .to_vec();
+            (data, offsets)
+        };
+
+        let (data_a, offsets_a) = run("attempt0");
+        let (data_b, offsets_b) = run("attempt1");
+        assert_eq!(
+            data_a, data_b,
+            "batch-granular round robin must produce byte-identical shuffle data on retry"
+        );
+        assert_eq!(
+            offsets_a, offsets_b,
+            "batch-granular round robin must produce identical partition offsets on retry"
+        );
+
+        // Sanity: the assignment must have distributed the 6 batches across the partitions
+        // (specifically, the offsets must show non-trivial partition sizes, not all in one
+        // bucket).
+        assert_eq!(offsets_a.len(), num_partitions + 1);
+        let sizes: Vec<i64> = offsets_a.windows(2).map(|w| w[1] - w[0]).collect();
+        let nonempty = sizes.iter().filter(|&&s| s > 0).count();
+        assert!(
+            nonempty >= 2,
+            "batch-granular round robin should populate at least 2 partitions across 6 batches; got sizes {sizes:?}"
+        );
+
+        // And every batch's rows must show up somewhere in the output.
+        let total_rows: usize = read_all_ipc_batches(&data_a)
+            .iter()
+            .map(|b| b.num_rows())
+            .sum();
+        assert_eq!(total_rows, num_rows * num_batches, "no rows may be dropped");
+    }
+
+    #[test]
+    #[cfg_attr(miri, ignore)]
+    fn test_round_robin_batch_granular_starts_at_map_partition() {
+        // Every mapper seeding 0 would send its first batch to partition 0, so a stage whose
+        // tasks each emit fewer batches than there are output partitions would leave the tail
+        // partitions empty everywhere. `start_partition` carries the Spark map partition id, so
+        // mapper i starts at partition i and the stage covers the full range.
+        let num_partitions = 8;
+        let batch = create_batch(100);
+        let batches: Vec<RecordBatch> = (0..2).map(|_| batch.clone()).collect();
+        let dir = tempfile::tempdir().unwrap();
+
+        let nonempty_partitions = |start_partition: usize| -> Vec<usize> {
+            let data_file = dir.path().join(format!("start{start_partition}.out"));
+            let exec = ShuffleWriterExec::try_new(
+                Arc::new(DataSourceExec::new(Arc::new(
+                    MemorySourceConfig::try_new(
+                        std::slice::from_ref(&batches),
+                        batch.schema(),
+                        None,
+                    )
+                    .unwrap(),
+                ))),
+                CometPartitioning::RoundRobin(
+                    num_partitions,
+                    RoundRobinStrategy::WholeBatch { start_partition },
+                ),
+                CompressionCodec::Zstd(1),
+                data_file.to_str().unwrap().to_string(),
+                false,
+                1024 * 1024,
+                None,
+            )
+            .unwrap();
+
+            let ctx = SessionContext::new_with_config_rt(
+                SessionConfig::new(),
+                Arc::new(RuntimeEnvBuilder::new().build().unwrap()),
+            );
+            let stream = exec.execute(0, ctx.task_ctx()).unwrap();
+            Runtime::new().unwrap().block_on(collect(stream)).unwrap();
+
+            let offsets = exec
+                .partition_offsets()
+                .expect("local destination publishes offsets")
+                .get()
+                .expect("writer published its partition offsets")
+                .to_vec();
+            (0..num_partitions)
+                .filter(|&p| offsets[p + 1] > offsets[p])
+                .collect()
+        };
+
+        assert_eq!(nonempty_partitions(0), vec![0, 1]);
+        assert_eq!(
+            nonempty_partitions(3),
+            vec![3, 4],
+            "mapper 3 must not start where mapper 0 does"
+        );
+        // A map partition id above the output partition count wraps rather than panicking.
+        assert_eq!(nonempty_partitions(num_partitions + 1), vec![1, 2]);
+    }
+
+    #[test]
+    #[cfg_attr(miri, ignore)]
+    fn test_round_robin_batch_granular_whole_batch_per_partition() {
+        // Each input batch must land entirely on a single output partition. Reading back
+        // the IPC blocks from any one partition should give whole batches (num_rows each),
+        // not row-level slices.
+        use std::fs;
+
+        let num_rows = 500;
+        let num_batches = 4;
+        let num_partitions = 8;
+
+        let batch = create_batch(num_rows);
+        let batches: Vec<RecordBatch> = (0..num_batches).map(|_| batch.clone()).collect();
+
+        let dir = tempfile::tempdir().unwrap();
+        let data_file = dir.path().join("data.out");
+
+        let partitions = std::slice::from_ref(&batches);
+        let exec = ShuffleWriterExec::try_new(
+            Arc::new(DataSourceExec::new(Arc::new(
+                MemorySourceConfig::try_new(partitions, batch.schema(), None).unwrap(),
+            ))),
+            CometPartitioning::RoundRobin(
+                num_partitions,
+                RoundRobinStrategy::WholeBatch { start_partition: 0 },
+            ),
+            CompressionCodec::Zstd(1),
+            data_file.to_str().unwrap().to_string(),
+            false,
+            1024 * 1024,
+            None,
+        )
+        .unwrap();
+
+        let ctx = SessionContext::new_with_config_rt(
+            SessionConfig::new(),
+            Arc::new(RuntimeEnvBuilder::new().build().unwrap()),
+        );
+        let stream = exec.execute(0, ctx.task_ctx()).unwrap();
+        Runtime::new().unwrap().block_on(collect(stream)).unwrap();
+
+        let data = fs::read(&data_file).unwrap();
+        let offsets = exec
+            .partition_offsets()
+            .expect("local destination publishes offsets")
+            .get()
+            .expect("writer published its partition offsets");
+        assert_eq!(offsets.len(), num_partitions + 1);
+
+        for p in 0..num_partitions {
+            let start = offsets[p] as usize;
+            let end = offsets[p + 1] as usize;
+            if start == end {
+                continue;
+            }
+            let batches = read_all_ipc_batches(&data[start..end]);
+            for b in &batches {
+                assert_eq!(
+                    b.num_rows(),
+                    num_rows,
+                    "each shuffle block in partition {p} must contain a whole input batch"
+                );
+            }
+        }
+    }
+
     /// Test that batch coalescing in BufBatchWriter reduces output size by
     /// writing fewer, larger IPC blocks instead of many small ones.
     #[test]
@@ -1615,7 +1884,7 @@ mod test {
             Arc::new(DataSourceExec::new(Arc::new(
                 MemorySourceConfig::try_new(partitions, Arc::clone(&schema), None).unwrap(),
             ))),
-            CometPartitioning::RoundRobin(num_partitions, 0),
+            CometPartitioning::RoundRobin(num_partitions, RoundRobinStrategy::default()),
             CompressionCodec::Zstd(1),
             data_file.to_str().unwrap().to_string(),
             false,
@@ -1699,7 +1968,7 @@ mod test {
             Arc::new(DataSourceExec::new(Arc::new(
                 MemorySourceConfig::try_new(partitions, Arc::clone(&schema), None).unwrap(),
             ))),
-            CometPartitioning::RoundRobin(num_partitions, 0),
+            CometPartitioning::RoundRobin(num_partitions, RoundRobinStrategy::default()),
             CompressionCodec::Zstd(1),
             data_file.to_str().unwrap().to_string(),
             false,
diff --git a/native/shuffle/src/writers/local/local_partition_writer.rs b/native/shuffle/src/writers/local/local_partition_writer.rs
index d76ff23eff..48775f0030 100644
--- a/native/shuffle/src/writers/local/local_partition_writer.rs
+++ b/native/shuffle/src/writers/local/local_partition_writer.rs
@@ -77,7 +77,7 @@ enum DataOutput {
 /// Writes shuffle output to a single data file and publishes the byte offset where
 /// each partition begins through [`PartitionOffsets`]. See [`DataOutput`] for how the
 /// single- and multi-partition modes differ.
-pub(crate) struct LocalPartitionWriter {
+pub struct LocalPartitionWriter {
     partition_offsets: Arc<PartitionOffsets>,
     data_output: DataOutput,
     /// Compression state shared by every block this task writes; the per-partition
@@ -97,7 +97,7 @@ pub(crate) struct LocalPartitionWriter {
 }
 
 impl LocalPartitionWriter {
-    pub(crate) fn try_new(
+    pub fn try_new(
         output_data_file: String,
         partition_offsets: Arc<PartitionOffsets>,
         shuffle_block_writer: ShuffleBlockWriter,
diff --git a/native/shuffle/src/writers/mod.rs b/native/shuffle/src/writers/mod.rs
index 4586e46c25..350a1fd393 100644
--- a/native/shuffle/src/writers/mod.rs
+++ b/native/shuffle/src/writers/mod.rs
@@ -24,7 +24,7 @@ mod shuffle_block_writer;
 
 pub(crate) use buf_batch_writer::BufBatchWriter;
 pub(crate) use checksum::Checksum;
-pub(crate) use local::local_partition_writer::LocalPartitionWriter;
+pub use local::local_partition_writer::LocalPartitionWriter;
 pub(crate) use partition_writer::PartitionWriter;
 pub(crate) use rss::rss_partition_writer::RssPartitionWriter;
 pub use shuffle_block_writer::{CompressionCodec, ShuffleBlockWriter};
diff --git a/native/shuffle/src/writers/partition_writer.rs b/native/shuffle/src/writers/partition_writer.rs
index 0fcec17c00..fdba82cc27 100644
--- a/native/shuffle/src/writers/partition_writer.rs
+++ b/native/shuffle/src/writers/partition_writer.rs
@@ -32,7 +32,7 @@ use arrow::record_batch::RecordBatch;
 /// ascending id order, then a single [`finish_all`](PartitionWriter::finish_all).
 ///
 /// [`LocalPartitionWriter`]: crate::writers::local::local_partition_writer::LocalPartitionWriter
-pub(crate) trait PartitionWriter: Send {
+pub trait PartitionWriter: Send {
     /// Stages the batches from `iter` for partition `pid` without finalizing it.
     ///
     /// Used to stream single-partition output and to stage multi-partition
diff --git a/spark/src/main/scala/org/apache/comet/CometConf.scala b/spark/src/main/scala/org/apache/comet/CometConf.scala
index d371f1ba44..ca945bdd61 100644
--- a/spark/src/main/scala/org/apache/comet/CometConf.scala
+++ b/spark/src/main/scala/org/apache/comet/CometConf.scala
@@ -475,6 +475,38 @@ object CometConf extends ShimCometConf {
         "The maximum number of columns to hash for round robin partitioning must be non-negative.")
       .createWithDefault(0)
 
+  val COMET_SHUFFLE_NATIVE_ROUND_ROBIN_PARTITIONING_BATCH_GRANULAR: ConfigEntry[Boolean] =
+    conf("spark.comet.shuffle.native.partitioning.roundrobin.batchGranular")
+      .category(CATEGORY_SHUFFLE)
+      .doc("When true, Comet's native round-robin shuffle assigns each RecordBatch as a whole " +
+        "to one output partition instead of hashing every column of every row. This is much " +
+        "cheaper on wide/nested schemas. Distribution is quasi-even at batch granularity. " +
+        "Retry-safe only when the upstream operator emits the same batches in the same order " +
+        "under retry (Comet's Parquet scan and other order-preserving operators satisfy this). " +
+        s"Has no effect unless ${COMET_SHUFFLE_NATIVE_ROUND_ROBIN_PARTITIONING_ENABLED.key} " +
+        "is also true.")
+      .booleanConf
+      .createWithDefault(false)
+
+  val COMET_SHUFFLE_NATIVE_ROUND_ROBIN_PARTITIONING_FAIL_ON_RETRY: ConfigEntry[Boolean] =
+    conf("spark.comet.shuffle.native.partitioning.roundrobin.batchGranular.failOnRetry")
+      .category(CATEGORY_SHUFFLE)
+      .doc(
+        "Only applies when " +
+          s"${COMET_SHUFFLE_NATIVE_ROUND_ROBIN_PARTITIONING_BATCH_GRANULAR.key} is true. " +
+          "Batch-granular round robin places rows by position rather than by content, so a " +
+          "re-executed map task can send rows to different output partitions than the attempt " +
+          "it replaces, silently dropping and duplicating rows once any of that output has " +
+          "been fetched. When true, such a map task fails immediately instead of writing " +
+          "output, which fails the stage and therefore the job. A retry here means any task " +
+          "attempt after the first (speculative attempts included) or any attempt of a " +
+          "re-submitted stage, which is what an executor loss produces. Set to false to keep " +
+          "Spark's normal fault tolerance, leaving safety to whole-stage rollback, which Comet " +
+          "requests by declaring the shuffle input RDD INDETERMINATE whenever its own input " +
+          "is not DETERMINATE.")
+      .booleanConf
+      .createWithDefault(true)
+
   val COMET_SHUFFLE_CONVERT_FROM_SPARK_PLAN_ENABLED: ConfigEntry[Boolean] =
     conf("spark.comet.shuffle.convertFromSparkPlan.enabled")
       .withAlternative(s"$COMET_EXEC_CONFIG_PREFIX.shuffle.convertFromSparkPlan.enabled")
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleInputRDD.scala b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleInputRDD.scala
index bced55b34d..1682772702 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleInputRDD.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleInputRDD.scala
@@ -20,7 +20,7 @@
 package org.apache.spark.sql.comet.execution.shuffle
 
 import org.apache.spark._
-import org.apache.spark.rdd.RDD
+import org.apache.spark.rdd.{DeterministicLevel, RDD}
 import org.apache.spark.sql.comet.{CometExecRDD, CometMetricNode}
 import org.apache.spark.sql.vectorized.ColumnarBatch
 
@@ -33,6 +33,10 @@ import org.apache.comet.CometShuffleBlockIterator
  * [[CometNativeShuffleInputIterator]]. The iterator reports `hasNext = false`;
  * [[CometNativeShuffleWriter]] downcasts it and reads those slots directly to drive the unified
  * `ShuffleWriter(child = childNativeOp)` plan.
+ *
+ * @param positionalRoundRobin
+ *   whether the writer fed by this RDD will place rows by position instead of by content; see
+ *   [[CometShuffleExchangeExec.usesPositionalRoundRobin]] and `getOutputDeterministicLevel`.
  */
 private[shuffle] class CometNativeShuffleInputRDD(
     sc: SparkContext,
@@ -40,7 +44,8 @@ private[shuffle] class CometNativeShuffleInputRDD(
     numPartitionsParam: Int,
     shuffleScanIndices: Set[Int],
     spillMetricNode: CometMetricNode,
-    @transient perPartitionByKey: Map[String, Array[Array[Byte]]] = Map.empty)
+    @transient perPartitionByKey: Map[String, Array[Array[Byte]]] = Map.empty,
+    positionalRoundRobin: Boolean = false)
     extends RDD[Product2[Int, ColumnarBatch]](
       sc,
       inputRDDs.map(rdd => new OneToOneDependency(rdd))) {
@@ -57,7 +62,29 @@ private[shuffle] class CometNativeShuffleInputRDD(
       numPartitionsParam,
       shuffleScanIndices,
       spillMetricNode,
-      perPartitionByKey)
+      perPartitionByKey,
+      positionalRoundRobin)
+
+  /**
+   * Spark handles the retry hazard of positional round robin declaratively rather than
+   * per-operator: it wraps the repartition in a `MapPartitionsRDD` with `isOrderSensitive = true`
+   * (Comet's own JVM path does this in `prepareJVMShuffleDependency`), and that RDD reports
+   * `INDETERMINATE` whenever its parent is `UNORDERED`, which makes the DAGScheduler roll the
+   * whole stage back instead of re-running one task. The native path has no `MapPartitionsRDD` to
+   * carry the flag, so apply the same rule here.
+   *
+   * Letting the parent level discriminate is what keeps a plain scan on the cheap per-task retry
+   * path. See the `WholeBatch` retry-safety section of
+   * `docs/source/contributor-guide/native_shuffle.md`.
+   */
+  override protected def getOutputDeterministicLevel: DeterministicLevel.Value = {
+    val inheritedLevel = super.getOutputDeterministicLevel
+    if (positionalRoundRobin && inheritedLevel != DeterministicLevel.DETERMINATE) {
+      DeterministicLevel.INDETERMINATE
+    } else {
+      inheritedLevel
+    }
+  }
 
   override protected def getPartitions: Array[Partition] =
     (0 until numPartitionsParam).map { i =>
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleWriter.scala b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleWriter.scala
index fce4291deb..7c0278b57a 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleWriter.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleWriter.scala
@@ -38,7 +38,7 @@ import org.apache.spark.sql.execution.metric.SQLMetric
 import org.apache.spark.sql.types.StructField
 import org.apache.spark.util.{ThreadUtils, Utils}
 
-import org.apache.comet.{CometConf, CometExecIterator, CometShuffleSizeLimitException}
+import org.apache.comet.{CometConf, CometExecIterator, CometRuntimeException, CometShuffleSizeLimitException}
 import org.apache.comet.serde.{OperatorOuterClass, PartitioningOuterClass, QueryPlanSerde}
 import org.apache.comet.serde.OperatorOuterClass.{CompressionCodec, Operator}
 import org.apache.comet.serde.operator.schema2Proto
@@ -108,7 +108,57 @@ class CometNativeShuffleWriter[K, V](
     }
   }
 
+  /**
+   * Refuse to re-execute a map task whose round-robin placement is positional. See the
+   * `WholeBatch` retry-safety section of `docs/source/contributor-guide/native_shuffle.md`.
+   *
+   * [[CometNativeShuffleInputRDD.getOutputDeterministicLevel]] is the other half of the defence.
+   * This half does not trust the parent's declared determinism level to describe what the
+   * upstream operator actually replays, so while the strategy is opt-in and unproven the default
+   * is to fail rather than rely on rollback.
+   *
+   * Both counters are needed and neither subsumes the other. `attemptNumber` covers a task re-run
+   * inside the current stage attempt: a task-level failure, speculation, or an executor lost
+   * while the stage was still running. `stageAttemptNumber` covers a re-submitted stage, which is
+   * what a fetch failure against a dead executor's shuffle output produces, and which also
+   * re-runs map tasks that had already succeeded elsewhere.
+   *
+   * There is no Spark API for failing an application from inside a task, and the two exceptions
+   * Spark declines to retry are matched by class name (`NotSerializableException`,
+   * `TaskOutputFileAlreadyExistException`), so this throws an ordinary exception. Because the
+   * condition is sticky (a later attempt only has a higher `attemptNumber`) every remaining
+   * attempt fails here too and the task set aborts after `spark.task.maxFailures`, failing the
+   * stage and the job. Each of those attempts fails before doing any work.
+   */
+  private def failIfRetryingPositionalRoundRobin(): Unit = {
+    if (!CometShuffleExchangeExec.usesPositionalRoundRobin(outputPartitioning)) return
+    if (!CometConf.COMET_SHUFFLE_NATIVE_ROUND_ROBIN_PARTITIONING_FAIL_ON_RETRY.get()) return
+    // Nullable for the same reason as the `Option(context)` guard on `cancellationWatch` above.
+    val taskContext = context
+    if (taskContext == null) return
+    val taskAttempt = taskContext.attemptNumber()
+    val stageAttempt = taskContext.stageAttemptNumber()
+    if (taskAttempt == 0 && stageAttempt == 0) return
+
+    val batchGranularKey =
+      CometConf.COMET_SHUFFLE_NATIVE_ROUND_ROBIN_PARTITIONING_BATCH_GRANULAR.key
+    val failOnRetryKey =
+      CometConf.COMET_SHUFFLE_NATIVE_ROUND_ROBIN_PARTITIONING_FAIL_ON_RETRY.key
+    throw new CometRuntimeException(
+      s"Refusing to re-execute a Comet native round-robin shuffle map task: shuffle $shuffleId, " +
+        s"map partition ${taskContext.partitionId()}, stage ${taskContext.stageId()} " +
+        s"(attempt $stageAttempt), task attempt $taskAttempt. `$batchGranularKey` is enabled, " +
+        "so each Arrow batch is assigned to an output partition by position rather than by " +
+        "hashing its rows, and this attempt may place rows differently than the attempt it " +
+        "replaces, duplicating and dropping rows on the reduce side without any error. Failing " +
+        s"the job instead. Set `$batchGranularKey` to false to use content-hash round robin, " +
+        s"which is safe to re-execute, or set `$failOnRetryKey` to false to allow retries and " +
+        "rely on Spark rolling the stage back.")
+  }
+
   private def writeInternal(inputs: Iterator[Product2[K, V]]): Unit = {
+    failIfRetryingPositionalRoundRobin()
+
     val localOutput = if (remoteDestination.isEmpty) {
       val resolver =
         SparkEnv.get.shuffleManager.shuffleBlockResolver.asInstanceOf[IndexShuffleBlockResolver]
@@ -421,6 +471,8 @@ class CometNativeShuffleWriter[K, V](
         partitioning.setNumPartitions(effectivePartitionCount)
         partitioning.setMaxHashColumns(
           CometConf.COMET_SHUFFLE_NATIVE_ROUND_ROBIN_PARTITIONING_MAX_HASH_COLUMNS.get())
+        partitioning.setBatchGranular(
+          CometConf.COMET_SHUFFLE_NATIVE_ROUND_ROBIN_PARTITIONING_BATCH_GRANULAR.get())
 
         val partitioningBuilder = PartitioningOuterClass.Partitioning.newBuilder()
         shuffleWriterBuilder.setPartitioning(
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala
index 2a53964375..e5629d032c 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala
@@ -127,7 +127,8 @@ case class CometShuffleExchangeExec(
           ctx.numPartitions,
           ctx.shuffleScanIndices,
           CometMetricNode(metrics, Seq(nativeChildMetricNode)),
-          ctx.perPartitionByKey)
+          ctx.perPartitionByKey,
+          CometShuffleExchangeExec.usesPositionalRoundRobin(outputPartitioning))
       case None =>
         // Non-native child (e.g. CometSparkToColumnarExec): no subtree to inline. The dep gets
         // built via the convenience overload below; we just need a real RDD of batches.
@@ -298,6 +299,25 @@ object CometShuffleExchangeExec
     if (shuffleSupported(op).isDefined) Compatible() else Unsupported()
   }
 
+  /**
+   * True when this exchange will run the native round-robin writer in its positional
+   * (batch-granular) mode, where an input `RecordBatch` is assigned whole to one output partition
+   * by a per-task counter rather than by hashing its rows. See the `WholeBatch` retry-safety
+   * section of `docs/source/contributor-guide/native_shuffle.md` for why that is unsafe to
+   * re-execute.
+   *
+   * Must stay in step with `PhysicalPlanner::create_partitioning`, which turns the
+   * `batch_granular` proto field into `RoundRobinStrategy::WholeBatch`. Nothing enforces that.
+   *
+   * The `numPartitions > 1` guard mirrors `isRoundRobin` in `prepareJVMShuffleDependency`. With a
+   * single output partition every row lands in the same place, so there is no placement to get
+   * wrong, and native routes that case to `SinglePartitionShufflePartitioner` regardless.
+   */
+  def usesPositionalRoundRobin(outputPartitioning: Partitioning): Boolean =
+    outputPartitioning.isInstanceOf[RoundRobinPartitioning] &&
+      outputPartitioning.numPartitions > 1 &&
+      CometConf.COMET_SHUFFLE_NATIVE_ROUND_ROBIN_PARTITIONING_BATCH_GRANULAR.get()
+
   override def createExec(
       nativeOp: OperatorOuterClass.Operator,
       op: ShuffleExchangeExec): CometNativeExec = {
@@ -777,7 +797,8 @@ object CometShuffleExchangeExec
       Seq(streamRDD),
       rdd.getNumPartitions,
       shuffleScanIndices = Set.empty,
-      spillMetricNode = CometMetricNode(metrics, Seq(childMetricNode)))
+      spillMetricNode = CometMetricNode(metrics, Seq(childMetricNode)),
+      positionalRoundRobin = usesPositionalRoundRobin(outputPartitioning))
 
     val ctx = NativeExecContext(
       inputs = Seq(streamRDD),
diff --git a/spark/src/test/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeRoundRobinRetrySuite.scala b/spark/src/test/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeRoundRobinRetrySuite.scala
new file mode 100644
index 0000000000..e70fe7b121
--- /dev/null
+++ b/spark/src/test/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeRoundRobinRetrySuite.scala
@@ -0,0 +1,194 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.comet.execution.shuffle
+
+import java.util.Properties
+
+import org.apache.spark.{TaskContext, TaskContextImpl}
+import org.apache.spark.executor.TaskMetrics
+import org.apache.spark.rdd.DeterministicLevel
+import org.apache.spark.sql.{CometTestBase, DataFrame}
+import org.apache.spark.sql.catalyst.expressions.Literal
+import org.apache.spark.sql.catalyst.plans.physical.{HashPartitioning, Partitioning, RangePartitioning, RoundRobinPartitioning, SinglePartition}
+import org.apache.spark.sql.comet.CometMetricNode
+import org.apache.spark.sql.internal.SQLConf
+
+import org.apache.comet.{CometConf, CometRuntimeException}
+
+/**
+ * Retry safety for the batch-granular (positional) native round-robin strategy.
+ *
+ * That strategy assigns a whole Arrow batch to an output partition from a per-task counter, so a
+ * row's destination depends on the order and the framing of the batches the upstream operator
+ * produced rather than on the row itself. Re-executing a map task can therefore write a different
+ * partitioning of the same rows, which duplicates and drops rows on the reduce side once any of
+ * the replaced output has been fetched. This suite covers the writer refusing to run a retried
+ * map task. The other defence, declaring the shuffle input RDD indeterminate so the DAGScheduler
+ * rolls the stage back, is covered in [[CometNativeShuffleInputRDDSuite]].
+ *
+ * Lives in the `execution.shuffle` package so it can construct the `private[shuffle]`
+ * [[CometNativeShuffleInputRDD]] and the `private[spark]` [[TaskContextImpl]] directly.
+ */
+class CometNativeRoundRobinRetrySuite extends CometTestBase {
+
+  private val batchGranularKey =
+    CometConf.COMET_SHUFFLE_NATIVE_ROUND_ROBIN_PARTITIONING_BATCH_GRANULAR.key
+  private val failOnRetryKey =
+    CometConf.COMET_SHUFFLE_NATIVE_ROUND_ROBIN_PARTITIONING_FAIL_ON_RETRY.key
+
+  /** Mirrors `TaskContext.empty()`, which hard-codes both attempt numbers to zero. */
+  private def taskContextFor(stageAttempt: Int, taskAttempt: Int): TaskContext =
+    new TaskContextImpl(
+      0,
+      stageAttempt,
+      0,
+      0L,
+      taskAttempt,
+      1,
+      null,
+      new Properties,
+      null,
+      TaskMetrics.empty,
+      1)
+
+  private def writerFor(
+      outputPartitioning: Partitioning,
+      taskContext: TaskContext): CometNativeShuffleWriter[Int, Any] =
+    new CometNativeShuffleWriter[Int, Any](
+      NativeShuffleSpec(null, CometMetricNode(Map.empty), null),
+      outputPartitioning,
+      Nil,
+      Map.empty,
+      4,
+      0,
+      0L,
+      taskContext,
+      null)
+
+  /**
+   * `write` is driven with a plain iterator rather than a [[CometNativeShuffleInputIterator]], so
+   * anything that gets past the retry guard fails soon after on that. Returning the message lets
+   * a caller assert on which of the two happened without depending on the unrelated failure's
+   * type.
+   */
+  private def writeFailureMessage(writer: CometNativeShuffleWriter[Int, Any]): String = {
+    val failure = intercept[Throwable](writer.write(Iterator.empty))
+    Option(failure.getMessage).getOrElse("")
+  }
+
+  private val refusalPrefix = "Refusing to re-execute a Comet native round-robin"
+
+  test("usesPositionalRoundRobin claims only multi-partition round robin with the config on") {
+    val partitionings = Seq(
+      RoundRobinPartitioning(4),
+      // One output partition puts every row in the same place, so there is no placement to get
+      // wrong. `isRoundRobin` in `prepareJVMShuffleDependency` excludes it for the same reason.
+      RoundRobinPartitioning(1),
+      SinglePartition,
+      HashPartitioning(Seq(Literal(1)), 4),
+      RangePartitioning(Nil, 4))
+    withSQLConf(batchGranularKey -> "true") {
+      val claimed = partitionings.filter(CometShuffleExchangeExec.usesPositionalRoundRobin)
+      assert(claimed == Seq(RoundRobinPartitioning(4)))
+    }
+    withSQLConf(batchGranularKey -> "false") {
+      assert(!partitionings.exists(CometShuffleExchangeExec.usesPositionalRoundRobin))
+    }
+  }
+
+  test("positional round robin refuses to run a retried map task") {
+    withSQLConf(batchGranularKey -> "true") {
+      // A task re-run inside the current stage attempt, and a re-submitted stage, which is what
+      // losing an executor produces. Neither counter subsumes the other.
+      Seq(("task attempt", 0, 1), ("stage attempt", 1, 0), ("both", 2, 3)).foreach {
+        case (label, stageAttempt, taskAttempt) =>
+          val writer =
+            writerFor(RoundRobinPartitioning(4), taskContextFor(stageAttempt, taskAttempt))
+          val failure = intercept[CometRuntimeException](writer.write(Iterator.empty))
+          assert(failure.getMessage.startsWith(refusalPrefix), label)
+          // The message has to say which knob to turn; an operator hitting this mid-job has no
+          // other signal that positional placement is what failed them.
+          assert(failure.getMessage.contains(batchGranularKey), label)
+          assert(failure.getMessage.contains(failOnRetryKey), label)
+      }
+    }
+  }
+
+  test("the retry guard fires only for positional round robin on a retried attempt") {
+    // Everything here must reach the writer. Hash and range partitioning place rows by content,
+    // content-hash round robin does too, a first attempt is not a retry, and failOnRetry=false
+    // hands retry handling back to the DAGScheduler.
+    Seq(
+      ("first attempt", "true", "true", RoundRobinPartitioning(4): Partitioning, 0, 0),
+      ("batchGranular off", "false", "true", RoundRobinPartitioning(4), 1, 1),
+      ("failOnRetry off", "true", "false", RoundRobinPartitioning(4), 1, 1),
+      ("single output partition", "true", "true", RoundRobinPartitioning(1), 1, 1),
+      ("hash partitioning", "true", "true", HashPartitioning(Seq(Literal(1)), 4), 1, 1),
+      ("range partitioning", "true", "true", RangePartitioning(Nil, 4), 1, 1),
+      ("single partition", "true", "true", SinglePartition, 1, 1)).foreach {
+      case (label, granular, failOnRetry, partitioning, stageAttempt, taskAttempt) =>
+        withSQLConf(batchGranularKey -> granular, failOnRetryKey -> failOnRetry) {
+          val writer = writerFor(partitioning, taskContextFor(stageAttempt, taskAttempt))
+          assert(!writeFailureMessage(writer).startsWith(refusalPrefix), label)
+        }
+    }
+  }
+
+  /**
+   * The determinism level the DAGScheduler would read off a planned round-robin exchange.
+   * `ShuffleMapStage.rdd` is the dependency's RDD, so this is exactly what
+   * `Stage.isIndeterminate` sees.
+   */
+  private def roundRobinStageLevel(df: DataFrame): DeterministicLevel.Value = {
+    val plan = df.queryExecution.executedPlan
+    val exchanges = plan.collect {
+      case exchange: CometShuffleExchangeExec
+          if exchange.outputPartitioning.isInstanceOf[RoundRobinPartitioning] =>
+        exchange
+    }
+    assert(exchanges.size == 1, s"expected exactly one round robin exchange in\n$plan")
+    assert(exchanges.head.shuffleType == CometNativeShuffle, s"not native shuffle in\n$plan")
+    exchanges.head.shuffleDependency.rdd.outputDeterministicLevel
+  }
+
+  test("a planned positional round robin exchange carries the declaration to its stage") {
+    withSQLConf(
+      CometConf.COMET_EXEC_ENABLED.key -> "true",
+      CometConf.COMET_SHUFFLE_ENABLED.key -> "true",
+      CometConf.COMET_SHUFFLE_MODE.key -> "native",
+      CometConf.COMET_SHUFFLE_NATIVE_ROUND_ROBIN_PARTITIONING_ENABLED.key -> "true",
+      batchGranularKey -> "true",
+      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+      withParquetTable((0 until 100).map(i => (i % 10, i)), "tbl") {
+        // Straight over a scan. The scan replays identically, so positional assignment is
+        // reproducible and per-task retry stays as cheap as it is for hash placement.
+        assert(
+          roundRobinStageLevel(sql("SELECT * FROM tbl").repartition(4)) ==
+            DeterministicLevel.DETERMINATE)
+
+        // Below an aggregate's exchange. The reduce side sees shuffle blocks in arrival order, so
+        // a replay can frame the batches differently and the stage has to be rolled back whole.
+        assert(
+          roundRobinStageLevel(sql("SELECT _1, sum(_2) FROM tbl GROUP BY _1").repartition(4)) ==
+            DeterministicLevel.INDETERMINATE)
+      }
+    }
+  }
+}
diff --git a/spark/src/test/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleInputRDDSuite.scala b/spark/src/test/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleInputRDDSuite.scala
index 99d9fe93e0..92a06486e8 100644
--- a/spark/src/test/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleInputRDDSuite.scala
+++ b/spark/src/test/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleInputRDDSuite.scala
@@ -45,28 +45,56 @@ import org.apache.comet.serde.OperatorOuterClass.Operator
  */
 class CometNativeShuffleInputRDDSuite extends CometTestBase {
 
+  /** An input RDD that reports exactly `level`, standing in for a real upstream subtree. */
+  private def parentWithLevel(level: DeterministicLevel.Value): RDD[AnyRef] =
+    new RDD[AnyRef](spark.sparkContext, Nil) {
+      override protected def getOutputDeterministicLevel: DeterministicLevel.Value = level
+      override protected def getPartitions: Array[Partition] = Array.empty
+      override def compute(split: Partition, context: TaskContext): Iterator[AnyRef] =
+        Iterator.empty
+    }
+
+  private def inputWithParent(
+      level: DeterministicLevel.Value,
+      positionalRoundRobin: Boolean = false): CometNativeShuffleInputRDD =
+    new CometNativeShuffleInputRDD(
+      spark.sparkContext,
+      Seq(parentWithLevel(level)),
+      0,
+      Set.empty,
+      CometMetricNode(Map.empty),
+      positionalRoundRobin = positionalRoundRobin)
+
   test("native shuffle input preserves its parents' determinism") {
+    // Content-hash round robin, like every other partitioning, places rows by hash, so its
+    // output is a pure function of the rows and nothing here is indeterminate on its own account.
     Seq(
       DeterministicLevel.DETERMINATE,
       DeterministicLevel.UNORDERED,
       DeterministicLevel.INDETERMINATE).foreach { level =>
-      val parent = new RDD[AnyRef](spark.sparkContext, Nil) {
-        override protected def getOutputDeterministicLevel: DeterministicLevel.Value = level
-        override protected def getPartitions: Array[Partition] = Array.empty
-        override def compute(split: Partition, context: TaskContext): Iterator[AnyRef] =
-          Iterator.empty
-      }
-      val input = new CometNativeShuffleInputRDD(
-        spark.sparkContext,
-        Seq(parent),
-        0,
-        Set.empty,
-        CometMetricNode(Map.empty))
+      val input = inputWithParent(level)
       assert(input.outputDeterministicLevel == level)
       assert(input.copyForLocalShuffle().outputDeterministicLevel == level)
     }
   }
 
+  test("positional round robin declares indeterminate output unless its parent is determinate") {
+    // A determinate parent (a plain scan) replays identically, so positional assignment is
+    // reproducible and per-task retry stays cheap. Anything below another exchange is unordered,
+    // which is where Spark's own round robin flips to indeterminate too.
+    Seq(
+      DeterministicLevel.DETERMINATE -> DeterministicLevel.DETERMINATE,
+      DeterministicLevel.UNORDERED -> DeterministicLevel.INDETERMINATE,
+      DeterministicLevel.INDETERMINATE -> DeterministicLevel.INDETERMINATE).foreach {
+      case (parentLevel, expected) =>
+        val input = inputWithParent(parentLevel, positionalRoundRobin = true)
+        assert(input.outputDeterministicLevel == expected, s"parent was $parentLevel")
+        assert(
+          input.copyForLocalShuffle().outputDeterministicLevel == expected,
+          s"local fallback lost the declaration for parent $parentLevel")
+    }
+  }
+
   test("local shuffle input is an independent sibling with the same partition inputs") {
     val upstream = new RDD[AnyRef](spark.sparkContext, Nil) {
       override protected def getPartitions: Array[Partition] = Array.tabulate(2) { i =>