diff --git a/docs/source/contributor-guide/native_shuffle.md b/docs/source/contributor-guide/native_shuffle.md
index d9d675e7de3..1608b70f467 100644
--- a/docs/source/contributor-guide/native_shuffle.md
+++ b/docs/source/contributor-guide/native_shuffle.md
@@ -151,12 +151,12 @@ Native shuffle (`CometExchange`) is selected when all of the following condition
    - Writes field count header
    - Writes compressed IPC stream
 
-6. **Output files**: Two files are produced:
-   - **Data file**: Concatenated partition data
-   - **Index file**: Array of 8-byte little-endian offsets marking partition boundaries
+6. **Output**: One data file holds the concatenated partition data. The writer records the byte
+   offset where each partition begins, plus the total length, and keeps them in memory.
 
-7. **Commit**: Back in JVM, `CometNativeShuffleWriter` reads the index file to get partition
-   lengths and commits via Spark's `IndexShuffleBlockResolver`.
+7. **Commit**: Back in JVM, `CometNativeShuffleWriter` fetches the offsets with
+   `Native.getShufflePartitionOffsets`, converts them to partition lengths, and commits via
+   Spark's `IndexShuffleBlockResolver.writeMetadataFileAndCommit`, which writes Spark's index file.
 
 ### Read Path
 
diff --git a/native/core/src/execution/jni_api.rs b/native/core/src/execution/jni_api.rs
index 0107847c9e0..6eb58f07ee4 100644
--- a/native/core/src/execution/jni_api.rs
+++ b/native/core/src/execution/jni_api.rs
@@ -101,7 +101,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;
 
@@ -1200,6 +1200,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 50f708a283b..38b940475e4 100644
--- a/native/core/src/execution/planner.rs
+++ b/native/core/src/execution/planner.rs
@@ -49,7 +49,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;
@@ -4166,7 +4166,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()),
         });
     };
 
@@ -4178,12 +4178,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
             {
@@ -4194,16 +4188,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"
@@ -4213,11 +4197,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(),
                 ));
@@ -5216,15 +5200,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(),
                 },
             )),
         }
@@ -5241,15 +5221,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:?}"),
         }
@@ -5259,49 +5239,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()
         };
 
@@ -5312,29 +5281,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()
         };
 
@@ -5345,21 +5296,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 {
@@ -5747,28 +5683,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> =
@@ -5786,10 +5704,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..b1d6a173bfb 100644
--- a/native/proto/src/lib.rs
+++ b/native/proto/src/lib.rs
@@ -68,15 +68,12 @@ mod tests {
 
     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 +86,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,9 +116,12 @@ 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");
+        // a new plan carries no index path, so a reader expecting tag 4 sees it unset
+        assert!(decoded.output_index_file.is_empty());
     }
 
+    /// A plan still carrying the retired index path decodes cleanly: tag 4 is reserved, so it is
+    /// skipped as an unknown field.
     #[test]
     fn new_shuffle_writer_decodes_legacy_plan_without_destination() {
         let legacy = LegacyShuffleWriter {
@@ -133,7 +131,6 @@ mod tests {
         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 fb372727d19..aaa1af076ea 100644
--- a/native/proto/src/proto/operator.proto
+++ b/native/proto/src/proto/operator.proto
@@ -857,10 +857,12 @@ 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 {
+  reserved 2;
+  reserved "output_index_file";
+
   string output_data_file = 1;
-  string output_index_file = 2;
 }
 
 // Marker for remote shuffle output. The task-owned callback is bound outside
@@ -869,10 +871,12 @@ message RssPartitionWriter {}
 
 message ShuffleWriter {
   spark.spark_partitioning.Partitioning partitioning = 1;
+  reserved 4;
+  reserved "output_index_file";
+
   // 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;
diff --git a/native/shuffle/benches/shuffle_writer.rs b/native/shuffle/benches/shuffle_writer.rs
index c535ccbb137..cc94375436e 100644
--- a/native/shuffle/benches/shuffle_writer.rs
+++ b/native/shuffle/benches/shuffle_writer.rs
@@ -220,7 +220,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..051e18fbdd8 100644
--- a/native/shuffle/src/bin/shuffle_bench.rs
+++ b/native/shuffle/src/bin/shuffle_bench.rs
@@ -98,7 +98,7 @@ struct Args {
     #[arg(long, default_value_t = 0)]
     warmup: usize,
 
-    /// Output directory for shuffle data/index files
+    /// Output directory for the shuffle data file
     #[arg(long, default_value = "/tmp/comet_shuffle_bench")]
     output_dir: PathBuf,
 
@@ -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 938c13e0fda..0eb18b517f3 100644
--- a/native/shuffle/src/lib.rs
+++ b/native/shuffle/src/lib.rs
@@ -36,5 +36,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/partitioners/traits.rs b/native/shuffle/src/partitioners/traits.rs
index 5bd41a62a50..904c7d00883 100644
--- a/native/shuffle/src/partitioners/traits.rs
+++ b/native/shuffle/src/partitioners/traits.rs
@@ -22,6 +22,6 @@ use datafusion::common::Result;
 pub(crate) trait ShufflePartitioner: Send {
     /// Insert a batch into the partitioner
     async fn insert_batch(&mut self, batch: RecordBatch) -> Result<()>;
-    /// Write shuffle data and shuffle index file to disk
+    /// Write the buffered shuffle data to the partition writer
     fn shuffle_write(&mut self) -> Result<()>;
 }
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 5c85168221b..5fb665150b7 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(),
@@ -539,7 +571,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,
@@ -597,7 +629,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,
@@ -679,7 +711,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,
@@ -741,7 +773,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,
@@ -884,7 +916,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,
@@ -992,7 +1024,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(
@@ -1002,7 +1033,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,
@@ -1127,7 +1157,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,
@@ -1175,9 +1204,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(
@@ -1187,7 +1216,6 @@ mod test {
                 CometPartitioning::RoundRobin(num_partitions, 0),
                 CompressionCodec::Zstd(1),
                 data_file.clone(),
-                index_file.clone(),
                 false,
                 1024 * 1024,
                 None,
@@ -1210,6 +1238,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();
@@ -1227,20 +1263,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"
                 );
             }
         }
@@ -1508,13 +1534,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,
@@ -1588,7 +1612,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(
@@ -1597,7 +1620,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,
@@ -1628,29 +1650,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"
             );
         }
@@ -1677,7 +1696,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(
@@ -1686,7 +1704,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,
@@ -1709,17 +1726,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 ae68cd1b28e..72787236f93 100644
--- a/native/shuffle/src/writers/local/local_partition_writer.rs
+++ b/native/shuffle/src/writers/local/local_partition_writer.rs
@@ -20,7 +20,7 @@ use crate::metrics::ShufflePartitionerMetrics;
 use crate::writers::local::spill::SpillWriter;
 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;
@@ -70,11 +70,11 @@ enum DataOutput {
 
 /// Local file-based [`PartitionWriter`] implementation.
 ///
-/// Writes shuffle output to a single data file plus an index file recording the
-/// byte offset where each partition begins. See [`DataOutput`] for how the
+/// 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 {
-    output_index_file: String,
+    partition_offsets: Arc<PartitionOffsets>,
     data_output: DataOutput,
     /// Compression state shared by every block this task writes; the per-partition
     /// `BufBatchWriter`s borrow it (see [`ShuffleCodecContext`]). Retention is bounded:
@@ -95,7 +95,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,
@@ -139,7 +139,7 @@ impl LocalPartitionWriter {
             }
         };
         Ok(Self {
-            output_index_file,
+            partition_offsets,
             data_output,
             codec_context: ShuffleCodecContext::default(),
             offsets: vec![0u64; num_output_partitions + 1],
@@ -348,22 +348,18 @@ 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)?;
 
         // The shuffle output is complete; nothing else encodes through this context.
         self.codec_context.release_zstd();
@@ -406,7 +402,7 @@ mod tests {
                 .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,
             // batch_size below the row count so the write serializes into the scratch.
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 e33d0600527..93b396ce0f2 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 959522b094d..fce4291deb6 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) {
@@ -203,7 +195,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 =>
@@ -218,6 +213,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) =>
@@ -248,22 +245,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(
@@ -295,7 +287,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(_) =>
@@ -305,9 +297,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()
@@ -315,7 +307,6 @@ class CometNativeShuffleWriter[K, V](
               OperatorOuterClass.LocalPartitionWriter
                 .newBuilder()
                 .setOutputDataFile(dataFile)
-                .setOutputIndexFile(indexFile)
                 .build())
             .build())
     }
@@ -487,8 +478,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 ff6ef3b10f0..620a9243c5e 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometNativeShuffleSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometNativeShuffleSuite.scala
@@ -350,18 +350,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()
@@ -372,16 +369,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()
@@ -399,23 +393,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