diff --git a/native/core/src/execution/jni_api.rs b/native/core/src/execution/jni_api.rs
index 65a2d68ec18..06aac12ac77 100644
--- a/native/core/src/execution/jni_api.rs
+++ b/native/core/src/execution/jni_api.rs
@@ -100,7 +100,7 @@ use tokio::sync::mpsc;
 use crate::execution::memory_pools::{create_memory_pool, parse_memory_pool_config};
 use crate::execution::operators::{ScanExec, ShuffleScanExec};
 use crate::execution::shuffle::{
-    decode_remote_shuffle_batch, read_ipc_compressed, CompressionCodec,
+    decode_remote_shuffle_batch, read_ipc_compressed, CompressionCodec, ShuffleWriterExec,
 };
 use crate::execution::spark_plan::SparkPlan;
 
@@ -1155,6 +1155,60 @@ fn get_execution_context<'a>(id: i64) -> &'a mut ExecutionContext {
     }
 }
 
+/// Returns the partition offsets published by a finished native shuffle write.
+///
+/// The returned array holds `num_output_partitions + 1` offsets, the last being the total data
+/// file length.
+#[no_mangle]
+pub extern "system" fn Java_org_apache_comet_Native_getShufflePartitionOffsets(
+    e: EnvUnowned,
+    _class: JClass,
+    exec_context: jlong,
+) -> jlongArray {
+    try_unwrap_or_throw(&e, |env| {
+        let context = get_execution_context(exec_context);
+
+        let root_op = context.root_op.as_ref().ok_or_else(|| {
+            CometError::Internal(
+                "Cannot read shuffle partition offsets before the plan has been executed"
+                    .to_string(),
+            )
+        })?;
+
+        // `ExecutionPlan` has `Any` as a supertrait but no `as_any` method of its own, so upcast
+        // the trait object before downcasting to the writer.
+        let writer = (root_op.native_plan.as_ref() as &dyn std::any::Any)
+            .downcast_ref::<ShuffleWriterExec>()
+            .ok_or_else(|| {
+                CometError::Internal(
+                    "Shuffle partition offsets are only available on a native shuffle write plan"
+                        .to_string(),
+                )
+            })?;
+
+        let offsets = writer
+            .partition_offsets()
+            .ok_or_else(|| {
+                CometError::Internal(
+                    "Shuffle partition offsets are not published by a remote shuffle destination"
+                        .to_string(),
+                )
+            })?
+            .get()
+            .ok_or_else(|| {
+                CometError::Internal(
+                    "Shuffle writer has not published its partition offsets; the plan was not \
+                     drained to completion"
+                        .to_string(),
+                )
+            })?;
+
+        let long_array = env.new_long_array(offsets.len())?;
+        long_array.set_region(env, 0, offsets)?;
+        Ok(long_array.into_raw())
+    })
+}
+
 /// Used by Comet shuffle external sorter to write sorted records to disk.
 /// # Safety
 /// This function is inherently unsafe since it deals with raw pointers passed from JNI.
diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs
index fe01bdcfeec..0c2938438da 100644
--- a/native/core/src/execution/planner.rs
+++ b/native/core/src/execution/planner.rs
@@ -47,7 +47,7 @@ use crate::execution::{
     planner::expression_registry::ExpressionRegistry,
     planner::operator_registry::OperatorRegistry,
     serde::{to_arrow_datatype, to_arrow_field},
-    shuffle::{SchemaAlignExec, ShuffleWriterDestination, ShuffleWriterExec},
+    shuffle::{PartitionOffsets, SchemaAlignExec, ShuffleWriterDestination, ShuffleWriterExec},
 };
 use crate::jvm_bridge::{jni_call, JVMClasses, ShufflePartitionPusher};
 use arrow::compute::CastOptions;
@@ -4143,7 +4143,7 @@ fn shuffle_writer_destination(
 
         return Ok(ShuffleWriterDestination::Local {
             output_data_file: writer.output_data_file.clone(),
-            output_index_file: writer.output_index_file.clone(),
+            partition_offsets: Arc::new(PartitionOffsets::default()),
         });
     };
 
@@ -4155,12 +4155,6 @@ fn shuffle_writer_destination(
                 ));
             }
 
-            if local.output_index_file.is_empty() {
-                return Err(GeneralError(
-                    "Local shuffle partition writer is missing its output index file".to_string(),
-                ));
-            }
-
             if !writer.output_data_file.is_empty()
                 && writer.output_data_file != local.output_data_file
             {
@@ -4171,16 +4165,6 @@ fn shuffle_writer_destination(
                 ));
             }
 
-            if !writer.output_index_file.is_empty()
-                && writer.output_index_file != local.output_index_file
-            {
-                return Err(GeneralError(
-                    "Local shuffle partition writer output index file conflicts with the legacy \
-                     shuffle output index file"
-                        .to_string(),
-                ));
-            }
-
             if shuffle_partition_pusher.is_some() {
                 return Err(GeneralError(
                     "Local shuffle partition writer cannot use a remote shuffle callback"
@@ -4190,11 +4174,11 @@ fn shuffle_writer_destination(
 
             Ok(ShuffleWriterDestination::Local {
                 output_data_file: local.output_data_file.clone(),
-                output_index_file: local.output_index_file.clone(),
+                partition_offsets: Arc::new(PartitionOffsets::default()),
             })
         }
         Some(spark_operator::partition_writer::Writer::Rss(_)) => {
-            if !writer.output_data_file.is_empty() || !writer.output_index_file.is_empty() {
+            if !writer.output_data_file.is_empty() {
                 return Err(GeneralError(
                     "RSS shuffle partition writer cannot have local output files".to_string(),
                 ));
@@ -5155,15 +5139,11 @@ mod tests {
         }
     }
 
-    fn local_shuffle_partition_writer(
-        output_data_file: &str,
-        output_index_file: &str,
-    ) -> spark_operator::PartitionWriter {
+    fn local_shuffle_partition_writer(output_data_file: &str) -> spark_operator::PartitionWriter {
         spark_operator::PartitionWriter {
             writer: Some(spark_operator::partition_writer::Writer::Local(
                 spark_operator::LocalPartitionWriter {
                     output_data_file: output_data_file.to_string(),
-                    output_index_file: output_index_file.to_string(),
                 },
             )),
         }
@@ -5180,15 +5160,15 @@ mod tests {
     fn assert_local_shuffle_destination(
         writer: &spark_operator::ShuffleWriter,
         expected_data_file: &str,
-        expected_index_file: &str,
     ) {
         match super::shuffle_writer_destination(writer, None).unwrap() {
             ShuffleWriterDestination::Local {
                 output_data_file,
-                output_index_file,
+                partition_offsets,
             } => {
                 assert_eq!(output_data_file, expected_data_file);
-                assert_eq!(output_index_file, expected_index_file);
+                // A fresh destination has not run a writer yet, so nothing is published.
+                assert!(partition_offsets.get().is_none());
             }
             destination => panic!("expected a local shuffle destination, got {destination:?}"),
         }
@@ -5198,49 +5178,38 @@ mod tests {
     fn shuffle_partition_writer_legacy_paths_remain_supported() {
         let writer = spark_operator::ShuffleWriter {
             output_data_file: "legacy.data".to_string(),
-            output_index_file: "legacy.index".to_string(),
             ..Default::default()
         };
 
-        assert_local_shuffle_destination(&writer, "legacy.data", "legacy.index");
+        assert_local_shuffle_destination(&writer, "legacy.data");
     }
 
     #[test]
     fn shuffle_partition_writer_uses_nested_local_paths() {
         let writer = spark_operator::ShuffleWriter {
-            partition_writer: Some(local_shuffle_partition_writer(
-                "shuffle.data",
-                "shuffle.index",
-            )),
+            partition_writer: Some(local_shuffle_partition_writer("shuffle.data")),
             ..Default::default()
         };
 
-        assert_local_shuffle_destination(&writer, "shuffle.data", "shuffle.index");
+        assert_local_shuffle_destination(&writer, "shuffle.data");
     }
 
     #[test]
     fn shuffle_partition_writer_accepts_matching_legacy_paths() {
         let writer = spark_operator::ShuffleWriter {
             output_data_file: "shuffle.data".to_string(),
-            output_index_file: "shuffle.index".to_string(),
-            partition_writer: Some(local_shuffle_partition_writer(
-                "shuffle.data",
-                "shuffle.index",
-            )),
+            partition_writer: Some(local_shuffle_partition_writer("shuffle.data")),
             ..Default::default()
         };
 
-        assert_local_shuffle_destination(&writer, "shuffle.data", "shuffle.index");
+        assert_local_shuffle_destination(&writer, "shuffle.data");
     }
 
     #[test]
     fn shuffle_partition_writer_rejects_conflicting_legacy_data_path() {
         let writer = spark_operator::ShuffleWriter {
             output_data_file: "legacy.data".to_string(),
-            partition_writer: Some(local_shuffle_partition_writer(
-                "shuffle.data",
-                "shuffle.index",
-            )),
+            partition_writer: Some(local_shuffle_partition_writer("shuffle.data")),
             ..Default::default()
         };
 
@@ -5251,29 +5220,11 @@ mod tests {
         );
     }
 
-    #[test]
-    fn shuffle_partition_writer_rejects_conflicting_legacy_index_path() {
-        let writer = spark_operator::ShuffleWriter {
-            output_index_file: "legacy.index".to_string(),
-            partition_writer: Some(local_shuffle_partition_writer(
-                "shuffle.data",
-                "shuffle.index",
-            )),
-            ..Default::default()
-        };
-
-        let error = super::shuffle_writer_destination(&writer, None).unwrap_err();
-        assert!(
-            error.to_string().contains("output index file conflicts"),
-            "unexpected error: {error}"
-        );
-    }
-
     #[test]
     fn shuffle_partition_writer_rejects_empty_local_data_path() {
         let writer = spark_operator::ShuffleWriter {
             output_data_file: "legacy.data".to_string(),
-            partition_writer: Some(local_shuffle_partition_writer("", "shuffle.index")),
+            partition_writer: Some(local_shuffle_partition_writer("")),
             ..Default::default()
         };
 
@@ -5284,21 +5235,6 @@ mod tests {
         );
     }
 
-    #[test]
-    fn shuffle_partition_writer_rejects_empty_local_index_path() {
-        let writer = spark_operator::ShuffleWriter {
-            output_index_file: "legacy.index".to_string(),
-            partition_writer: Some(local_shuffle_partition_writer("shuffle.data", "")),
-            ..Default::default()
-        };
-
-        let error = super::shuffle_writer_destination(&writer, None).unwrap_err();
-        assert!(
-            error.to_string().contains("missing its output index file"),
-            "unexpected error: {error}"
-        );
-    }
-
     #[test]
     fn shuffle_partition_writer_rejects_missing_destination() {
         let writer = spark_operator::ShuffleWriter {
@@ -5686,28 +5622,10 @@ mod tests {
         );
     }
 
-    #[test]
-    fn shuffle_partition_writer_rejects_rss_with_legacy_index_path() {
-        let writer = spark_operator::ShuffleWriter {
-            output_index_file: "legacy.index".to_string(),
-            partition_writer: Some(rss_shuffle_partition_writer()),
-            ..Default::default()
-        };
-        let callback: Arc<dyn ShufflePartitionPusher> =
-            Arc::new(RecordingShufflePartitionPusher::default());
-
-        let error = super::shuffle_writer_destination(&writer, Some(&callback)).unwrap_err();
-        assert!(
-            error.to_string().contains("cannot have local output files"),
-            "unexpected error: {error}"
-        );
-    }
-
     #[test]
     fn shuffle_partition_writer_rejects_callback_for_legacy_local_destination() {
         let writer = spark_operator::ShuffleWriter {
             output_data_file: "legacy.data".to_string(),
-            output_index_file: "legacy.index".to_string(),
             ..Default::default()
         };
         let callback: Arc<dyn ShufflePartitionPusher> =
@@ -5725,10 +5643,7 @@ mod tests {
     #[test]
     fn shuffle_partition_writer_rejects_callback_for_explicit_local_destination() {
         let writer = spark_operator::ShuffleWriter {
-            partition_writer: Some(local_shuffle_partition_writer(
-                "shuffle.data",
-                "shuffle.index",
-            )),
+            partition_writer: Some(local_shuffle_partition_writer("shuffle.data")),
             ..Default::default()
         };
         let callback: Arc<dyn ShufflePartitionPusher> =
diff --git a/native/proto/src/lib.rs b/native/proto/src/lib.rs
index c814760c4b1..dd5ccfbe05b 100644
--- a/native/proto/src/lib.rs
+++ b/native/proto/src/lib.rs
@@ -62,21 +62,16 @@ mod tests {
     struct LegacyShuffleWriter {
         #[prost(string, tag = "3")]
         output_data_file: String,
-        #[prost(string, tag = "4")]
-        output_index_file: String,
     }
 
     fn local_shuffle_writer() -> ShuffleWriter {
         let output_data_file = "/tmp/shuffle.data".to_string();
-        let output_index_file = "/tmp/shuffle.index".to_string();
 
         ShuffleWriter {
             output_data_file: output_data_file.clone(),
-            output_index_file: output_index_file.clone(),
             partition_writer: Some(PartitionWriter {
                 writer: Some(partition_writer::Writer::Local(LocalPartitionWriter {
                     output_data_file,
-                    output_index_file,
                 })),
             }),
             ..Default::default()
@@ -89,14 +84,12 @@ mod tests {
         let decoded = ShuffleWriter::decode(encoded.as_slice()).unwrap();
 
         assert_eq!(decoded.output_data_file, "/tmp/shuffle.data");
-        assert_eq!(decoded.output_index_file, "/tmp/shuffle.index");
         let Some(partition_writer::Writer::Local(local)) =
             decoded.partition_writer.and_then(|writer| writer.writer)
         else {
             panic!("expected a local shuffle partition writer");
         };
         assert_eq!(local.output_data_file, "/tmp/shuffle.data");
-        assert_eq!(local.output_index_file, "/tmp/shuffle.index");
     }
 
     #[test]
@@ -121,19 +114,16 @@ mod tests {
         let decoded = LegacyShuffleWriter::decode(encoded.as_slice()).unwrap();
 
         assert_eq!(decoded.output_data_file, "/tmp/shuffle.data");
-        assert_eq!(decoded.output_index_file, "/tmp/shuffle.index");
     }
 
     #[test]
     fn new_shuffle_writer_decodes_legacy_plan_without_destination() {
         let legacy = LegacyShuffleWriter {
             output_data_file: "/tmp/legacy.data".to_string(),
-            output_index_file: "/tmp/legacy.index".to_string(),
         };
         let decoded = ShuffleWriter::decode(legacy.encode_to_vec().as_slice()).unwrap();
 
         assert_eq!(decoded.output_data_file, "/tmp/legacy.data");
-        assert_eq!(decoded.output_index_file, "/tmp/legacy.index");
         assert!(decoded.partition_writer.is_none());
     }
 }
diff --git a/native/proto/src/proto/operator.proto b/native/proto/src/proto/operator.proto
index 75a6e06d998..66db291b388 100644
--- a/native/proto/src/proto/operator.proto
+++ b/native/proto/src/proto/operator.proto
@@ -806,10 +806,9 @@ message PartitionWriter {
   }
 }
 
-// Local shuffle output consists of a data file and its partition-offset index.
+// Local shuffle output consists of a data file. The partition offsets are returned to the JVM via JNI
 message LocalPartitionWriter {
   string output_data_file = 1;
-  string output_index_file = 2;
 }
 
 // Marker for remote shuffle output. The task-owned callback is bound outside
@@ -819,26 +818,25 @@ message RssPartitionWriter {}
 message ShuffleWriter {
   spark.spark_partitioning.Partitioning partitioning = 1;
   // Retained for compatibility with native binaries that predate partition_writer.
-  // Local plans also carry these paths in partition_writer.local.
+  // Local plans also carry this path in partition_writer.local.
   string output_data_file = 3;
-  string output_index_file = 4;
-  CompressionCodec codec = 5;
-  int32 compression_level = 6;
-  bool tracing_enabled = 7;
+  CompressionCodec codec = 4;
+  int32 compression_level = 5;
+  bool tracing_enabled = 6;
   // Size of the write buffer in bytes used when writing shuffle data to disk.
   // Larger values may improve write performance but use more memory.
-  int32 write_buffer_size = 8;
+  int32 write_buffer_size = 7;
   // Spark-declared output schema of the writer's child. When the child is an inlined native
   // subtree, the native planner casts the child's actual output to this schema before
   // serializing to shuffle blocks, since there is no FFI boundary or ScanExec between them
   // to absorb DataFusion-vs-Spark type drift. Empty when the child is a placeholder Scan;
   // that path already has a cast point upstream.
-  repeated SparkStructField expected_output_schema = 9;
+  repeated SparkStructField expected_output_schema = 8;
   // Maximum number of bytes that the writer buffers in memory before spilling to disk.
   // Zero means no limit, in which case spilling is driven only by memory pool pressure.
-  uint64 max_buffer_bytes = 10;
+  uint64 max_buffer_bytes = 9;
   // Explicit output destination. When absent, use the legacy output file fields.
-  PartitionWriter partition_writer = 11;
+  PartitionWriter partition_writer = 10;
 }
 
 message ParquetWriter {
diff --git a/native/shuffle/benches/shuffle_writer.rs b/native/shuffle/benches/shuffle_writer.rs
index c9c088e2802..1e1e68565e6 100644
--- a/native/shuffle/benches/shuffle_writer.rs
+++ b/native/shuffle/benches/shuffle_writer.rs
@@ -221,7 +221,6 @@ fn create_shuffle_writer_exec(
         partitioning,
         compression_codec,
         "/tmp/data.out".to_string(),
-        "/tmp/index.out".to_string(),
         false,
         1024 * 1024,
         None,
diff --git a/native/shuffle/src/bin/shuffle_bench.rs b/native/shuffle/src/bin/shuffle_bench.rs
index cd43d41dbc1..7630d39dd4e 100644
--- a/native/shuffle/src/bin/shuffle_bench.rs
+++ b/native/shuffle/src/bin/shuffle_bench.rs
@@ -126,7 +126,6 @@ fn main() {
     // Create output directory
     fs::create_dir_all(&args.output_dir).expect("Failed to create output directory");
     let data_file = args.output_dir.join("data.out");
-    let index_file = args.output_dir.join("index.out");
 
     let (schema, total_rows) = read_parquet_metadata(&args.input, args.limit);
 
@@ -189,7 +188,6 @@ fn main() {
                 &hash_col_indices,
                 &args,
                 data_file.to_str().unwrap(),
-                index_file.to_str().unwrap(),
             )
         };
         let data_size = fs::metadata(&data_file).map(|m| m.len()).unwrap_or(0);
@@ -253,7 +251,6 @@ fn main() {
     }
 
     let _ = fs::remove_file(&data_file);
-    let _ = fs::remove_file(&index_file);
 }
 
 fn print_shuffle_metrics(metrics: &MetricsSet, total_wall_time_secs: f64) {
@@ -398,7 +395,6 @@ fn run_shuffle_write(
     hash_col_indices: &[usize],
     args: &Args,
     data_file: &str,
-    index_file: &str,
 ) -> (f64, Option<MetricsSet>, Option<MetricsSet>) {
     let partitioning = build_partitioning(
         &args.partitioning,
@@ -420,7 +416,6 @@ fn run_shuffle_write(
             args.max_buffer_bytes,
             args.limit,
             data_file.to_string(),
-            index_file.to_string(),
         )
         .await
         .unwrap();
@@ -444,7 +439,6 @@ async fn execute_shuffle_write(
     max_buffer_bytes: Option<usize>,
     limit: usize,
     data_file: String,
-    index_file: String,
 ) -> datafusion::common::Result<(MetricsSet, MetricsSet)> {
     let config = SessionConfig::new().with_batch_size(batch_size);
     let mut runtime_builder = RuntimeEnvBuilder::new();
@@ -483,7 +477,6 @@ async fn execute_shuffle_write(
         partitioning,
         codec,
         data_file,
-        index_file,
         false,
         write_buffer_size,
         max_buffer_bytes,
@@ -537,7 +530,6 @@ fn run_concurrent_shuffle_writes(
             let task_dir = args.output_dir.join(format!("task_{task_id}"));
             fs::create_dir_all(&task_dir).expect("Failed to create task output directory");
             let data_file = task_dir.join("data.out").to_str().unwrap().to_string();
-            let index_file = task_dir.join("index.out").to_str().unwrap().to_string();
 
             let input_str = input_path.to_str().unwrap().to_string();
             let codec = codec.clone();
@@ -564,7 +556,6 @@ fn run_concurrent_shuffle_writes(
                     max_buffer_bytes,
                     limit,
                     data_file,
-                    index_file,
                 )
                 .await
                 .unwrap()
diff --git a/native/shuffle/src/lib.rs b/native/shuffle/src/lib.rs
index 766634eb71e..d60d9b6eab9 100644
--- a/native/shuffle/src/lib.rs
+++ b/native/shuffle/src/lib.rs
@@ -34,5 +34,5 @@ pub use comet_partitioning::CometPartitioning;
 pub use ipc::{read_ipc_compressed, read_ipc_compressed_validated};
 pub use remote_schema::{decode_remote_shuffle_batch, validate_remote_schema};
 pub use schema_align::SchemaAlignExec;
-pub use shuffle_writer::{ShuffleWriterDestination, ShuffleWriterExec};
+pub use shuffle_writer::{PartitionOffsets, ShuffleWriterDestination, ShuffleWriterExec};
 pub use writers::{CompressionCodec, ShuffleBlockWriter};
diff --git a/native/shuffle/src/rss_execution_tests.rs b/native/shuffle/src/rss_execution_tests.rs
index 491562c5732..e9269edc9e5 100644
--- a/native/shuffle/src/rss_execution_tests.rs
+++ b/native/shuffle/src/rss_execution_tests.rs
@@ -16,8 +16,8 @@
 // under the License.
 
 use crate::{
-    read_ipc_compressed, CometPartitioning, CompressionCodec, ShuffleWriterDestination,
-    ShuffleWriterExec,
+    read_ipc_compressed, CometPartitioning, CompressionCodec, PartitionOffsets,
+    ShuffleWriterDestination, ShuffleWriterExec,
 };
 use arrow::array::{Array, Int32Array, RecordBatch, RecordBatchOptions};
 use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
@@ -422,18 +422,18 @@ fn rss_callback_survives_execution_plan_child_replacement() {
 }
 
 #[test]
-fn explicit_local_destination_preserves_data_and_index_files() {
+fn explicit_local_destination_writes_data_and_publishes_offsets() {
     let batch = int_batch(0, 8);
     let directory = tempfile::tempdir().unwrap();
     let data_file = directory.path().join("shuffle.data");
-    let index_file = directory.path().join("shuffle.index");
+    let offsets = Arc::new(PartitionOffsets::default());
     let execution = ShuffleWriterExec::try_new_with_destination(
         memory_input(vec![batch.clone()], batch.schema()),
         CometPartitioning::SinglePartition,
         CompressionCodec::None,
         ShuffleWriterDestination::Local {
             output_data_file: data_file.to_str().unwrap().to_string(),
-            output_index_file: index_file.to_str().unwrap().to_string(),
+            partition_offsets: Arc::clone(&offsets),
         },
         false,
         1024 * 1024,
@@ -442,7 +442,14 @@ fn explicit_local_destination_preserves_data_and_index_files() {
     .unwrap();
 
     run_execution(&execution).unwrap();
-    let frame = std::fs::read(data_file).unwrap();
+    let frame = std::fs::read(&data_file).unwrap();
     assert_eq!(decode_frame(&frame).num_rows(), 8);
-    assert_eq!(std::fs::read(index_file).unwrap().len(), 16);
+    // A single-partition writer publishes two offsets, the partition start and the total
+    // length, in memory rather than through an index file.
+    let published = offsets
+        .get()
+        .expect("writer published its partition offsets");
+    assert_eq!(published.len(), 2);
+    assert_eq!(published[0], 0);
+    assert_eq!(published[1] as usize, frame.len());
 }
diff --git a/native/shuffle/src/shuffle_writer.rs b/native/shuffle/src/shuffle_writer.rs
index 8d668de7725..b1930f7d79a 100644
--- a/native/shuffle/src/shuffle_writer.rs
+++ b/native/shuffle/src/shuffle_writer.rs
@@ -45,18 +45,39 @@ use futures::{StreamExt, TryStreamExt};
 use std::{
     fmt,
     fmt::{Debug, Formatter},
-    sync::Arc,
+    sync::{Arc, OnceLock},
 };
 
+/// One-shot slot carrying a local shuffle task's partition offsets out of the writer.
+#[derive(Debug, Default)]
+pub struct PartitionOffsets(OnceLock<Vec<i64>>);
+
+impl PartitionOffsets {
+    /// Publishes the finished task's offsets. Errors if called more than once.
+    pub fn set(&self, offsets: Vec<i64>) -> Result<()> {
+        self.0.set(offsets).map_err(|_| {
+            DataFusionError::Execution(
+                "shuffle write error: partition offsets were already published".to_string(),
+            )
+        })
+    }
+
+    /// The finished task's offsets, or `None` if the writer has not completed.
+    pub fn get(&self) -> Option<&[i64]> {
+        self.0.get().map(Vec::as_slice)
+    }
+}
+
 /// Storage destination for a native shuffle writer.
 #[derive(Clone)]
 pub enum ShuffleWriterDestination {
-    /// Writes partition data and offsets to local shuffle files.
+    /// Writes partition data to a local shuffle file and publishes the partition offsets in
+    /// memory.
     Local {
         /// Path of the local shuffle data file.
         output_data_file: String,
-        /// Path of the local shuffle index file.
-        output_index_file: String,
+        /// One offset per partition written, plus a trailing total.
+        partition_offsets: Arc<PartitionOffsets>,
     },
     /// Pushes complete encoded partition blocks to a task-owned callback.
     Rss {
@@ -72,11 +93,11 @@ impl Debug for ShuffleWriterDestination {
         match self {
             Self::Local {
                 output_data_file,
-                output_index_file,
+                partition_offsets,
             } => f
                 .debug_struct("Local")
                 .field("output_data_file", output_data_file)
-                .field("output_index_file", output_index_file)
+                .field("partition_offsets", &partition_offsets.get().is_some())
                 .finish(),
             Self::Rss { max_frame_size, .. } => f
                 .debug_struct("Rss")
@@ -110,14 +131,14 @@ pub struct ShuffleWriterExec {
 }
 
 impl ShuffleWriterExec {
-    /// Creates a shuffle writer that writes to local data and index files.
+    /// Creates a shuffle writer that writes partition data to a local file and exposes its
+    /// partition offsets.
     #[allow(clippy::too_many_arguments)]
     pub fn try_new(
         input: Arc<dyn ExecutionPlan>,
         partitioning: CometPartitioning,
         codec: CompressionCodec,
         output_data_file: String,
-        output_index_file: String,
         tracing_enabled: bool,
         write_buffer_size: usize,
         max_buffer_bytes: Option<usize>,
@@ -128,7 +149,7 @@ impl ShuffleWriterExec {
             codec,
             ShuffleWriterDestination::Local {
                 output_data_file,
-                output_index_file,
+                partition_offsets: Arc::new(PartitionOffsets::default()),
             },
             tracing_enabled,
             write_buffer_size,
@@ -136,6 +157,17 @@ impl ShuffleWriterExec {
         )
     }
 
+    /// Returns this task's partition offsets, for a local destination. `None` for a remote
+    /// destination, where the pusher reports partition lengths instead.
+    pub fn partition_offsets(&self) -> Option<&Arc<PartitionOffsets>> {
+        match &self.destination {
+            ShuffleWriterDestination::Local {
+                partition_offsets, ..
+            } => Some(partition_offsets),
+            ShuffleWriterDestination::Rss { .. } => None,
+        }
+    }
+
     /// Creates a shuffle writer for a local or task-owned remote destination.
     pub fn try_new_with_destination(
         input: Arc<dyn ExecutionPlan>,
@@ -286,12 +318,12 @@ async fn external_shuffle(
     let mut repartitioner = match destination {
         ShuffleWriterDestination::Local {
             output_data_file,
-            output_index_file,
+            partition_offsets,
         } => {
             let shuffle_block_writer = ShuffleBlockWriter::try_new(schema.as_ref(), codec.clone())?;
             let writer = LocalPartitionWriter::try_new(
                 output_data_file,
-                output_index_file,
+                partition_offsets,
                 shuffle_block_writer,
                 partitioning.partition_count(),
                 context.session_config().batch_size(),
@@ -550,7 +582,7 @@ mod test {
                 .unwrap();
         let local_partition_writer = LocalPartitionWriter::try_new(
             "/tmp/data.out".to_string(),
-            "/tmp/index.out".to_string(),
+            Arc::new(PartitionOffsets::default()),
             shuffle_block_writer,
             num_partitions,
             1024,
@@ -572,21 +604,19 @@ mod test {
 
         repartitioner.insert_batch(batch.clone()).await.unwrap();
 
-        {
-            let spill_writers = repartitioner.partition_writer().get_spill_writers();
-            assert_eq!(spill_writers.len(), 2);
-
-            assert!(!spill_writers[0].has_spill_file());
-            assert!(!spill_writers[1].has_spill_file());
-        }
+        assert!(!repartitioner
+            .partition_writer()
+            .get_spill()
+            .has_spill_file());
 
         repartitioner.spill(0).unwrap();
 
-        // after spill, there should be spill files
+        // after spill, both partitions' blocks are in the one spill file
         {
-            let spill_writers = repartitioner.partition_writer().get_spill_writers();
-            assert!(spill_writers[0].has_spill_file());
-            assert!(spill_writers[1].has_spill_file());
+            let spill = repartitioner.partition_writer().get_spill();
+            assert!(spill.has_spill_file());
+            assert!(!spill.ranges(0).unwrap().is_empty());
+            assert!(!spill.ranges(1).unwrap().is_empty());
         }
 
         // insert another batch after spilling
@@ -622,7 +652,7 @@ mod test {
             ShuffleBlockWriter::try_new(schema.as_ref(), CompressionCodec::Lz4Frame).unwrap();
         let local_partition_writer = LocalPartitionWriter::try_new(
             dir.path().join("data.out").to_str().unwrap().to_string(),
-            dir.path().join("index.out").to_str().unwrap().to_string(),
+            Arc::new(PartitionOffsets::default()),
             shuffle_block_writer,
             num_partitions,
             1024,
@@ -684,7 +714,7 @@ mod test {
             ShuffleBlockWriter::try_new(schema.as_ref(), CompressionCodec::Lz4Frame).unwrap();
         let local_partition_writer = LocalPartitionWriter::try_new(
             dir.path().join("data.out").to_str().unwrap().to_string(),
-            dir.path().join("index.out").to_str().unwrap().to_string(),
+            Arc::new(PartitionOffsets::default()),
             shuffle_block_writer,
             num_partitions,
             batch_size,
@@ -827,7 +857,7 @@ mod test {
                 .unwrap();
         let local_partition_writer = LocalPartitionWriter::try_new(
             dir.path().join("data.out").to_str().unwrap().to_string(),
-            dir.path().join("index.out").to_str().unwrap().to_string(),
+            Arc::new(PartitionOffsets::default()),
             shuffle_block_writer,
             num_partitions,
             1024,
@@ -852,13 +882,14 @@ mod test {
         }
         repartitioner.shuffle_write().unwrap();
 
-        let actual_spilled_bytes: usize = repartitioner
+        let actual_spilled_bytes = repartitioner
             .partition_writer()
-            .get_spill_writers()
-            .iter()
-            .filter_map(|writer| writer.path().unwrap())
-            .map(|path| usize::try_from(std::fs::metadata(path).unwrap().len()).unwrap())
-            .sum();
+            .get_spill()
+            .path()
+            .unwrap()
+            .map_or(0, |path| {
+                usize::try_from(std::fs::metadata(path).unwrap().len()).unwrap()
+            });
         assert_eq!(
             spilled_bytes.value(),
             actual_spilled_bytes,
@@ -935,7 +966,6 @@ mod test {
         let batch = create_batch(1000);
         let batches = (0..20).map(|_| batch.clone()).collect::<Vec<_>>();
         let data_file = dir.join(format!("{tag}_data.out"));
-        let index_file = dir.join(format!("{tag}_index.out"));
 
         let exec = ShuffleWriterExec::try_new(
             Arc::new(DataSourceExec::new(Arc::new(
@@ -945,7 +975,6 @@ mod test {
             CometPartitioning::Hash(vec![Arc::new(Column::new("a", 0))], 16),
             CompressionCodec::Zstd(1),
             data_file.to_str().unwrap().to_string(),
-            index_file.to_str().unwrap().to_string(),
             false,
             1024 * 1024,
             max_buffer_bytes,
@@ -1070,7 +1099,6 @@ mod test {
                 partitioning,
                 CompressionCodec::Zstd(1),
                 "/tmp/data.out".to_string(),
-                "/tmp/index.out".to_string(),
                 false,
                 1024 * 1024, // write_buffer_size: 1MB default
                 None,
@@ -1118,9 +1146,9 @@ mod test {
         let batches = (0..num_batches).map(|_| batch.clone()).collect::<Vec<_>>();
 
         // Run shuffle twice and compare results
+        let mut offsets_per_run: Vec<Vec<i64>> = Vec::new();
         for run in 0..2 {
             let data_file = format!("/tmp/rr_data_{}.out", run);
-            let index_file = format!("/tmp/rr_index_{}.out", run);
 
             let partitions = std::slice::from_ref(&batches);
             let exec = ShuffleWriterExec::try_new(
@@ -1130,7 +1158,6 @@ mod test {
                 CometPartitioning::RoundRobin(num_partitions, 0),
                 CompressionCodec::Zstd(1),
                 data_file.clone(),
-                index_file.clone(),
                 false,
                 1024 * 1024,
                 None,
@@ -1153,6 +1180,14 @@ mod test {
                 while stream.next().await.is_some() {}
             });
 
+            offsets_per_run.push(
+                exec.partition_offsets()
+                    .expect("local destination publishes offsets")
+                    .get()
+                    .expect("writer published its partition offsets")
+                    .to_vec(),
+            );
+
             if run == 1 {
                 // Compare data files
                 let mut data0 = Vec::new();
@@ -1170,20 +1205,10 @@ mod test {
                     "Round robin shuffle data should be identical across runs"
                 );
 
-                // Compare index files
-                let mut index0 = Vec::new();
-                fs::File::open("/tmp/rr_index_0.out")
-                    .unwrap()
-                    .read_to_end(&mut index0)
-                    .unwrap();
-                let mut index1 = Vec::new();
-                fs::File::open("/tmp/rr_index_1.out")
-                    .unwrap()
-                    .read_to_end(&mut index1)
-                    .unwrap();
+                // Compare the published partition offsets
                 assert_eq!(
-                    index0, index1,
-                    "Round robin shuffle index should be identical across runs"
+                    offsets_per_run[0], offsets_per_run[1],
+                    "Round robin shuffle partition offsets should be identical across runs"
                 );
             }
         }
@@ -1431,13 +1456,11 @@ mod test {
 
         let dir = tempfile::tempdir().unwrap();
         let data_file = dir.path().join("data.out").to_str().unwrap().to_string();
-        let index_file = dir.path().join("index.out").to_str().unwrap().to_string();
-
         let block_writer =
             ShuffleBlockWriter::try_new(schema.as_ref(), CompressionCodec::Lz4Frame).unwrap();
         let writer = LocalPartitionWriter::try_new(
             data_file.clone(),
-            index_file,
+            Arc::new(PartitionOffsets::default()),
             block_writer,
             1, // single partition
             batch_size,
@@ -1511,7 +1534,6 @@ mod test {
 
         let dir = tempfile::tempdir().unwrap();
         let data_file = dir.path().join("data.out");
-        let index_file = dir.path().join("index.out");
 
         let exec = ShuffleWriterExec::try_new(
             Arc::new(DataSourceExec::new(Arc::new(
@@ -1520,7 +1542,6 @@ mod test {
             CometPartitioning::RoundRobin(num_partitions, 0),
             CompressionCodec::Zstd(1),
             data_file.to_str().unwrap().to_string(),
-            index_file.to_str().unwrap().to_string(),
             false,
             1024 * 1024,
             None,
@@ -1551,29 +1572,26 @@ mod test {
             "Row count should survive roundtrip"
         );
 
-        // Verify index file structure: num_partitions + 1 offsets
-        let mut index_data = Vec::new();
-        fs::File::open(&index_file)
-            .unwrap()
-            .read_to_end(&mut index_data)
-            .unwrap();
-        let expected_index_size = (num_partitions + 1) * 8;
-        assert_eq!(index_data.len(), expected_index_size);
+        // Verify the published offsets: num_partitions + 1 of them
+        let offsets = exec
+            .partition_offsets()
+            .expect("local destination publishes offsets")
+            .get()
+            .expect("writer published its partition offsets")
+            .to_vec();
+        assert_eq!(offsets.len(), num_partitions + 1);
 
         // First offset should be 0
-        let first_offset = i64::from_le_bytes(index_data[0..8].try_into().unwrap());
-        assert_eq!(first_offset, 0);
+        assert_eq!(offsets[0], 0);
 
         // Second offset should equal data file length (partition 0 holds all data)
         let data_len = data.len() as i64;
-        let second_offset = i64::from_le_bytes(index_data[8..16].try_into().unwrap());
-        assert_eq!(second_offset, data_len);
+        assert_eq!(offsets[1], data_len);
 
         // All remaining offsets should equal data file length (empty partitions)
-        for i in 2..=num_partitions {
-            let offset = i64::from_le_bytes(index_data[i * 8..(i + 1) * 8].try_into().unwrap());
+        for (i, offset) in offsets.iter().enumerate().skip(2) {
             assert_eq!(
-                offset, data_len,
+                *offset, data_len,
                 "Partition {i} offset should equal data length"
             );
         }
@@ -1600,7 +1618,6 @@ mod test {
 
         let dir = tempfile::tempdir().unwrap();
         let data_file = dir.path().join("data.out");
-        let index_file = dir.path().join("index.out");
 
         let exec = ShuffleWriterExec::try_new(
             Arc::new(DataSourceExec::new(Arc::new(
@@ -1609,7 +1626,6 @@ mod test {
             CometPartitioning::RoundRobin(num_partitions, 0),
             CompressionCodec::Zstd(1),
             data_file.to_str().unwrap().to_string(),
-            index_file.to_str().unwrap().to_string(),
             false,
             1024 * 1024,
             None,
@@ -1632,17 +1648,16 @@ mod test {
             .unwrap();
         assert!(data.is_empty(), "Data file should be empty with zero rows");
 
-        // Index file should have all-zero offsets
-        let mut index_data = Vec::new();
-        fs::File::open(&index_file)
-            .unwrap()
-            .read_to_end(&mut index_data)
-            .unwrap();
-        let expected_index_size = (num_partitions + 1) * 8;
-        assert_eq!(index_data.len(), expected_index_size);
-        for i in 0..=num_partitions {
-            let offset = i64::from_le_bytes(index_data[i * 8..(i + 1) * 8].try_into().unwrap());
-            assert_eq!(offset, 0, "All offsets should be 0 with zero rows");
+        // partition offsets should be all zero
+        let offsets = exec
+            .partition_offsets()
+            .expect("local destination publishes offsets")
+            .get()
+            .expect("writer published its partition offsets")
+            .to_vec();
+        assert_eq!(offsets.len(), num_partitions + 1);
+        for offset in &offsets {
+            assert_eq!(*offset, 0, "All offsets should be 0 with zero rows");
         }
     }
 }
diff --git a/native/shuffle/src/writers/local/local_partition_writer.rs b/native/shuffle/src/writers/local/local_partition_writer.rs
index e22e339f949..cf8ce6f5574 100644
--- a/native/shuffle/src/writers/local/local_partition_writer.rs
+++ b/native/shuffle/src/writers/local/local_partition_writer.rs
@@ -16,15 +16,17 @@
 // under the License.
 
 use crate::metrics::ShufflePartitionerMetrics;
-use crate::writers::local::spill::SpillWriter;
+use crate::writers::local::spill::PartitionedSpill;
 use crate::writers::partition_writer::PartitionWriter;
 use crate::writers::BufBatchWriter;
-use crate::ShuffleBlockWriter;
+use crate::{PartitionOffsets, ShuffleBlockWriter};
 use arrow::array::RecordBatch;
 use datafusion::common::DataFusionError;
 use datafusion::execution::runtime_env::RuntimeEnv;
 use std::fs::{File, OpenOptions};
-use std::io::{BufWriter, Seek, Write};
+use std::io::{BufWriter, ErrorKind, Read, Seek, SeekFrom, Write};
+use std::ops::Range;
+use std::os::unix::fs::FileExt;
 use std::sync::Arc;
 
 /// Output target for the shuffle data file.
@@ -54,10 +56,12 @@ enum DataOutput {
     Multi {
         output_writer: BufWriter<File>,
         shuffle_block_writer: ShuffleBlockWriter,
-        /// One spill file per output partition, buffered until `finish_partition`
-        /// merges them into the shuffle output.
-        spill_writers: Vec<SpillWriter>,
-        /// Runtime used to allocate the temporary spill files.
+        /// Spilled blocks for every partition, in one file.
+        spill: PartitionedSpill,
+        /// Read handle on the spill file and a write-buffer-sized scratch for its ranges,
+        /// opened on first use.
+        spill_reader: Option<(File, Vec<u8>)>,
+        /// Runtime used to allocate the temporary spill file.
         runtime: Arc<RuntimeEnv>,
         /// Byte buffer recycled through the short-lived per-partition `BufBatchWriter`s.
         /// Partitions are written strictly one at a time, so a single buffer keeps its
@@ -73,7 +77,7 @@ enum DataOutput {
 /// byte offset where each partition begins. See [`DataOutput`] for how the
 /// single- and multi-partition modes differ.
 pub(crate) struct LocalPartitionWriter {
-    output_index_file: String,
+    partition_offsets: Arc<PartitionOffsets>,
     data_output: DataOutput,
     /// Start offset of each partition in the data file, plus a trailing entry
     /// with the total length so partition sizes are simple offset differences.
@@ -90,7 +94,7 @@ pub(crate) struct LocalPartitionWriter {
 impl LocalPartitionWriter {
     pub(crate) fn try_new(
         output_data_file: String,
-        output_index_file: String,
+        partition_offsets: Arc<PartitionOffsets>,
         shuffle_block_writer: ShuffleBlockWriter,
         num_output_partitions: usize,
         batch_size: usize,
@@ -116,25 +120,23 @@ impl LocalPartitionWriter {
             }
         } else {
             let output_writer = BufWriter::with_capacity(write_buffer_size, output_file);
-            let spill_writers = (0..num_output_partitions)
-                .map(|_| {
-                    SpillWriter::try_new(
-                        shuffle_block_writer.clone(),
-                        write_buffer_size,
-                        batch_size,
-                    )
-                })
-                .collect::<datafusion::common::Result<Vec<_>>>()?;
+            let spill = PartitionedSpill::new(
+                shuffle_block_writer.clone(),
+                write_buffer_size,
+                batch_size,
+                num_output_partitions,
+            );
             DataOutput::Multi {
                 output_writer,
                 shuffle_block_writer,
-                spill_writers,
+                spill,
+                spill_reader: None,
                 runtime,
                 recycled_buffer: Vec::new(),
             }
         };
         Ok(Self {
-            output_index_file,
+            partition_offsets,
             data_output,
             offsets: vec![0u64; num_output_partitions + 1],
             batch_size,
@@ -145,10 +147,10 @@ impl LocalPartitionWriter {
     }
 
     #[cfg(test)]
-    pub(crate) fn get_spill_writers(&self) -> &Vec<SpillWriter> {
+    pub(crate) fn get_spill(&self) -> &PartitionedSpill {
         match &self.data_output {
-            DataOutput::Multi { spill_writers, .. } => spill_writers,
-            DataOutput::Single { .. } => panic!("single-partition output has no spill writers"),
+            DataOutput::Multi { spill, .. } => spill,
+            DataOutput::Single { .. } => panic!("single-partition output does not spill"),
         }
     }
 }
@@ -183,15 +185,12 @@ impl PartitionWriter for LocalPartitionWriter {
                 }
             }
             DataOutput::Multi {
-                spill_writers,
+                spill,
                 runtime,
                 recycled_buffer,
                 ..
             } => {
-                // Multi-partition output buffers each partition's batches into its own
-                // spill file. `finish_partition` later merges the spill files (and any
-                // remaining in-memory batches) into the shuffle output in partition order.
-                spill_writers[pid].write(iter, runtime, metrics, recycled_buffer)?;
+                spill.write(pid, iter, runtime, metrics, recycled_buffer)?;
             }
         }
 
@@ -231,23 +230,32 @@ impl PartitionWriter for LocalPartitionWriter {
             DataOutput::Multi {
                 output_writer,
                 shuffle_block_writer,
-                spill_writers,
+                spill,
+                spill_reader,
                 recycled_buffer,
                 ..
             } => {
                 self.offsets[pid] = output_writer.stream_position()?;
 
-                // if we wrote a spill file for this partition then copy the
-                // contents into the shuffle file
-                if let Some(writer) = spill_writers.get(pid) {
-                    if let Some(spill_path) = writer.path()? {
-                        // Use raw File handle (not BufReader) so that std::io::copy
-                        // can use copy_file_range/sendfile for zero-copy on Linux.
-                        let mut spill_file = File::open(spill_path)?;
-                        let mut write_timer = metrics.write_time.timer();
-                        std::io::copy(&mut spill_file, output_writer)?;
-                        write_timer.stop();
+                let mut flush_timer = metrics.write_time.timer();
+                spill.flush()?;
+                flush_timer.stop();
+                let ranges = spill.ranges(pid)?;
+                if !ranges.is_empty() {
+                    if spill_reader.is_none() {
+                        let path = spill.path()?.ok_or_else(|| {
+                            DataFusionError::Internal(
+                                "shuffle spill ranges recorded without a spill file".to_string(),
+                            )
+                        })?;
+                        *spill_reader = Some((File::open(path)?, vec![0; write_buffer_size]));
                     }
+                    let (spill_file, buffer) = spill_reader.as_mut().unwrap();
+                    let mut write_timer = metrics.write_time.timer();
+                    for range in ranges {
+                        copy_spill_range(spill_file, buffer, range, output_writer)?;
+                    }
+                    write_timer.stop();
                 }
 
                 // Write in memory batches to output data file. Each partition uses its
@@ -306,34 +314,72 @@ impl PartitionWriter for LocalPartitionWriter {
         // add one extra offset at last to ease partition length computation
         self.offsets[self.num_output_partitions] = final_offset;
 
-        let mut write_timer = metrics.write_time.timer();
-        let mut output_index = BufWriter::new(
-            File::create(self.output_index_file.clone())
-                .map_err(|e| DataFusionError::Execution(format!("shuffle write error: {e:?}")))?,
-        );
-
-        for offset in &self.offsets {
-            let offset_i64 = i64::try_from(*offset).map_err(|_| {
-                DataFusionError::Execution(format!(
-                    "shuffle write error: offset overflow ({offset})"
-                ))
-            })?;
-            output_index.write_all(&offset_i64.to_le_bytes())?;
-        }
-        output_index.flush()?;
-        write_timer.stop();
+        let offsets = self
+            .offsets
+            .iter()
+            .map(|offset| {
+                i64::try_from(*offset).map_err(|_| {
+                    DataFusionError::Execution(format!(
+                        "shuffle write error: offset overflow ({offset})"
+                    ))
+                })
+            })
+            .collect::<Result<Vec<_>, _>>()?;
+        self.partition_offsets.set(offsets)?;
 
         Ok(())
     }
 }
 
+/// Appends `range` of the spill file to `output`, reading it through `buffer` when it fits.
+fn copy_spill_range(
+    spill_file: &mut File,
+    buffer: &mut [u8],
+    range: &Range<u64>,
+    output: &mut BufWriter<File>,
+) -> datafusion::common::Result<()> {
+    let len = range.end - range.start;
+    let truncated = || {
+        DataFusionError::Execution(format!(
+            "shuffle spill file truncated: range {range:?} extends past its end"
+        ))
+    };
+    match usize::try_from(len)
+        .ok()
+        .and_then(|len| buffer.get_mut(..len))
+    {
+        // one pread instead of io::copy's lseek, two statx and copy_file_range
+        Some(chunk) => {
+            spill_file
+                .read_exact_at(chunk, range.start)
+                .map_err(|e| match e.kind() {
+                    ErrorKind::UnexpectedEof => truncated(),
+                    _ => e.into(),
+                })?;
+            output.write_all(chunk)?;
+        }
+        None => {
+            spill_file.seek(SeekFrom::Start(range.start))?;
+            // raw File, not BufReader, so the copy can use copy_file_range on Linux
+            let copied = std::io::copy(&mut Read::by_ref(spill_file).take(len), output)?;
+            if copied != len {
+                return Err(truncated());
+            }
+        }
+    }
+    Ok(())
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
     use crate::writers::local::spill::pathless_backend;
     use crate::CompressionCodec;
     use arrow::array::Int64Array;
+    use arrow::compute::concat_batches;
     use arrow::datatypes::{DataType, Field, Schema};
+    use datafusion::execution::disk_manager::{DiskManagerBuilder, DiskManagerMode};
+    use datafusion::execution::runtime_env::RuntimeEnvBuilder;
     use datafusion::physical_plan::metrics::ExecutionPlanMetricsSet;
 
     fn test_batch() -> RecordBatch {
@@ -349,18 +395,28 @@ mod tests {
         batch: &RecordBatch,
         dir: &tempfile::TempDir,
         runtime: Arc<RuntimeEnv>,
+    ) -> LocalPartitionWriter {
+        partition_writer_with(batch, 2, 1 << 20, dir, runtime)
+    }
+
+    fn partition_writer_with(
+        batch: &RecordBatch,
+        num_partitions: usize,
+        write_buffer_size: usize,
+        dir: &tempfile::TempDir,
+        runtime: Arc<RuntimeEnv>,
     ) -> LocalPartitionWriter {
         let block_writer =
             ShuffleBlockWriter::try_new(batch.schema_ref().as_ref(), CompressionCodec::None)
                 .unwrap();
         LocalPartitionWriter::try_new(
             dir.path().join("data.out").to_str().unwrap().to_string(),
-            dir.path().join("index.out").to_str().unwrap().to_string(),
+            Arc::new(PartitionOffsets::default()),
             block_writer,
-            2,
+            num_partitions,
             // batch_size below the row count so the write serializes into the scratch.
             10,
-            1 << 20,
+            write_buffer_size,
             runtime,
         )
         .unwrap()
@@ -416,4 +472,161 @@ mod tests {
             "unexpected error: {err}"
         );
     }
+
+    fn int_batch(values: std::ops::Range<i64>) -> RecordBatch {
+        let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int64, false)]));
+        RecordBatch::try_new(schema, vec![Arc::new(Int64Array::from_iter_values(values))]).unwrap()
+    }
+
+    fn decode_blocks(bytes: &[u8]) -> Vec<RecordBatch> {
+        let mut batches = Vec::new();
+        let mut pos = 0;
+        while pos < bytes.len() {
+            let len = u64::from_le_bytes(bytes[pos..pos + 8].try_into().unwrap()) as usize;
+            batches.push(crate::read_ipc_compressed(&bytes[pos + 16..pos + 8 + len]).unwrap());
+            pos += 8 + len;
+        }
+        batches
+    }
+
+    fn count_files(dir: &std::path::Path) -> usize {
+        std::fs::read_dir(dir)
+            .unwrap()
+            .map(|entry| {
+                let path = entry.unwrap().path();
+                if path.is_dir() {
+                    count_files(&path)
+                } else {
+                    1
+                }
+            })
+            .sum()
+    }
+
+    /// Partition data spread over spill rounds written in different partition orders, plus a
+    /// final in-memory batch, reads back per partition in write order, whether the spilled
+    /// ranges are read through the scratch buffer or copied.
+    #[test]
+    fn spilled_partitions_read_back_in_write_order() {
+        // a 64-byte write buffer is smaller than one block, so every range is copied
+        for write_buffer_size in [1 << 20, 64] {
+            let dir = tempfile::tempdir().unwrap();
+            let schema = test_batch().schema();
+            let mut writer = partition_writer_with(
+                &test_batch(),
+                4,
+                write_buffer_size,
+                &dir,
+                Arc::new(RuntimeEnv::default()),
+            );
+            let metrics = ShufflePartitionerMetrics::new(&ExecutionPlanMetricsSet::new(), 0);
+            let mut expected: Vec<Vec<RecordBatch>> = vec![Vec::new(); 4];
+            let mut next = 0i64;
+            let mut batch = || {
+                let batch = int_batch(next..next + 10);
+                next += 10;
+                batch
+            };
+
+            for order in [[3, 2, 1, 0], [0, 1, 2, 3]] {
+                for pid in order {
+                    let b = batch();
+                    expected[pid].push(b.clone());
+                    writer
+                        .write(pid, &mut vec![Ok(b)].into_iter(), &metrics)
+                        .unwrap();
+                }
+            }
+            for (pid, rows) in expected.iter_mut().enumerate() {
+                let b = batch();
+                rows.push(b.clone());
+                writer
+                    .finish_partition(pid, &mut vec![Ok(b)].into_iter(), &metrics)
+                    .unwrap();
+            }
+            writer.finish_all(&metrics).unwrap();
+
+            let offsets = writer.partition_offsets.get().unwrap().to_vec();
+            let data = std::fs::read(dir.path().join("data.out")).unwrap();
+            for (pid, rows) in expected.iter().enumerate() {
+                let bytes = &data[offsets[pid] as usize..offsets[pid + 1] as usize];
+                let actual = concat_batches(&schema, &decode_blocks(bytes)).unwrap();
+                assert_eq!(
+                    actual,
+                    concat_batches(&schema, rows).unwrap(),
+                    "partition {pid}, write buffer {write_buffer_size}"
+                );
+            }
+        }
+    }
+
+    /// A task spills to one file however many partitions it has.
+    #[test]
+    fn spilling_every_partition_creates_one_file() {
+        let spill_dir = tempfile::tempdir().unwrap();
+        let output_dir = tempfile::tempdir().unwrap();
+        let runtime = Arc::new(
+            RuntimeEnvBuilder::new()
+                .with_disk_manager_builder(DiskManagerBuilder::default().with_mode(
+                    DiskManagerMode::Directories(vec![spill_dir.path().to_path_buf()]),
+                ))
+                .build()
+                .unwrap(),
+        );
+        let num_partitions = 64;
+        let mut writer =
+            partition_writer_with(&test_batch(), num_partitions, 1 << 20, &output_dir, runtime);
+        let metrics = ShufflePartitionerMetrics::new(&ExecutionPlanMetricsSet::new(), 0);
+
+        for _ in 0..3 {
+            for pid in 0..num_partitions {
+                writer
+                    .write(pid, &mut vec![Ok(test_batch())].into_iter(), &metrics)
+                    .unwrap();
+            }
+        }
+        assert_eq!(count_files(spill_dir.path()), 1);
+    }
+
+    /// A spill file shorter than its recorded ranges fails the task instead of writing a short
+    /// partition, whether the range is read through the scratch buffer or copied.
+    #[test]
+    fn finish_partition_fails_when_spill_file_is_truncated() {
+        for write_buffer_size in [1 << 20, 64] {
+            let dir = tempfile::tempdir().unwrap();
+            let mut writer = partition_writer_with(
+                &test_batch(),
+                2,
+                write_buffer_size,
+                &dir,
+                Arc::new(RuntimeEnv::default()),
+            );
+            let metrics = ShufflePartitionerMetrics::new(&ExecutionPlanMetricsSet::new(), 0);
+            for pid in 0..2 {
+                writer
+                    .write(pid, &mut vec![Ok(test_batch())].into_iter(), &metrics)
+                    .unwrap();
+            }
+            // finishing partition 0 flushes the buffered spill bytes before the file is cut
+            writer
+                .finish_partition(0, &mut std::iter::empty(), &metrics)
+                .unwrap();
+
+            let path = writer.get_spill().path().unwrap().unwrap().to_path_buf();
+            std::fs::OpenOptions::new()
+                .write(true)
+                .open(&path)
+                .unwrap()
+                .set_len(1)
+                .unwrap();
+
+            let err = writer
+                .finish_partition(1, &mut std::iter::empty(), &metrics)
+                .expect_err("a truncated spill file must fail the partition");
+            assert!(
+                err.to_string().contains("truncated"),
+                "write buffer {write_buffer_size}: unexpected error: {err}"
+            );
+        }
+    }
 }
diff --git a/native/shuffle/src/writers/local/spill.rs b/native/shuffle/src/writers/local/spill.rs
index 77fe009046d..a5d541aa51f 100644
--- a/native/shuffle/src/writers/local/spill.rs
+++ b/native/shuffle/src/writers/local/spill.rs
@@ -23,115 +23,162 @@ use datafusion::common::DataFusionError;
 use datafusion::execution::runtime_env::RuntimeEnv;
 use datafusion::execution::SpillFile as DfSpillFile;
 use datafusion::execution::SpillWriter as DfSpillWriter;
+use std::io::{BufWriter, Write};
+use std::ops::Range;
 use std::sync::Arc;
 
 struct ActiveSpillFile {
     temp_file: Arc<dyn DfSpillFile>,
-    writer: Box<dyn DfSpillWriter>,
+    /// Shared by every partition; bytes reach the file when it fills or on
+    /// [`PartitionedSpill::flush`].
+    writer: BufWriter<Box<dyn DfSpillWriter>>,
 }
 
-pub(crate) struct SpillWriter {
+/// Forwards writes but ignores `flush`, so `BufBatchWriter`'s flush at the end of each partition
+/// leaves the bytes buffered.
+struct DeferFlush<'a, W: Write>(&'a mut W);
+
+impl<W: Write> Write for DeferFlush<'_, W> {
+    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+        self.0.write(buf)
+    }
+
+    fn write_all(&mut self, buf: &[u8]) -> std::io::Result<()> {
+        self.0.write_all(buf)
+    }
+
+    fn flush(&mut self) -> std::io::Result<()> {
+        Ok(())
+    }
+}
+
+/// One spill file shared by every output partition of a task, with the ranges each partition's
+/// blocks occupy.
+pub(crate) struct PartitionedSpill {
     shuffle_block_writer: ShuffleBlockWriter,
     write_buffer_size: usize,
     batch_size: usize,
     spill_file: Option<ActiveSpillFile>,
+    /// Bytes appended to the spill file so far.
+    len: u64,
+    /// Per partition, the spill file ranges holding its blocks, in write order.
+    ranges: Vec<Vec<Range<u64>>>,
+    /// Set when a write fails partway, after which `len` may not match the file.
+    failed: bool,
 }
 
-impl SpillWriter {
-    pub(crate) fn try_new(
+impl PartitionedSpill {
+    pub(crate) fn new(
         shuffle_block_writer: ShuffleBlockWriter,
         write_buffer_size: usize,
         batch_size: usize,
-    ) -> datafusion::common::Result<Self> {
-        Ok(Self {
+        num_partitions: usize,
+    ) -> Self {
+        Self {
             shuffle_block_writer,
             write_buffer_size,
             batch_size,
             spill_file: None,
-        })
+            len: 0,
+            ranges: vec![Vec::new(); num_partitions],
+            failed: false,
+        }
     }
 
-    /// `recycled_buffer` is a scratch byte buffer shared by the sequential per-partition
-    /// spill writes; it is left drained on return so one buffer's capacity serves every
-    /// partition instead of each write regrowing its own.
+    /// Appends partition `pid`'s batches to the spill file. `recycled_buffer` is left drained,
+    /// including on error.
     pub(crate) fn write<I: Iterator<Item = datafusion::common::Result<RecordBatch>>>(
         &mut self,
+        pid: usize,
         iter: &mut I,
         runtime: &RuntimeEnv,
         metrics: &ShufflePartitionerMetrics,
         recycled_buffer: &mut Vec<u8>,
     ) -> datafusion::common::Result<()> {
-        if let Some(batch) = iter.next() {
-            self.ensure_spill_file_created(runtime)?;
-
-            let result = (|| {
-                let mut buf_batch_writer = BufBatchWriter::new(
-                    &mut self.shuffle_block_writer,
-                    &mut self.spill_file.as_mut().unwrap().writer,
-                    self.write_buffer_size,
-                    self.batch_size,
-                );
+        self.check_usable()?;
+        let Some(batch) = iter.next() else {
+            return Ok(());
+        };
+        self.ensure_spill_file_created(runtime)?;
+
+        let result = (|| {
+            let mut buf_batch_writer = BufBatchWriter::new(
+                &mut self.shuffle_block_writer,
+                DeferFlush(&mut self.spill_file.as_mut().unwrap().writer),
+                self.write_buffer_size,
+                self.batch_size,
+            );
+            buf_batch_writer.write(
+                &batch?,
+                recycled_buffer,
+                &metrics.encode_time,
+                &metrics.write_time,
+            )?;
+            for batch in iter.by_ref() {
+                let batch = batch?;
                 buf_batch_writer.write(
-                    &batch?,
+                    &batch,
                     recycled_buffer,
                     &metrics.encode_time,
                     &metrics.write_time,
                 )?;
-                for batch in iter.by_ref() {
-                    let batch = batch?;
-                    buf_batch_writer.write(
-                        &batch,
-                        recycled_buffer,
-                        &metrics.encode_time,
-                        &metrics.write_time,
-                    )?;
-                }
-                buf_batch_writer.flush(
-                    recycled_buffer,
-                    &metrics.encode_time,
-                    &metrics.write_time,
-                )?;
-                // `SpillWriter` is not `Seek`, so bytes are tracked by the writer itself rather
-                // than measured via stream position.
-                let bytes_written = buf_batch_writer.bytes_written();
-                usize::try_from(bytes_written).map_err(|_| {
-                    DataFusionError::Execution(format!(
-                        "Spill file byte count exceeds platform capacity: {bytes_written}"
-                    ))
-                })
-            })();
-            // An errored spill must hand back a drained buffer, or its bytes leak into
-            // the next partition's block.
-            let total_bytes_written = result.inspect_err(|_| recycled_buffer.clear())?;
-            metrics.spilled_bytes.add(total_bytes_written);
+            }
+            buf_batch_writer.flush(recycled_buffer, &metrics.encode_time, &metrics.write_time)?;
+            Ok::<_, DataFusionError>(buf_batch_writer.bytes_written())
+        })();
+
+        let bytes_written = match result {
+            Ok(bytes_written) => bytes_written,
+            Err(error) => {
+                // bytes may already be in the file, so later ranges could not be trusted
+                self.failed = true;
+                recycled_buffer.clear();
+                return Err(error);
+            }
+        };
+
+        if bytes_written > 0 {
+            let start = self.len;
+            self.len += bytes_written;
+            self.ranges[pid].push(start..self.len);
         }
+        metrics
+            .spilled_bytes
+            .add(usize::try_from(bytes_written).map_err(|_| {
+                DataFusionError::Execution(format!(
+                    "Spill file byte count exceeds platform capacity: {bytes_written}"
+                ))
+            })?);
         Ok(())
     }
 
-    fn ensure_spill_file_created(
-        &mut self,
-        runtime: &RuntimeEnv,
-    ) -> datafusion::common::Result<()> {
-        if self.spill_file.is_none() {
-            // Spill file is not yet created, create it
-            let temp_file = runtime
-                .disk_manager
-                .create_tmp_file("shuffle writer spill")?;
-            let writer = temp_file.open_writer()?;
-            self.spill_file = Some(ActiveSpillFile { temp_file, writer });
+    /// The spill file ranges holding partition `pid`'s blocks, in write order.
+    pub(crate) fn ranges(&self, pid: usize) -> datafusion::common::Result<&[Range<u64>]> {
+        self.check_usable()?;
+        Ok(&self.ranges[pid])
+    }
+
+    /// Writes buffered spill bytes to the spill file.
+    pub(crate) fn flush(&mut self) -> datafusion::common::Result<()> {
+        self.check_usable()?;
+        if let Some(spill_file) = self.spill_file.as_mut() {
+            if let Err(error) = spill_file.writer.flush() {
+                // the file holds an unknown prefix of the buffered bytes
+                self.failed = true;
+                return Err(error.into());
+            }
         }
         Ok(())
     }
 
-    /// Local filesystem path holding this partition's spilled bytes.
+    /// Local filesystem path of the spill file.
     ///
-    /// * `Ok(None)` — nothing was spilled for this partition.
+    /// * `Ok(None)` — nothing was spilled.
     /// * `Ok(Some(path))` — the spilled bytes live at `path`.
     /// * `Err(..)` — bytes were spilled but the backend exposes no local path.
     ///
-    /// The last case must stay distinct from `Ok(None)`: a caller that treated it as
-    /// "nothing to copy" would drop the spilled bytes while still recording the
-    /// partition offsets, silently truncating the partition in the shuffle file.
+    /// The last case must stay distinct from `Ok(None)`, or spilled bytes would be dropped while
+    /// partition offsets still counted them.
     pub(crate) fn path(&self) -> datafusion::common::Result<Option<&std::path::Path>> {
         match self.spill_file.as_ref() {
             None => Ok(None),
@@ -146,6 +193,29 @@ impl SpillWriter {
         }
     }
 
+    fn check_usable(&self) -> datafusion::common::Result<()> {
+        if self.failed {
+            return Err(DataFusionError::Execution(
+                "Shuffle spill file is unusable after a failed write".to_string(),
+            ));
+        }
+        Ok(())
+    }
+
+    fn ensure_spill_file_created(
+        &mut self,
+        runtime: &RuntimeEnv,
+    ) -> datafusion::common::Result<()> {
+        if self.spill_file.is_none() {
+            let temp_file = runtime
+                .disk_manager
+                .create_tmp_file("shuffle writer spill")?;
+            let writer = BufWriter::with_capacity(self.write_buffer_size, temp_file.open_writer()?);
+            self.spill_file = Some(ActiveSpillFile { temp_file, writer });
+        }
+        Ok(())
+    }
+
     #[cfg(test)]
     pub(crate) fn has_spill_file(&self) -> bool {
         self.spill_file.is_some()
@@ -251,32 +321,34 @@ mod tests {
         .unwrap()
     }
 
-    fn spill_writer(batch: &RecordBatch, batch_size: usize) -> SpillWriter {
+    fn partitioned_spill(batch: &RecordBatch, num_partitions: usize) -> PartitionedSpill {
         let block_writer =
             ShuffleBlockWriter::try_new(batch.schema_ref().as_ref(), CompressionCodec::None)
                 .unwrap();
-        SpillWriter::try_new(block_writer, 1 << 20, batch_size).unwrap()
+        // batch_size below the row count so a write serializes into the scratch
+        PartitionedSpill::new(block_writer, 1 << 20, 10, num_partitions)
     }
 
-    /// A spill whose batch iterator fails after a batch was already encoded must hand
-    /// back a drained scratch; leftover bytes would land in the next partition's block.
-    #[test]
-    fn write_error_drains_recycled_buffer() {
-        let batch = test_batch();
-        // batch_size below the row count so the first write serializes into the scratch.
-        let mut spill = spill_writer(&batch, 10);
-        let runtime = RuntimeEnv::default();
-        let metrics = ShufflePartitionerMetrics::new(&ExecutionPlanMetricsSet::new(), 0);
-        let mut recycled = Vec::new();
+    fn metrics() -> ShufflePartitionerMetrics {
+        ShufflePartitionerMetrics::new(&ExecutionPlanMetricsSet::new(), 0)
+    }
+
+    fn failing_write(spill: &mut PartitionedSpill, recycled: &mut Vec<u8>) {
         let mut iter = vec![
-            Ok(batch),
+            Ok(test_batch()),
             Err(DataFusionError::Execution("injected failure".to_string())),
         ]
         .into_iter();
-
         assert!(spill
-            .write(&mut iter, &runtime, &metrics, &mut recycled)
+            .write(0, &mut iter, &RuntimeEnv::default(), &metrics(), recycled)
             .is_err());
+    }
+
+    #[test]
+    fn write_error_drains_recycled_buffer() {
+        let mut spill = partitioned_spill(&test_batch(), 2);
+        let mut recycled = Vec::new();
+        failing_write(&mut spill, &mut recycled);
         assert!(
             recycled.is_empty(),
             "errored spill left {} bytes in the recycled buffer",
@@ -284,34 +356,118 @@ mod tests {
         );
     }
 
-    /// A partition that never spilled has no path, and that is not an error.
+    #[test]
+    fn failed_write_makes_spill_unusable() {
+        let mut spill = partitioned_spill(&test_batch(), 2);
+        let mut recycled = Vec::new();
+        failing_write(&mut spill, &mut recycled);
+
+        let err = spill
+            .write(
+                1,
+                &mut vec![Ok(test_batch())].into_iter(),
+                &RuntimeEnv::default(),
+                &metrics(),
+                &mut recycled,
+            )
+            .expect_err("write after a failed write");
+        assert!(
+            err.to_string().contains("unusable"),
+            "unexpected error: {err}"
+        );
+        assert!(spill.ranges(1).is_err());
+    }
+
+    #[test]
+    fn partitions_share_one_file_in_write_order() {
+        let mut spill = partitioned_spill(&test_batch(), 2);
+        let runtime = RuntimeEnv::default();
+        let mut recycled = Vec::new();
+        for pid in [1, 0, 1] {
+            spill
+                .write(
+                    pid,
+                    &mut vec![Ok(test_batch())].into_iter(),
+                    &runtime,
+                    &metrics(),
+                    &mut recycled,
+                )
+                .unwrap();
+        }
+
+        let first = spill.ranges(1).unwrap().to_vec();
+        let second = spill.ranges(0).unwrap().to_vec();
+        assert_eq!(first.len(), 2);
+        assert_eq!(second.len(), 1);
+        assert_eq!(first[0].start, 0);
+        assert_eq!(second[0].start, first[0].end);
+        assert_eq!(first[1].start, second[0].end);
+    }
+
+    #[test]
+    fn writes_stay_buffered_until_flush() {
+        let mut spill = partitioned_spill(&test_batch(), 2);
+        let runtime = RuntimeEnv::default();
+        let mut recycled = Vec::new();
+        for pid in [0, 1] {
+            spill
+                .write(
+                    pid,
+                    &mut vec![Ok(test_batch())].into_iter(),
+                    &runtime,
+                    &metrics(),
+                    &mut recycled,
+                )
+                .unwrap();
+        }
+        let path = spill.path().unwrap().unwrap().to_path_buf();
+        let spilled = spill.ranges(1).unwrap()[0].end;
+
+        assert_eq!(std::fs::metadata(&path).unwrap().len(), 0);
+        spill.flush().unwrap();
+        assert_eq!(std::fs::metadata(&path).unwrap().len(), spilled);
+    }
+
+    #[test]
+    fn empty_write_records_no_range() {
+        let mut spill = partitioned_spill(&test_batch(), 2);
+        spill
+            .write(
+                0,
+                &mut std::iter::empty(),
+                &RuntimeEnv::default(),
+                &metrics(),
+                &mut Vec::new(),
+            )
+            .unwrap();
+        assert!(!spill.has_spill_file());
+        assert!(spill.ranges(0).unwrap().is_empty());
+    }
+
     #[test]
     fn path_is_none_when_nothing_spilled() {
-        let batch = test_batch();
-        let spill = spill_writer(&batch, 10);
+        let spill = partitioned_spill(&test_batch(), 2);
         assert!(!spill.has_spill_file());
         assert_eq!(spill.path().unwrap(), None);
     }
 
-    /// Spilling to a backend with no local path must report an error rather than the
-    /// `None` that means "nothing spilled" — see `path`'s doc comment.
     #[test]
     fn path_errors_when_backend_has_no_local_path() {
-        let batch = test_batch();
-        let mut spill = spill_writer(&batch, 10);
-        let runtime = pathless_backend::runtime();
-        let metrics = ShufflePartitionerMetrics::new(&ExecutionPlanMetricsSet::new(), 0);
-        let mut recycled = Vec::new();
-        let mut iter = vec![Ok(batch)].into_iter();
-
+        let mut spill = partitioned_spill(&test_batch(), 2);
         spill
-            .write(&mut iter, &runtime, &metrics, &mut recycled)
+            .write(
+                0,
+                &mut vec![Ok(test_batch())].into_iter(),
+                &pathless_backend::runtime(),
+                &metrics(),
+                &mut Vec::new(),
+            )
             .unwrap();
         assert!(spill.has_spill_file());
 
         let err = spill
             .path()
-            .expect_err("a spill file with no local path must not look like an empty partition");
+            .expect_err("a spill file with no local path must not look like nothing spilled");
         assert!(
             err.to_string().contains("no local path"),
             "unexpected error: {err}"
diff --git a/spark/src/main/scala/org/apache/comet/CometExecIterator.scala b/spark/src/main/scala/org/apache/comet/CometExecIterator.scala
index e2c132904d5..4da95af18d0 100644
--- a/spark/src/main/scala/org/apache/comet/CometExecIterator.scala
+++ b/spark/src/main/scala/org/apache/comet/CometExecIterator.scala
@@ -64,6 +64,10 @@ import org.apache.comet.vector.NativeUtil
  *   Paths to encrypted Parquet files that need key unwrapping.
  * @param shufflePartitionPusher
  *   Optional task-owned callback that receives remote shuffle output.
+ * @param capturePartitionOffsets
+ *   Whether to read the shuffle writer's partition offsets when the plan reaches the end of its
+ *   output, for a plan rooted at a native shuffle writer with a local destination. Remote shuffle
+ *   reports its partition lengths through its pusher instead, so it leaves this false.
  */
 class CometExecIterator(
     val id: Long,
@@ -77,7 +81,8 @@ class CometExecIterator(
     encryptedFilePaths: Seq[String] = Seq.empty,
     shuffleBlockIterators: Map[Int, CometShuffleBlockIterator] = Map.empty,
     taskFilePaths: Seq[String] = Seq.empty,
-    shufflePartitionPusher: Option[ShufflePartitionPusher] = None)
+    shufflePartitionPusher: Option[ShufflePartitionPusher] = None,
+    capturePartitionOffsets: Boolean = false)
     extends Iterator[ColumnarBatch]
     with Logging {
 
@@ -179,6 +184,30 @@ class CometExecIterator(
     }
   }
 
+  /** Set once by [[readPartitionOffsetsBeforeClose]]; `null` until then. */
+  private var partitionOffsets: Array[Long] = _
+
+  /**
+   * Partition offsets from a native shuffle write, or `null` if this iterator was not built to
+   * collect them or has not yet reached the end of its output.
+   */
+  def shufflePartitionOffsets: Array[Long] = partitionOffsets
+
+  /**
+   * Reads the shuffle writer's partition offsets out of the native plan, if this iterator was
+   * built to collect them.
+   *
+   * This has to run at end of stream rather than after iteration finishes. The offsets live in
+   * the native execution context; [[close]] releases that context, and [[hasNext]] calls
+   * [[close]] as soon as the plan runs out of output. So the final [[hasNext]] is the last point
+   * at which they can still be read.
+   */
+  private def readPartitionOffsetsBeforeClose(): Unit = {
+    if (capturePartitionOffsets && partitionOffsets == null) {
+      partitionOffsets = nativeLib.getShufflePartitionOffsets(plan)
+    }
+  }
+
   private var nextBatch: Option[ColumnarBatch] = None
   private var prevBatch: ColumnarBatch = null
   private var currentBatch: ColumnarBatch = null
@@ -248,6 +277,7 @@ class CometExecIterator(
     logTrace(s"Task $taskAttemptId memory pool usage is ${cometTaskMemoryManager.getUsed} bytes")
 
     if (nextBatch.isEmpty) {
+      readPartitionOffsetsBeforeClose()
       close()
       false
     } else {
diff --git a/spark/src/main/scala/org/apache/comet/Native.scala b/spark/src/main/scala/org/apache/comet/Native.scala
index af406632f15..c3dccca162c 100644
--- a/spark/src/main/scala/org/apache/comet/Native.scala
+++ b/spark/src/main/scala/org/apache/comet/Native.scala
@@ -116,6 +116,21 @@ class Native extends NativeBase {
       arrayAddrs: Array[Long],
       schemaAddrs: Array[Long]): Long
 
+  /**
+   * Returns the partition offsets published by a finished native shuffle write.
+   *
+   * The writer knows every offset once its plan completes, so they are handed back in memory
+   * rather than through a temporary index file. Call only after the plan has been fully drained,
+   * and only for a plan whose root is a native shuffle writer with a local destination.
+   *
+   * @param plan
+   *   the address to native query plan.
+   * @return
+   *   `numPartitions + 1` offsets, the last being the total data file length, so that partition
+   *   lengths are successive differences.
+   */
+  @native def getShufflePartitionOffsets(plan: Long): Array[Long]
+
   /**
    * Release and drop the native query plan object and context object.
    *
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 21dd7686302..4b10019341f 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
@@ -19,7 +19,6 @@
 
 package org.apache.spark.sql.comet.execution.shuffle
 
-import java.nio.{ByteBuffer, ByteOrder}
 import java.nio.file.{Files, Paths}
 import java.util.concurrent.{ScheduledFuture, TimeUnit}
 import java.util.concurrent.atomic.AtomicBoolean
@@ -69,8 +68,6 @@ class CometNativeShuffleWriter[K, V](
     extends ShuffleWriter[K, V]
     with Logging {
 
-  private val OFFSET_LENGTH = 8
-
   var partitionLengths: Array[Long] = _
   var mapStatus: MapStatus = _
   private var stopped = false
@@ -116,12 +113,7 @@ class CometNativeShuffleWriter[K, V](
       val resolver =
         SparkEnv.get.shuffleManager.shuffleBlockResolver.asInstanceOf[IndexShuffleBlockResolver]
       val dataFile = resolver.getDataFile(shuffleId, mapId)
-      val indexFile = resolver.getIndexFile(shuffleId, mapId)
-      Some(
-        LocalShuffleOutput(
-          resolver,
-          dataFile.getPath.replace(".data", ".data.tmp"),
-          indexFile.getPath.replace(".index", ".index.tmp")))
+      Some(LocalShuffleOutput(resolver, dataFile.getPath.replace(".data", ".data.tmp")))
     } else {
       None
     }
@@ -142,8 +134,8 @@ class CometNativeShuffleWriter[K, V](
     val shuffleBlockIters = shuffleInputIter.shuffleBlockIterators
 
     val unifiedPlan = localOutput match {
-      case Some(output) => buildUnifiedPlan(output.dataFile, output.indexFile)
-      case None => buildUnifiedPlan("", "")
+      case Some(output) => buildUnifiedPlan(output.dataFile)
+      case None => buildUnifiedPlan("")
     }
     val ctx = spec.execContext
     val finalNativePlan = if (ctx.commonByKey.nonEmpty) {
@@ -198,7 +190,10 @@ class CometNativeShuffleWriter[K, V](
       ctx.broadcastedHadoopConfForEncryption,
       ctx.encryptedFilePaths,
       shuffleBlockIters,
-      shufflePartitionPusher = remoteDestination.map(_.callback))
+      shufflePartitionPusher = remoteDestination.map(_.callback),
+      // Only a local destination publishes partition offsets; RSS reports lengths through its
+      // pusher instead.
+      capturePartitionOffsets = localOutput.isDefined)
 
     // Register subqueries against the iterator id so native callbacks resolve them to values.
     ctx.subqueries.foreach { sub =>
@@ -213,6 +208,8 @@ class CometNativeShuffleWriter[K, V](
     }
 
     CometNativeShuffleWriter.drainAndClose(cometIter, () => cometIter.close())
+    // Captured by the iterator at end of stream, before it released the native plan that owns it.
+    val partitionOffsets = cometIter.shufflePartitionOffsets
 
     remoteDestination match {
       case Some(destination) =>
@@ -243,22 +240,17 @@ class CometNativeShuffleWriter[K, V](
       case None =>
         val output = localOutput.get
         val tempDataFilePath = Paths.get(output.dataFile)
-        val tempIndexFilePath = Paths.get(output.indexFile)
-
-        var offset = 0L
-        partitionLengths = Files
-          .readAllBytes(tempIndexFilePath)
-          .grouped(OFFSET_LENGTH)
-          .drop(1)
-          .map(indexBytes => {
-            val partitionOffset =
-              ByteBuffer.wrap(indexBytes).order(ByteOrder.LITTLE_ENDIAN).getLong
-            val partitionLength = partitionOffset - offset
-            offset = partitionOffset
-            partitionLength
-          })
-          .toArray
-        Files.delete(tempIndexFilePath)
+
+        require(
+          partitionOffsets != null && partitionOffsets.length >= 1,
+          "Native shuffle returned no partition offsets")
+        partitionLengths = new Array[Long](partitionOffsets.length - 1)
+        var partition = 0
+        while (partition < partitionLengths.length) {
+          partitionLengths(partition) =
+            partitionOffsets(partition + 1) - partitionOffsets(partition)
+          partition += 1
+        }
 
         metricsReporter.incBytesWritten(Files.size(tempDataFilePath))
         output.resolver.writeMetadataFileAndCommit(
@@ -290,7 +282,7 @@ class CometNativeShuffleWriter[K, V](
    * Build the unified `ShuffleWriter(child = childNativeOp)` plan with the partitioning serde,
    * compression settings, and output file paths.
    */
-  private[shuffle] def buildUnifiedPlan(dataFile: String, indexFile: String): Operator = {
+  private[shuffle] def buildUnifiedPlan(dataFile: String): Operator = {
     val shuffleWriterBuilder = OperatorOuterClass.ShuffleWriter.newBuilder()
     remoteDestination match {
       case Some(_) =>
@@ -300,9 +292,9 @@ class CometNativeShuffleWriter[K, V](
             .setRss(OperatorOuterClass.RssPartitionWriter.getDefaultInstance)
             .build())
       case None =>
-        // Keep legacy paths for older native libraries while newer libraries use the destination.
+        // Keep the legacy path for older native libraries while newer libraries use the
+        // destination. Partition offsets come back over JNI.
         shuffleWriterBuilder.setOutputDataFile(dataFile)
-        shuffleWriterBuilder.setOutputIndexFile(indexFile)
         shuffleWriterBuilder.setPartitionWriter(
           OperatorOuterClass.PartitionWriter
             .newBuilder()
@@ -310,7 +302,6 @@ class CometNativeShuffleWriter[K, V](
               OperatorOuterClass.LocalPartitionWriter
                 .newBuilder()
                 .setOutputDataFile(dataFile)
-                .setOutputIndexFile(indexFile)
                 .build())
             .build())
     }
@@ -482,8 +473,7 @@ class CometNativeShuffleWriter[K, V](
 
   private final case class LocalShuffleOutput(
       resolver: IndexShuffleBlockResolver,
-      dataFile: String,
-      indexFile: String)
+      dataFile: String)
 }
 
 private[shuffle] object CometNativeShuffleWriter {
diff --git a/spark/src/test/scala/org/apache/comet/exec/CometNativeShuffleSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometNativeShuffleSuite.scala
index d5e768f886e..a65d86fd203 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometNativeShuffleSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometNativeShuffleSuite.scala
@@ -344,18 +344,15 @@ class CometNativeShuffleSuite extends CometTestBase with AdaptiveSparkPlanHelper
     assert(results.sameElements(Array((true, true, true, true))))
   }
 
-  test("native shuffle plan preserves local partition writer and legacy output paths") {
+  test("native shuffle plan preserves local partition writer and legacy output path") {
     val dataFile = "/tmp/comet-shuffle.data"
-    val indexFile = "/tmp/comet-shuffle.index"
     val localWriter = OperatorOuterClass.LocalPartitionWriter
       .newBuilder()
       .setOutputDataFile(dataFile)
-      .setOutputIndexFile(indexFile)
       .build()
     val writer = OperatorOuterClass.ShuffleWriter
       .newBuilder()
       .setOutputDataFile(dataFile)
-      .setOutputIndexFile(indexFile)
       .setPartitionWriter(
         OperatorOuterClass.PartitionWriter.newBuilder().setLocal(localWriter).build())
       .build()
@@ -366,16 +363,13 @@ class CometNativeShuffleSuite extends CometTestBase with AdaptiveSparkPlanHelper
     assert(decoded.getPartitionWriter.hasLocal)
     assert(!decoded.getPartitionWriter.hasRss)
     assert(decoded.getPartitionWriter.getLocal.getOutputDataFile == dataFile)
-    assert(decoded.getPartitionWriter.getLocal.getOutputIndexFile == indexFile)
     assert(decoded.getOutputDataFile == dataFile)
-    assert(decoded.getOutputIndexFile == indexFile)
   }
 
   test("native shuffle plan preserves RSS partition writer and excludes local destination") {
     val localWriter = OperatorOuterClass.LocalPartitionWriter
       .newBuilder()
       .setOutputDataFile("/tmp/comet-shuffle.data")
-      .setOutputIndexFile("/tmp/comet-shuffle.index")
       .build()
     val partitionWriter = OperatorOuterClass.PartitionWriter
       .newBuilder()
@@ -393,23 +387,19 @@ class CometNativeShuffleSuite extends CometTestBase with AdaptiveSparkPlanHelper
     assert(decoded.getPartitionWriter.hasRss)
     assert(!decoded.getPartitionWriter.hasLocal)
     assert(decoded.getOutputDataFile.isEmpty)
-    assert(decoded.getOutputIndexFile.isEmpty)
   }
 
   test("legacy native shuffle plans remain valid without a partition writer") {
     val dataFile = "/tmp/legacy-shuffle.data"
-    val indexFile = "/tmp/legacy-shuffle.index"
     val writer = OperatorOuterClass.ShuffleWriter
       .newBuilder()
       .setOutputDataFile(dataFile)
-      .setOutputIndexFile(indexFile)
       .build()
 
     val decoded = OperatorOuterClass.ShuffleWriter.parseFrom(writer.toByteArray)
 
     assert(!decoded.hasPartitionWriter)
     assert(decoded.getOutputDataFile == dataFile)
-    assert(decoded.getOutputIndexFile == indexFile)
   }
 
   // TODO: this test takes a long time to run, we should reduce the test time.
diff --git a/spark/src/test/scala/org/apache/spark/sql/comet/execution/shuffle/CometCelebornNativeShuffleWriterSuite.scala b/spark/src/test/scala/org/apache/spark/sql/comet/execution/shuffle/CometCelebornNativeShuffleWriterSuite.scala
index a300bf8de9f..048b4c09a6e 100644
--- a/spark/src/test/scala/org/apache/spark/sql/comet/execution/shuffle/CometCelebornNativeShuffleWriterSuite.scala
+++ b/spark/src/test/scala/org/apache/spark/sql/comet/execution/shuffle/CometCelebornNativeShuffleWriterSuite.scala
@@ -209,10 +209,9 @@ class CometCelebornNativeShuffleWriterSuite extends CometTestBase {
               validations += 1
               true
             })
-          val plan = writer.buildUnifiedPlan("", "").getShuffleWriter
+          val plan = writer.buildUnifiedPlan("").getShuffleWriter
           assert(plan.getPartitionWriter.hasRss)
           assert(plan.getOutputDataFile.isEmpty)
-          assert(plan.getOutputIndexFile.isEmpty)
 
           writer.write(inputs)
           val status = writer.stop(success = true).get