From e8f09917d911b02cd05365b219db4bc7a144f1e3 Mon Sep 17 00:00:00 2001
From: Jack Ye <yezhaoqin@gmail.com>
Date: Mon, 5 Oct 2026 10:30:20 -0700
Subject: [PATCH 01/14] perf(mem_wal): group blob sidecar writes

---
 .../dataset/mem_wal/memtable/batch_store.rs   |  22 +++
 rust/lance/src/dataset/mem_wal/wal.rs         |  10 ++
 rust/lance/src/dataset/mem_wal/write.rs       | 144 +++++++++++++++---
 3 files changed, 156 insertions(+), 20 deletions(-)

diff --git a/rust/lance/src/dataset/mem_wal/memtable/batch_store.rs b/rust/lance/src/dataset/mem_wal/memtable/batch_store.rs
index b0384e6d896..3e44e3782a2 100644
--- a/rust/lance/src/dataset/mem_wal/memtable/batch_store.rs
+++ b/rust/lance/src/dataset/mem_wal/memtable/batch_store.rs
@@ -50,7 +50,9 @@ use arrow::array::ArrayData;
 use arrow_array::RecordBatch;
 use arrow_buffer::Buffer;
 use arrow_schema::DataType;
+use tokio::sync::Mutex as AsyncMutex;
 
+use crate::dataset::blob::BlobPreprocessor;
 use crate::dataset::mem_wal::MemTableDataTarget;
 
 /// A batch stored in the lock-free store.
@@ -231,6 +233,10 @@ pub struct BatchStore {
     /// entry so replay keeps the writer's generation boundaries, even if the
     /// shard is reopened with a different size limit.
     generation: u64,
+
+    /// Generation-scoped Blob v2 writer. Durable puts reuse its current pack;
+    /// the WAL path seals it before persisting any descriptors that reference it.
+    blob_preprocessor: AsyncMutex<Option<BlobPreprocessor>>,
 }
 
 // SAFETY: Safe to share across threads because:
@@ -238,6 +244,7 @@ pub struct BatchStore {
 // - Readers only access committed slots (index < committed_len)
 // - Atomic operations provide proper synchronization
 // - Slots are never modified after being written
+// - Blob preprocessing is serialized by its async mutex
 unsafe impl Sync for BatchStore {}
 unsafe impl Send for BatchStore {}
 
@@ -299,6 +306,7 @@ impl BatchStore {
             global_offset,
             target,
             generation,
+            blob_preprocessor: AsyncMutex::new(None),
         }
     }
 
@@ -602,6 +610,20 @@ impl BatchStore {
         self.target.as_ref()
     }
 
+    pub(crate) fn blob_preprocessor(&self) -> &AsyncMutex<Option<BlobPreprocessor>> {
+        &self.blob_preprocessor
+    }
+
+    /// Seal the generation's current packed sidecar before its descriptors are
+    /// made durable in the WAL. The preprocessor remains reusable so later puts
+    /// can start a new pack under the same generation target.
+    pub(crate) async fn finish_blob_pack(&self) -> crate::Result<()> {
+        if let Some(preprocessor) = self.blob_preprocessor.lock().await.as_mut() {
+            preprocessor.finish().await?;
+        }
+        Ok(())
+    }
+
     /// The local exclusive end of this store covered by a writer-global cursor.
     ///
     /// Saturating in both directions, and both directions are reachable in
diff --git a/rust/lance/src/dataset/mem_wal/wal.rs b/rust/lance/src/dataset/mem_wal/wal.rs
index bd37ca31e9a..202ed20433f 100644
--- a/rust/lance/src/dataset/mem_wal/wal.rs
+++ b/rust/lance/src/dataset/mem_wal/wal.rs
@@ -870,6 +870,16 @@ impl WalFlusher {
             return Ok(empty_flush_result());
         }
 
+        // Blob descriptors are inserted before their shared packed sidecar is
+        // finalized so several puts can use one object. Seal that object before
+        // any of those descriptors enter the WAL: after the append succeeds,
+        // replay is allowed to expose every descriptor in this range.
+        batch_store.finish_blob_pack().await.map_err(|error| {
+            Error::writer_poisoned(format!(
+                "failed to finalize Blob v2 pack before WAL append: {error}"
+            ))
+        })?;
+
         // Every position in the range must exist. `get` returns `None` only for a
         // position past `committed_len`, so a hole means we were asked to append a
         // batch the store never committed. Silently skipping it while still
diff --git a/rust/lance/src/dataset/mem_wal/write.rs b/rust/lance/src/dataset/mem_wal/write.rs
index 54f21efbe41..e57aefbb6ce 100644
--- a/rust/lance/src/dataset/mem_wal/write.rs
+++ b/rust/lance/src/dataset/mem_wal/write.rs
@@ -2183,9 +2183,9 @@ impl SharedWriterState {
         memtable: &MemTable,
         schema: &WriterSchema,
         batches: Vec<RecordBatch>,
-    ) -> Result<(Vec<RecordBatch>, Option<BlobPreprocessor>)> {
+    ) -> Result<(Vec<RecordBatch>, Option<BlobPreprocessor>, bool)> {
         let Some(target) = memtable.target() else {
-            return Ok((batches, None));
+            return Ok((batches, None, false));
         };
         let logical_schema = Schema::try_from(schema.storage.as_ref())?;
         let data_dir = target
@@ -2198,19 +2198,55 @@ impl SharedWriterState {
             .map(|session| session.store_registry())
             .unwrap_or_default();
         let source_store_params = self.config.store_params.clone().unwrap_or_default();
-        let mut preprocessor = BlobPreprocessor::new(
-            self.object_store.as_ref().clone(),
-            data_dir,
-            target.data_file_key().to_string(),
-            &logical_schema,
-            None,
-            true,
-            ExternalBlobMode::Reference,
-            source_store_registry,
-            source_store_params,
-            None,
-        )?
-        .for_mem_wal(memtable.blob_id_allocator());
+        let new_preprocessor = || {
+            BlobPreprocessor::new(
+                self.object_store.as_ref().clone(),
+                data_dir.clone(),
+                target.data_file_key().to_string(),
+                &logical_schema,
+                None,
+                true,
+                ExternalBlobMode::Reference,
+                source_store_registry,
+                source_store_params,
+                None,
+            )
+            .map(|preprocessor| preprocessor.for_mem_wal(memtable.blob_id_allocator()))
+        };
+
+        // A non-durable put becomes visible when its in-memory index apply
+        // finishes, so its sidecar must already be readable at that point.
+        // Preserve the original per-put finalization for that mode.
+        if !self.config.durable_write {
+            let mut preprocessor = new_preprocessor()?;
+
+            for batch in &batches {
+                preprocessor.validate_batch(batch).await?;
+            }
+            let prepared = async {
+                let mut prepared = Vec::with_capacity(batches.len());
+                for batch in batches {
+                    prepared.push(preprocessor.preprocess_batch(&batch).await?);
+                }
+                preprocessor.finish().await?;
+                Ok(prepared)
+            }
+            .await;
+            if prepared.is_err() {
+                preprocessor.abort();
+            }
+            return prepared.map(|batches| (batches, Some(preprocessor), false));
+        }
+
+        // Durable puts share one preprocessor per MemTable generation. Keeping
+        // its partial pack open lets sequential put_no_wait calls use one object;
+        // WalFlusher seals it before appending the corresponding descriptors.
+        let batch_store = memtable.batch_store();
+        let mut guard = batch_store.blob_preprocessor().lock().await;
+        if guard.is_none() {
+            *guard = Some(new_preprocessor()?);
+        }
+        let preprocessor = guard.as_mut().expect("Blob preprocessor initialized");
 
         for batch in &batches {
             preprocessor.validate_batch(batch).await?;
@@ -2220,14 +2256,17 @@ impl SharedWriterState {
             for batch in batches {
                 prepared.push(preprocessor.preprocess_batch(&batch).await?);
             }
-            preprocessor.finish().await?;
             Ok(prepared)
         }
         .await;
-        if prepared.is_err() {
+        if let Err(error) = &prepared {
             preprocessor.abort();
+            // The open pack may contain payloads from earlier unacknowledged
+            // puts. Once it is abandoned, none of those puts may be reported
+            // durable; fail the whole group and require replay/reopen.
+            self.wal_flusher.poison(error);
         }
-        prepared.map(|batches| (batches, Some(preprocessor)))
+        prepared.map(|batches| (batches, None, true))
     }
 
     /// Ask the index-apply task to cover `[indexed, end_batch_position)` of this
@@ -3422,7 +3461,8 @@ impl ShardWriter {
             //    payloads into the active target, then keep only prepared
             //    descriptors in the memtable and WAL.
             let batches = incoming.shape(&state.schema)?;
-            let (batches, mut blob_preprocessor) = writer_state
+            let batch_store = state.memtable.batch_store();
+            let (batches, mut blob_preprocessor, shared_blob_preprocessor) = writer_state
                 .prepare_batches(&state.memtable, &state.schema, batches)
                 .await?;
 
@@ -3432,6 +3472,16 @@ impl ShardWriter {
                 Err(error) => {
                     if let Some(preprocessor) = blob_preprocessor.as_mut() {
                         preprocessor.abort();
+                    } else if shared_blob_preprocessor {
+                        // Shared preparation may include payloads from earlier
+                        // unacknowledged puts, so abandoning it is a terminal
+                        // group-commit failure rather than a per-put rollback.
+                        if let Some(preprocessor) =
+                            batch_store.blob_preprocessor().lock().await.as_mut()
+                        {
+                            preprocessor.abort();
+                        }
+                        writer_state.wal_flusher.poison(&error);
                     }
                     return Err(error);
                 }
@@ -3442,7 +3492,6 @@ impl ShardWriter {
             //    afterwards hands the flush trigger the **new** store paired with
             //    the **old** store's end position, so the new store's watermark
             //    jumps past batches that were never appended.
-            let batch_store = state.memtable.batch_store();
             let indexes = state.memtable.indexes_arc();
 
             let start_pos = results.first().map(|(pos, _, _)| *pos).unwrap_or(0);
@@ -5323,6 +5372,61 @@ mod tests {
         paths
     }
 
+    #[tokio::test]
+    async fn test_durable_blob_puts_share_pack_until_wal_flush() {
+        let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
+        let shard_id = Uuid::new_v4();
+        let first = create_blob_v2_batch(0, &[BlobTestValue::Bytes(b"first".to_vec())]);
+        let second = create_blob_v2_batch(1, &[BlobTestValue::Bytes(b"later".to_vec())]);
+        let writer = ShardWriter::open(
+            store.clone(),
+            base_path.clone(),
+            base_uri,
+            ShardWriterConfig {
+                shard_id,
+                durable_write: true,
+                max_wal_buffer_size: usize::MAX,
+                max_wal_flush_interval: Some(Duration::from_secs(60)),
+                ..Default::default()
+            },
+            first.schema(),
+            vec![],
+        )
+        .await
+        .unwrap();
+
+        let (_, first_watcher) = writer.put_no_wait(vec![first]).await.unwrap();
+        let (_, second_watcher) = writer.put_no_wait(vec![second]).await.unwrap();
+        let fence = writer.force_seal_active().await.unwrap();
+
+        let mut first_watcher = first_watcher.unwrap();
+        let mut second_watcher = second_watcher.unwrap();
+        first_watcher.wait().await.unwrap();
+        second_watcher.wait().await.unwrap();
+        fence.wait().await.unwrap();
+
+        let tailer = WalTailer::new(store.clone(), base_path.clone(), shard_id);
+        let entry = tailer.read_entry(1).await.unwrap().unwrap();
+        assert_eq!(entry.batches.len(), 2);
+        assert!(tailer.read_entry(2).await.unwrap().is_none());
+
+        let target = entry.target.unwrap();
+        let data_dir = target.generation_path(&base_path, &shard_id).join(DATA_DIR);
+        let sidecars = blob_sidecars(store.as_ref(), &data_dir).await;
+        assert_eq!(sidecars.len(), 1, "both puts must share one packed object");
+        let payload = store
+            .inner
+            .get(&sidecars[0])
+            .await
+            .unwrap()
+            .bytes()
+            .await
+            .unwrap();
+        assert_eq!(payload.as_ref(), b"firstlater");
+
+        writer.close().await.unwrap();
+    }
+
     #[tokio::test]
     async fn test_blob_v2_target_is_reused_across_replay_and_flush() {
         let (store, base_path, base_uri, _temp_dir) = create_local_store().await;

From 7120dd8b7bb1503de4d1b6fba9303a58dd3dddbe Mon Sep 17 00:00:00 2001
From: Jack Ye <yezhaoqin@gmail.com>
Date: Mon, 5 Oct 2026 10:31:23 -0700
Subject: [PATCH 02/14] docs(mem_wal): describe generation blob allocator

---
 rust/lance/src/dataset/mem_wal/memtable.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/rust/lance/src/dataset/mem_wal/memtable.rs b/rust/lance/src/dataset/mem_wal/memtable.rs
index 33d2ef5f0c2..493d5a62c69 100644
--- a/rust/lance/src/dataset/mem_wal/memtable.rs
+++ b/rust/lance/src/dataset/mem_wal/memtable.rs
@@ -90,7 +90,7 @@ pub struct MemTable {
     /// Reserved SSTable/data-file identity for this generation.
     target: Option<MemTableDataTarget>,
 
-    /// Shared across the per-put Blob preprocessors for this target.
+    /// Shared by the generation-scoped Blob preprocessor and replay recovery.
     blob_id_allocator: BlobIdAllocator,
 
     /// Primary key bloom filter for staleness detection.

From bbf5a5d0a8f3be6d9b67be06dc9d33be08901000 Mon Sep 17 00:00:00 2001
From: Jack Ye <yezhaoqin@gmail.com>
Date: Mon, 5 Oct 2026 16:33:21 -0700
Subject: [PATCH 03/14] fix(mem_wal): fence failed blob groups

---
 .../dataset/mem_wal/memtable/batch_store.rs   | 59 ++++++++++++++++--
 rust/lance/src/dataset/mem_wal/write.rs       | 62 +++++++++++++++----
 2 files changed, 104 insertions(+), 17 deletions(-)

diff --git a/rust/lance/src/dataset/mem_wal/memtable/batch_store.rs b/rust/lance/src/dataset/mem_wal/memtable/batch_store.rs
index 3e44e3782a2..9f55a820bfb 100644
--- a/rust/lance/src/dataset/mem_wal/memtable/batch_store.rs
+++ b/rust/lance/src/dataset/mem_wal/memtable/batch_store.rs
@@ -55,6 +55,39 @@ use tokio::sync::Mutex as AsyncMutex;
 use crate::dataset::blob::BlobPreprocessor;
 use crate::dataset::mem_wal::MemTableDataTarget;
 
+#[derive(Default)]
+pub(crate) struct BlobPreprocessorState {
+    preprocessor: Option<BlobPreprocessor>,
+    failure: Option<String>,
+}
+
+impl BlobPreprocessorState {
+    pub(crate) fn get_or_insert_with(
+        &mut self,
+        create: impl FnOnce() -> crate::Result<BlobPreprocessor>,
+    ) -> crate::Result<&mut BlobPreprocessor> {
+        if let Some(failure) = &self.failure {
+            return Err(crate::Error::writer_poisoned(format!(
+                "Blob v2 pack is terminally failed: {failure}"
+            )));
+        }
+        if self.preprocessor.is_none() {
+            self.preprocessor = Some(create()?);
+        }
+        Ok(self
+            .preprocessor
+            .as_mut()
+            .expect("Blob preprocessor initialized"))
+    }
+
+    pub(crate) fn fail(&mut self, error: &crate::Error) {
+        if let Some(preprocessor) = self.preprocessor.as_mut() {
+            preprocessor.abort();
+        }
+        self.failure = Some(error.to_string());
+    }
+}
+
 /// A batch stored in the lock-free store.
 #[derive(Clone)]
 pub struct StoredBatch {
@@ -236,7 +269,7 @@ pub struct BatchStore {
 
     /// Generation-scoped Blob v2 writer. Durable puts reuse its current pack;
     /// the WAL path seals it before persisting any descriptors that reference it.
-    blob_preprocessor: AsyncMutex<Option<BlobPreprocessor>>,
+    blob_preprocessor: AsyncMutex<BlobPreprocessorState>,
 }
 
 // SAFETY: Safe to share across threads because:
@@ -306,7 +339,7 @@ impl BatchStore {
             global_offset,
             target,
             generation,
-            blob_preprocessor: AsyncMutex::new(None),
+            blob_preprocessor: AsyncMutex::new(BlobPreprocessorState::default()),
         }
     }
 
@@ -610,16 +643,32 @@ impl BatchStore {
         self.target.as_ref()
     }
 
-    pub(crate) fn blob_preprocessor(&self) -> &AsyncMutex<Option<BlobPreprocessor>> {
+    pub(crate) fn blob_preprocessor(&self) -> &AsyncMutex<BlobPreprocessorState> {
         &self.blob_preprocessor
     }
 
+    pub(crate) async fn fail_blob_pack(&self, error: &crate::Error) {
+        self.blob_preprocessor.lock().await.fail(error);
+    }
+
     /// Seal the generation's current packed sidecar before its descriptors are
     /// made durable in the WAL. The preprocessor remains reusable so later puts
     /// can start a new pack under the same generation target.
     pub(crate) async fn finish_blob_pack(&self) -> crate::Result<()> {
-        if let Some(preprocessor) = self.blob_preprocessor.lock().await.as_mut() {
-            preprocessor.finish().await?;
+        let mut state = self.blob_preprocessor.lock().await;
+        if let Some(failure) = &state.failure {
+            return Err(crate::Error::writer_poisoned(format!(
+                "cannot finalize terminally failed Blob v2 pack: {failure}"
+            )));
+        }
+        if let Some(preprocessor) = state.preprocessor.as_mut()
+            && let Err(error) = preprocessor.finish().await
+        {
+            preprocessor.abort();
+            state.failure = Some(error.to_string());
+            return Err(crate::Error::writer_poisoned(format!(
+                "failed to finalize Blob v2 pack: {error}"
+            )));
         }
         Ok(())
     }
diff --git a/rust/lance/src/dataset/mem_wal/write.rs b/rust/lance/src/dataset/mem_wal/write.rs
index e57aefbb6ce..9b197455232 100644
--- a/rust/lance/src/dataset/mem_wal/write.rs
+++ b/rust/lance/src/dataset/mem_wal/write.rs
@@ -2243,10 +2243,7 @@ impl SharedWriterState {
         // WalFlusher seals it before appending the corresponding descriptors.
         let batch_store = memtable.batch_store();
         let mut guard = batch_store.blob_preprocessor().lock().await;
-        if guard.is_none() {
-            *guard = Some(new_preprocessor()?);
-        }
-        let preprocessor = guard.as_mut().expect("Blob preprocessor initialized");
+        let preprocessor = guard.get_or_insert_with(new_preprocessor)?;
 
         for batch in &batches {
             preprocessor.validate_batch(batch).await?;
@@ -2260,10 +2257,11 @@ impl SharedWriterState {
         }
         .await;
         if let Err(error) = &prepared {
-            preprocessor.abort();
             // The open pack may contain payloads from earlier unacknowledged
-            // puts. Once it is abandoned, none of those puts may be reported
-            // durable; fail the whole group and require replay/reopen.
+            // puts. Latch a terminal pack failure so even a flush that was
+            // already queued cannot append their descriptors after the upload
+            // is abandoned.
+            guard.fail(error);
             self.wal_flusher.poison(error);
         }
         prepared.map(|batches| (batches, None, true))
@@ -3476,11 +3474,7 @@ impl ShardWriter {
                         // Shared preparation may include payloads from earlier
                         // unacknowledged puts, so abandoning it is a terminal
                         // group-commit failure rather than a per-put rollback.
-                        if let Some(preprocessor) =
-                            batch_store.blob_preprocessor().lock().await.as_mut()
-                        {
-                            preprocessor.abort();
-                        }
+                        batch_store.fail_blob_pack(&error).await;
                         writer_state.wal_flusher.poison(&error);
                     }
                     return Err(error);
@@ -5427,6 +5421,50 @@ mod tests {
         writer.close().await.unwrap();
     }
 
+    #[tokio::test]
+    async fn test_failed_shared_blob_pack_cannot_enter_wal() {
+        let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
+        let shard_id = Uuid::new_v4();
+        let batch = create_blob_v2_batch(0, &[BlobTestValue::Bytes(b"first".to_vec())]);
+        let writer = ShardWriter::open(
+            store.clone(),
+            base_path.clone(),
+            base_uri,
+            ShardWriterConfig {
+                shard_id,
+                durable_write: true,
+                max_wal_buffer_size: usize::MAX,
+                max_wal_flush_interval: Some(Duration::from_secs(60)),
+                ..Default::default()
+            },
+            batch.schema(),
+            vec![],
+        )
+        .await
+        .unwrap();
+
+        let (_, watcher) = writer.put_no_wait(vec![batch]).await.unwrap();
+        let batch_store = match &writer.mode {
+            WriterMode::MemTable { state, .. } => state.read().await.memtable.batch_store(),
+            WriterMode::WalOnly { .. } => unreachable!(),
+        };
+        batch_store
+            .fail_blob_pack(&Error::io("injected shared-pack failure"))
+            .await;
+
+        let error = writer
+            .wal_flusher
+            .flush(&WalFlushSource::BatchStore { batch_store }, 1)
+            .await
+            .unwrap_err();
+        assert!(error.to_string().contains("terminally failed Blob v2 pack"));
+
+        let mut watcher = watcher.unwrap();
+        assert!(watcher.wait().await.is_err());
+        let tailer = WalTailer::new(store, base_path, shard_id);
+        assert!(tailer.read_entry(1).await.unwrap().is_none());
+    }
+
     #[tokio::test]
     async fn test_blob_v2_target_is_reused_across_replay_and_flush() {
         let (store, base_path, base_uri, _temp_dir) = create_local_store().await;

From 9e70a800b446acbe81980411691b1e92774224a0 Mon Sep 17 00:00:00 2001
From: Jack Ye <yezhaoqin@gmail.com>
Date: Thu, 8 Oct 2026 08:50:56 -0700
Subject: [PATCH 04/14] perf(mem_wal): coalesce background flush triggers

---
 rust/lance/src/dataset/mem_wal/write.rs | 259 ++++++++++++------------
 1 file changed, 132 insertions(+), 127 deletions(-)

diff --git a/rust/lance/src/dataset/mem_wal/write.rs b/rust/lance/src/dataset/mem_wal/write.rs
index 9b197455232..e2fe3c8045b 100644
--- a/rust/lance/src/dataset/mem_wal/write.rs
+++ b/rust/lance/src/dataset/mem_wal/write.rs
@@ -1350,10 +1350,8 @@ struct WriterState {
     frozen_memtables: VecDeque<FrozenMemTable>,
     /// Flag to prevent duplicate memtable flush requests.
     flush_requested: bool,
-    /// Counter for WAL flush threshold crossings.
+    /// Current memtable's claimed WAL-size threshold crossings.
     wal_flush_trigger_count: usize,
-    /// Last time a WAL flush was triggered (for time-based flush).
-    last_wal_flush_trigger_time: u64,
 }
 
 /// Capture a point-in-time handle to one in-memory memtable (active or frozen
@@ -2327,6 +2325,7 @@ impl SharedWriterState {
         )?;
 
         let mut old_memtable = std::mem::replace(&mut state.memtable, new_memtable);
+        state.wal_flush_trigger_count = 0;
         // The outgoing memtable flushes with the indexes it was built with.
         let old_schema = std::mem::replace(&mut state.schema, next_schema);
         old_memtable.freeze(last_wal_entry_position);
@@ -2516,77 +2515,72 @@ impl SharedWriterState {
         )
     }
 
-    /// Check if WAL flush is needed and trigger if so.
+    /// Check whether the current memtable crossed a WAL size threshold.
+    ///
+    /// Time-based flushing is owned by [`WalFlushHandler::tickers`]. Keeping a
+    /// second clock here queues a fixed batch boundary after every slow remote
+    /// put, which defeats group commit. Size triggers are best-effort too: the
+    /// handler resolves the latest pending suffix when it drains the message.
+    /// Freeze and close use their own exact, completion-bearing triggers.
     ///
     /// Takes `&mut WriterState` directly since caller already holds the lock.
     fn maybe_trigger_wal_flush(&self, state: &mut WriterState) {
-        let threshold = self.config.max_wal_buffer_size;
-
-        let batch_count = state.memtable.batch_count();
         let total_bytes = state.memtable.batch_store().row_bytes();
-        let batch_store = state.memtable.batch_store();
-
-        // Check if there are any unflushed batches
-        let has_pending = batch_store.pending_wal_flush_count(self.wal_flusher.durable()) > 0;
-
-        // Check time-based trigger first
-        let time_trigger = if let Some(interval) = self.config.max_wal_flush_interval {
-            let interval_millis = interval.as_millis() as u64;
-            let last_trigger = state.last_wal_flush_trigger_time;
-            let now = now_millis();
-
-            // If last_trigger is 0, this is the first write - start the timer but don't flush
-            if last_trigger == 0 {
-                state.last_wal_flush_trigger_time = now;
-                None
-            } else {
-                let elapsed = now.saturating_sub(last_trigger);
-
-                if elapsed >= interval_millis && has_pending {
-                    state.last_wal_flush_trigger_time = now;
-                    Some(now)
-                } else {
-                    None
-                }
-            }
-        } else {
-            None
-        };
-
-        // If time trigger fired, send a flush message
-        if time_trigger.is_some() {
-            let _ = self.wal_flush_tx.send(TriggerWalFlush {
-                source: WalFlushSource::BatchStore { batch_store },
-                end_batch_position: batch_count,
-                done: None,
-            });
-            return;
-        }
-
-        // Check size-based trigger
-        if threshold == 0 {
+        if !claim_wal_size_trigger(
+            &mut state.wal_flush_trigger_count,
+            total_bytes,
+            self.config.max_wal_buffer_size,
+        ) {
             return;
         }
 
-        // Calculate how many thresholds have been crossed (1 at 10MB, 2 at 20MB, etc.)
-        let thresholds_crossed = total_bytes / threshold;
+        let _ = self.wal_flush_tx.send(TriggerWalFlush {
+            source: WalFlushSource::NextPending,
+            end_batch_position: 0,
+            done: None,
+        });
+    }
+}
 
-        // Trigger flush for each unclaimed threshold crossing
-        while state.wal_flush_trigger_count < thresholds_crossed {
-            state.wal_flush_trigger_count += 1;
-            // Update last trigger time so time-based trigger doesn't fire immediately after
-            state.last_wal_flush_trigger_time = now_millis();
+/// Claim all WAL-size thresholds crossed by the current put as one trigger.
+///
+/// A single append covers the whole pending suffix, so queueing one message per
+/// crossed threshold only creates stale boundaries under load. The counter is
+/// generation-local and is reset when the active memtable rotates.
+fn claim_wal_size_trigger(
+    claimed_thresholds: &mut usize,
+    total_bytes: usize,
+    threshold: usize,
+) -> bool {
+    if threshold == 0 {
+        return false;
+    }
+    let crossed = total_bytes / threshold;
+    if crossed <= *claimed_thresholds {
+        return false;
+    }
+    *claimed_thresholds = crossed;
+    true
+}
 
-            // Trigger WAL flush with captured batch range
-            let _ = self.wal_flush_tx.send(TriggerWalFlush {
-                source: WalFlushSource::BatchStore {
-                    batch_store: batch_store.clone(),
-                },
-                end_batch_position: batch_count,
-                done: None,
-            });
-        }
+/// Claim WAL-size thresholds for a queue whose byte count shrinks on drain.
+fn claim_pending_wal_size_trigger(
+    claimed_bytes: &mut usize,
+    pending_bytes: usize,
+    threshold: usize,
+) -> bool {
+    if threshold == 0 {
+        return false;
+    }
+    if pending_bytes < *claimed_bytes {
+        *claimed_bytes = 0;
     }
+    let crossed = (pending_bytes - *claimed_bytes) / threshold;
+    if crossed == 0 {
+        return false;
+    }
+    *claimed_bytes += crossed * threshold;
+    true
 }
 
 /// Trigger-tracking state for WAL-only mode (no MemTable).
@@ -2607,8 +2601,6 @@ struct WalOnlyTriggerState {
     /// fired. Resets to 0 when `pending_bytes` drops below it (drain
     /// happened since the last trigger).
     last_trigger_pending_bytes: usize,
-    /// Last time a WAL flush was triggered (for time-based trigger).
-    last_wal_flush_trigger_time: u64,
 }
 
 /// Per-mode state for `ShardWriter`.
@@ -3025,7 +3017,6 @@ impl ShardWriter {
             frozen_memtables: VecDeque::new(),
             flush_requested: false,
             wal_flush_trigger_count: 0,
-            last_wal_flush_trigger_time: 0,
         };
         // Seed before the first freeze: replay above may already have filled the
         // memtable, and nothing else publishes until it seals.
@@ -3598,18 +3589,9 @@ impl ShardWriter {
             .durable_write
             .then(|| self.wal_flusher.track_batch(None, 0, batch_positions.end));
 
-        // Time- and size-based triggers on the write path, for durable and
-        // non-durable puts alike — mirroring MemTable mode's
-        // `maybe_trigger_wal_flush`. The background ticker drives the append
-        // too; whichever fires first wins, and a redundant trigger is a cheap
-        // no-op because the flush snapshot/commit is idempotent.
-        self.maybe_trigger_wal_flush_wal_only(
-            state,
-            wal_flush_tx,
-            trigger,
-            batch_positions.end,
-            state.queue_bytes(),
-        );
+        // Size-based triggers run on the write path for durable and non-durable
+        // puts alike. The background ticker is the sole interval trigger.
+        self.maybe_trigger_wal_flush_wal_only(wal_flush_tx, trigger, state.queue_bytes());
 
         self.stats.record_put(start.elapsed());
 
@@ -3627,63 +3609,28 @@ impl ShardWriter {
         Ok(WriteResult { batch_positions })
     }
 
-    /// WAL-only-mode size+time trigger. Mirrors `SharedWriterState::maybe_trigger_wal_flush`
-    /// but reads its inputs from `WalOnlyState` (pending queue) instead of
-    /// the active MemTable.
+    /// WAL-only-mode size trigger. Time-based flushing is owned by the handler
+    /// ticker, as in MemTable mode.
     fn maybe_trigger_wal_flush_wal_only(
         &self,
-        state: &Arc<WalOnlyState>,
         wal_flush_tx: &mpsc::UnboundedSender<TriggerWalFlush>,
         trigger: &StdRwLock<WalOnlyTriggerState>,
-        end_batch_position: usize,
         pending_bytes: usize,
     ) {
-        let threshold = self.config.max_wal_buffer_size;
-        let has_pending = state.batch_count() > 0;
-
         let mut t = trigger.write().unwrap();
 
-        // Time-based trigger.
-        if let Some(interval) = self.config.max_wal_flush_interval {
-            let interval_millis = interval.as_millis() as u64;
-            let now = now_millis();
-            if t.last_wal_flush_trigger_time == 0 {
-                t.last_wal_flush_trigger_time = now;
-            } else {
-                let elapsed = now.saturating_sub(t.last_wal_flush_trigger_time);
-                if elapsed >= interval_millis && has_pending {
-                    t.last_wal_flush_trigger_time = now;
-                    let _ = wal_flush_tx.send(TriggerWalFlush {
-                        source: WalFlushSource::WalOnly {
-                            state: state.clone(),
-                        },
-                        end_batch_position,
-                        done: None,
-                    });
-                    return;
-                }
-            }
-        }
-
-        if threshold == 0 {
-            return;
-        }
-
-        // Size-based trigger: fire one trigger per `max_wal_buffer_size`
-        // crossed since the last time we triggered. If the pending queue
-        // shrank below the recorded baseline (a drain happened), reset the
-        // baseline first so the next crossing fires correctly.
-        if pending_bytes < t.last_trigger_pending_bytes {
-            t.last_trigger_pending_bytes = 0;
-        }
-        while pending_bytes >= t.last_trigger_pending_bytes + threshold {
-            t.last_trigger_pending_bytes += threshold;
-            t.last_wal_flush_trigger_time = now_millis();
+        // Claim every newly crossed `max_wal_buffer_size` boundary with one
+        // dynamically resolved trigger. If the pending queue shrank below the
+        // recorded baseline (a drain happened), reset the baseline first so the
+        // next crossing fires correctly.
+        if claim_pending_wal_size_trigger(
+            &mut t.last_trigger_pending_bytes,
+            pending_bytes,
+            self.config.max_wal_buffer_size,
+        ) {
             let _ = wal_flush_tx.send(TriggerWalFlush {
-                source: WalFlushSource::WalOnly {
-                    state: state.clone(),
-                },
-                end_batch_position,
+                source: WalFlushSource::NextPending,
+                end_batch_position: 0,
                 done: None,
             });
         }
@@ -9814,6 +9761,64 @@ mod tests {
         assert!(next_pending_store(frozen_list(), Arc::clone(&active), 3).is_none());
     }
 
+    #[test]
+    fn test_wal_size_trigger_coalesces_crossings() {
+        let threshold = 64;
+        let mut claimed = 0;
+
+        assert!(!claim_wal_size_trigger(&mut claimed, 63, threshold));
+        assert!(claim_wal_size_trigger(
+            &mut claimed,
+            3 * threshold + 1,
+            threshold
+        ));
+        assert_eq!(claimed, 3);
+
+        assert!(!claim_wal_size_trigger(
+            &mut claimed,
+            3 * threshold + 63,
+            threshold
+        ));
+        assert!(claim_wal_size_trigger(
+            &mut claimed,
+            4 * threshold,
+            threshold
+        ));
+        assert_eq!(claimed, 4);
+        assert!(!claim_wal_size_trigger(&mut claimed, usize::MAX, 0));
+    }
+
+    #[test]
+    fn test_pending_wal_size_trigger_coalesces_and_resets_after_drain() {
+        let threshold = 64;
+        let mut claimed_bytes = 0;
+
+        assert!(claim_pending_wal_size_trigger(
+            &mut claimed_bytes,
+            3 * threshold + 1,
+            threshold
+        ));
+        assert_eq!(claimed_bytes, 3 * threshold);
+        assert!(!claim_pending_wal_size_trigger(
+            &mut claimed_bytes,
+            3 * threshold + 63,
+            threshold
+        ));
+
+        assert!(!claim_pending_wal_size_trigger(
+            &mut claimed_bytes,
+            threshold - 1,
+            threshold
+        ));
+        assert_eq!(claimed_bytes, 0);
+        assert!(claim_pending_wal_size_trigger(
+            &mut claimed_bytes,
+            threshold,
+            threshold
+        ));
+        assert_eq!(claimed_bytes, threshold);
+    }
+
     /// A durable writer with no flush ticker cannot make progress in either
     /// mode — the ticker is the only thing that drives the WAL append the put
     /// waits on — so `open()` rejects it rather than letting a put block forever.

From aa8f80cddd1e844e6b1db9d4913e7da92e999a7f Mon Sep 17 00:00:00 2001
From: Jack Ye <yezhaoqin@gmail.com>
Date: Sat, 3 Oct 2026 18:23:31 -0700
Subject: [PATCH 05/14] fix(mem_wal): stream conditional file promotion

Route fixed-file promotion through client-streamed create-if-absent uploads so large S3 flushes avoid server-side copy timeouts without weakening writer fencing.
---
 Cargo.lock                         |  49 +++---
 Cargo.toml                         |   5 +-
 java/lance-jni/Cargo.lock          |  49 +++---
 java/lance-jni/Cargo.toml          |   5 +-
 python/Cargo.lock                  |  49 +++---
 python/Cargo.toml                  |   5 +-
 rust/lance-io/src/object_store.rs  | 254 ++++++++++++++++++++++++++---
 rust/lance-io/src/object_writer.rs |  73 ++++++---
 rust/lance/src/dataset/write.rs    |  35 +---
 9 files changed, 373 insertions(+), 151 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 7a3c5c0d91a..c498de23fa5 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1482,7 +1482,7 @@ version = "3.1.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "faf9468729b8cbcea668e36183cb69d317348c2e08e994829fb56ebfdfbaac34"
 dependencies = [
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]
@@ -2763,7 +2763,7 @@ dependencies = [
  "libc",
  "option-ext",
  "redox_users",
- "windows-sys 0.61.2",
+ "windows-sys 0.59.0",
 ]
 
 [[package]]
@@ -2906,7 +2906,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
 dependencies = [
  "libc",
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]
@@ -4215,7 +4215,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46"
 dependencies = [
  "hermit-abi",
  "libc",
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]
@@ -4527,7 +4527,7 @@ dependencies = [
  "lzma-sys",
  "mock_instant",
  "moka",
- "object_store 0.14.1",
+ "object_store 0.14.2",
  "parquet",
  "permutation",
  "pin-project",
@@ -4643,7 +4643,7 @@ dependencies = [
  "log",
  "moka",
  "num_cpus",
- "object_store 0.14.1",
+ "object_store 0.14.2",
  "pin-project",
  "proptest",
  "prost",
@@ -4788,7 +4788,7 @@ dependencies = [
  "lance-datagen",
  "lance-index",
  "lance-linalg",
- "object_store 0.14.1",
+ "object_store 0.14.2",
  "parquet",
  "rand 0.9.5",
  "tempfile",
@@ -4823,7 +4823,7 @@ dependencies = [
  "libc",
  "log",
  "num-traits",
- "object_store 0.14.1",
+ "object_store 0.14.2",
  "pretty_assertions",
  "proptest",
  "prost",
@@ -4912,7 +4912,7 @@ dependencies = [
  "log",
  "ndarray",
  "num-traits",
- "object_store 0.14.1",
+ "object_store 0.14.2",
  "prost",
  "prost-build",
  "prost-types",
@@ -4986,7 +4986,7 @@ dependencies = [
  "mock_instant",
  "mockall",
  "moka",
- "object_store 0.14.1",
+ "object_store 0.14.2",
  "object_store_opendal",
  "opendal",
  "opendal-service-goosefs",
@@ -5089,7 +5089,7 @@ dependencies = [
  "lance-namespace",
  "lance-table",
  "log",
- "object_store 0.14.1",
+ "object_store 0.14.2",
  "quick-xml 0.40.1",
  "rand 0.9.5",
  "reqwest 0.12.28",
@@ -5165,7 +5165,7 @@ dependencies = [
  "lance-select",
  "lance-testing",
  "log",
- "object_store 0.14.1",
+ "object_store 0.14.2",
  "pretty_assertions",
  "proptest",
  "prost",
@@ -5229,7 +5229,7 @@ dependencies = [
  "lance-core",
  "lance-file",
  "lance-io",
- "object_store 0.14.1",
+ "object_store 0.14.2",
  "tokio",
  "url",
 ]
@@ -5874,7 +5874,7 @@ version = "0.50.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5"
 dependencies = [
- "windows-sys 0.61.2",
+ "windows-sys 0.59.0",
 ]
 
 [[package]]
@@ -6061,13 +6061,12 @@ dependencies = [
 
 [[package]]
 name = "object_store"
-version = "0.14.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d354792e39fa5f0009e47623cf8b15b099bf9a652fa55c6f817fe28ac84fea50"
+version = "0.14.2"
+source = "git+https://github.com/apache/arrow-rs-object-store.git?rev=c10758c207c0103355f940f0c736fb8ecb157b04#c10758c207c0103355f940f0c736fb8ecb157b04"
 dependencies = [
  "async-trait",
  "aws-lc-rs",
- "base64 0.22.1",
+ "base64 0.23.0",
  "bytes",
  "chrono",
  "crc-fast",
@@ -6113,7 +6112,7 @@ dependencies = [
  "bytes",
  "chrono",
  "futures",
- "object_store 0.14.1",
+ "object_store 0.14.2",
  "opendal",
  "pin-project",
  "tokio",
@@ -7228,7 +7227,7 @@ dependencies = [
  "once_cell",
  "socket2",
  "tracing",
- "windows-sys 0.60.2",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]
@@ -7991,7 +7990,7 @@ dependencies = [
  "errno",
  "libc",
  "linux-raw-sys",
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]
@@ -8050,7 +8049,7 @@ dependencies = [
  "security-framework",
  "security-framework-sys",
  "webpki-root-certs",
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]
@@ -8923,7 +8922,7 @@ dependencies = [
  "getrandom 0.4.3",
  "once_cell",
  "rustix",
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]
@@ -9519,7 +9518,7 @@ version = "2.1.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8464ec13c3691491391d9fce00f6416c9a48e46972f72d7865688be2080192c9"
 dependencies = [
- "rand 0.10.1",
+ "rand 0.9.5",
 ]
 
 [[package]]
@@ -9969,7 +9968,7 @@ version = "0.1.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
 dependencies = [
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]
diff --git a/Cargo.toml b/Cargo.toml
index 4bc7bdfee6d..d76ce08502e 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -176,7 +176,7 @@ mock_instant = { version = "0.6.0" }
 moka = { version = "0.12", features = ["future", "sync"] }
 ndarray = { version = "0.16.1", features = ["matrixmultiply-threading"] }
 num-traits = "0.2"
-object_store = { version = "0.14.1" }
+object_store = { version = "0.14.2" }
 opendal = { version = "0.59.2" }
 object_store_opendal = { version = "0.60.2" }
 # Pin 0.2.2 so Cargo unifies past OpenDAL 0.59.2's `^0.2.1` req. 0.2.2 keeps
@@ -283,3 +283,6 @@ multiple-crate-versions = "allow"
 single_range_in_vec_init = "allow"
 large_futures = "deny"
 disallowed_macros = "deny"
+
+[patch.crates-io]
+object_store = { git = "https://github.com/apache/arrow-rs-object-store.git", rev = "c10758c207c0103355f940f0c736fb8ecb157b04" }
diff --git a/java/lance-jni/Cargo.lock b/java/lance-jni/Cargo.lock
index 840953d1c42..ea9c2f9fed3 100644
--- a/java/lance-jni/Cargo.lock
+++ b/java/lance-jni/Cargo.lock
@@ -93,7 +93,7 @@ version = "1.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc"
 dependencies = [
- "windows-sys 0.61.2",
+ "windows-sys 0.60.2",
 ]
 
 [[package]]
@@ -104,7 +104,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d"
 dependencies = [
  "anstyle",
  "once_cell_polyfill",
- "windows-sys 0.61.2",
+ "windows-sys 0.60.2",
 ]
 
 [[package]]
@@ -1222,7 +1222,7 @@ version = "3.1.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "faf9468729b8cbcea668e36183cb69d317348c2e08e994829fb56ebfdfbaac34"
 dependencies = [
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]
@@ -2330,7 +2330,7 @@ dependencies = [
  "libc",
  "option-ext",
  "redox_users",
- "windows-sys 0.61.2",
+ "windows-sys 0.60.2",
 ]
 
 [[package]]
@@ -2435,7 +2435,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
 dependencies = [
  "libc",
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]
@@ -3895,7 +3895,7 @@ dependencies = [
  "lance-tokenizer",
  "log",
  "moka",
- "object_store 0.14.1",
+ "object_store 0.14.2",
  "permutation",
  "pin-project",
  "prost",
@@ -3991,7 +3991,7 @@ dependencies = [
  "log",
  "moka",
  "num_cpus",
- "object_store 0.14.1",
+ "object_store 0.14.2",
  "pin-project",
  "prost",
  "quick_cache",
@@ -4104,7 +4104,7 @@ dependencies = [
  "lance-io",
  "log",
  "num-traits",
- "object_store 0.14.1",
+ "object_store 0.14.2",
  "prost",
  "prost-build",
  "prost-types",
@@ -4179,7 +4179,7 @@ dependencies = [
  "log",
  "ndarray",
  "num-traits",
- "object_store 0.14.1",
+ "object_store 0.14.2",
  "prost",
  "prost-build",
  "prost-types",
@@ -4243,7 +4243,7 @@ dependencies = [
  "log",
  "metrics",
  "moka",
- "object_store 0.14.1",
+ "object_store 0.14.2",
  "object_store_opendal",
  "opendal",
  "opendal-service-goosefs",
@@ -4291,7 +4291,7 @@ dependencies = [
  "log",
  "metrics",
  "metrics-util",
- "object_store 0.14.1",
+ "object_store 0.14.2",
  "prost",
  "prost-types",
  "roaring",
@@ -4351,7 +4351,7 @@ dependencies = [
  "lance-namespace",
  "lance-table",
  "log",
- "object_store 0.14.1",
+ "object_store 0.14.2",
  "rand 0.9.5",
  "reqwest 0.12.28",
  "roaring",
@@ -4413,7 +4413,7 @@ dependencies = [
  "lance-io",
  "lance-select",
  "log",
- "object_store 0.14.1",
+ "object_store 0.14.2",
  "prost",
  "prost-build",
  "prost-types",
@@ -4934,7 +4934,7 @@ version = "0.50.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5"
 dependencies = [
- "windows-sys 0.61.2",
+ "windows-sys 0.60.2",
 ]
 
 [[package]]
@@ -5096,13 +5096,12 @@ dependencies = [
 
 [[package]]
 name = "object_store"
-version = "0.14.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d354792e39fa5f0009e47623cf8b15b099bf9a652fa55c6f817fe28ac84fea50"
+version = "0.14.2"
+source = "git+https://github.com/apache/arrow-rs-object-store.git?rev=c10758c207c0103355f940f0c736fb8ecb157b04#c10758c207c0103355f940f0c736fb8ecb157b04"
 dependencies = [
  "async-trait",
  "aws-lc-rs",
- "base64 0.22.1",
+ "base64 0.23.0",
  "bytes",
  "chrono",
  "crc-fast",
@@ -5148,7 +5147,7 @@ dependencies = [
  "bytes",
  "chrono",
  "futures",
- "object_store 0.14.1",
+ "object_store 0.14.2",
  "opendal",
  "pin-project",
  "tokio",
@@ -6657,7 +6656,7 @@ dependencies = [
  "errno",
  "libc",
  "linux-raw-sys",
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]
@@ -6715,7 +6714,7 @@ dependencies = [
  "security-framework",
  "security-framework-sys",
  "webpki-root-certs",
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]
@@ -7203,7 +7202,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51"
 dependencies = [
  "libc",
- "windows-sys 0.61.2",
+ "windows-sys 0.60.2",
 ]
 
 [[package]]
@@ -7468,7 +7467,7 @@ dependencies = [
  "getrandom 0.4.3",
  "once_cell",
  "rustix",
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]
@@ -7932,7 +7931,7 @@ version = "2.1.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8464ec13c3691491391d9fce00f6416c9a48e46972f72d7865688be2080192c9"
 dependencies = [
- "rand 0.10.1",
+ "rand 0.9.5",
 ]
 
 [[package]]
@@ -8308,7 +8307,7 @@ version = "0.1.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
 dependencies = [
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]
diff --git a/java/lance-jni/Cargo.toml b/java/lance-jni/Cargo.toml
index b5248c3af02..043e36544b3 100644
--- a/java/lance-jni/Cargo.toml
+++ b/java/lance-jni/Cargo.toml
@@ -39,7 +39,7 @@ arrow-array = "58.0.0"
 arrow-schema = "58.0.0"
 datafusion = { version = "54.0.0", default-features = false }
 datafusion-common = "54.0.0"
-object_store = { version = "0.14.1" }
+object_store = { version = "0.14.2" }
 tokio = { version = "1.23", features = [
     "rt-multi-thread",
     "macros",
@@ -79,3 +79,6 @@ strip = true
 
 [lints.clippy]
 disallowed_macros = "deny"
+
+[patch.crates-io]
+object_store = { git = "https://github.com/apache/arrow-rs-object-store.git", rev = "c10758c207c0103355f940f0c736fb8ecb157b04" }
diff --git a/python/Cargo.lock b/python/Cargo.lock
index 5edb58b7353..48fdd90e8e6 100644
--- a/python/Cargo.lock
+++ b/python/Cargo.lock
@@ -108,7 +108,7 @@ version = "1.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc"
 dependencies = [
- "windows-sys 0.61.2",
+ "windows-sys 0.60.2",
 ]
 
 [[package]]
@@ -119,7 +119,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d"
 dependencies = [
  "anstyle",
  "once_cell_polyfill",
- "windows-sys 0.61.2",
+ "windows-sys 0.60.2",
 ]
 
 [[package]]
@@ -1298,7 +1298,7 @@ version = "3.1.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "faf9468729b8cbcea668e36183cb69d317348c2e08e994829fb56ebfdfbaac34"
 dependencies = [
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]
@@ -2509,7 +2509,7 @@ dependencies = [
  "libc",
  "option-ext",
  "redox_users",
- "windows-sys 0.61.2",
+ "windows-sys 0.60.2",
 ]
 
 [[package]]
@@ -2620,7 +2620,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
 dependencies = [
  "libc",
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]
@@ -4063,7 +4063,7 @@ dependencies = [
  "lance-tokenizer",
  "log",
  "moka",
- "object_store 0.14.1",
+ "object_store 0.14.2",
  "permutation",
  "pin-project",
  "prost",
@@ -4159,7 +4159,7 @@ dependencies = [
  "log",
  "moka",
  "num_cpus",
- "object_store 0.14.1",
+ "object_store 0.14.2",
  "pin-project",
  "prost",
  "quick_cache",
@@ -4289,7 +4289,7 @@ dependencies = [
  "lance-io",
  "log",
  "num-traits",
- "object_store 0.14.1",
+ "object_store 0.14.2",
  "prost",
  "prost-build",
  "prost-types",
@@ -4364,7 +4364,7 @@ dependencies = [
  "log",
  "ndarray",
  "num-traits",
- "object_store 0.14.1",
+ "object_store 0.14.2",
  "prost",
  "prost-build",
  "prost-types",
@@ -4428,7 +4428,7 @@ dependencies = [
  "log",
  "metrics",
  "moka",
- "object_store 0.14.1",
+ "object_store 0.14.2",
  "object_store_opendal",
  "opendal",
  "opendal-service-goosefs",
@@ -4498,7 +4498,7 @@ dependencies = [
  "lance-namespace",
  "lance-table",
  "log",
- "object_store 0.14.1",
+ "object_store 0.14.2",
  "rand 0.9.5",
  "reqwest 0.12.28",
  "roaring",
@@ -4562,7 +4562,7 @@ dependencies = [
  "lance-io",
  "lance-select",
  "log",
- "object_store 0.14.1",
+ "object_store 0.14.2",
  "prost",
  "prost-build",
  "prost-types",
@@ -5107,7 +5107,7 @@ version = "0.50.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5"
 dependencies = [
- "windows-sys 0.61.2",
+ "windows-sys 0.60.2",
 ]
 
 [[package]]
@@ -5269,13 +5269,12 @@ dependencies = [
 
 [[package]]
 name = "object_store"
-version = "0.14.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d354792e39fa5f0009e47623cf8b15b099bf9a652fa55c6f817fe28ac84fea50"
+version = "0.14.2"
+source = "git+https://github.com/apache/arrow-rs-object-store.git?rev=c10758c207c0103355f940f0c736fb8ecb157b04#c10758c207c0103355f940f0c736fb8ecb157b04"
 dependencies = [
  "async-trait",
  "aws-lc-rs",
- "base64 0.22.1",
+ "base64 0.23.0",
  "bytes",
  "chrono",
  "crc-fast",
@@ -5321,7 +5320,7 @@ dependencies = [
  "bytes",
  "chrono",
  "futures",
- "object_store 0.14.1",
+ "object_store 0.14.2",
  "opendal",
  "pin-project",
  "tokio",
@@ -6175,7 +6174,7 @@ dependencies = [
  "log",
  "metrics",
  "metrics-util",
- "object_store 0.14.1",
+ "object_store 0.14.2",
  "prost",
  "prost-types",
  "pyo3",
@@ -7040,7 +7039,7 @@ dependencies = [
  "errno",
  "libc",
  "linux-raw-sys",
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]
@@ -7098,7 +7097,7 @@ dependencies = [
  "security-framework",
  "security-framework-sys",
  "webpki-root-certs",
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]
@@ -7598,7 +7597,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51"
 dependencies = [
  "libc",
- "windows-sys 0.61.2",
+ "windows-sys 0.60.2",
 ]
 
 [[package]]
@@ -7903,7 +7902,7 @@ dependencies = [
  "getrandom 0.4.3",
  "once_cell",
  "rustix",
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]
@@ -8389,7 +8388,7 @@ version = "2.1.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8464ec13c3691491391d9fce00f6416c9a48e46972f72d7865688be2080192c9"
 dependencies = [
- "rand 0.10.1",
+ "rand 0.9.5",
 ]
 
 [[package]]
@@ -8765,7 +8764,7 @@ version = "0.1.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
 dependencies = [
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]
diff --git a/python/Cargo.toml b/python/Cargo.toml
index 31d9ff0666f..f77c572f649 100644
--- a/python/Cargo.toml
+++ b/python/Cargo.toml
@@ -24,7 +24,7 @@ arrow-array = "58.0.0"
 arrow-cast = "58.0.0"
 arrow-data = "58.0.0"
 arrow-schema = "58.0.0"
-object_store = "0.14.1"
+object_store = "0.14.2"
 url = "2.5.7"
 datafusion = { version = "54.0.0", default-features = false }
 datafusion-ffi = "54.0.0"
@@ -105,5 +105,8 @@ debug-assertions = false
 strip = "debuginfo"
 incremental = false
 
+[patch.crates-io]
+object_store = { git = "https://github.com/apache/arrow-rs-object-store.git", rev = "c10758c207c0103355f940f0c736fb8ecb157b04" }
+
 [lints.clippy]
 disallowed_macros = "deny"
diff --git a/rust/lance-io/src/object_store.rs b/rust/lance-io/src/object_store.rs
index 21cf127a321..9866b7bee78 100644
--- a/rust/lance-io/src/object_store.rs
+++ b/rust/lance-io/src/object_store.rs
@@ -84,6 +84,12 @@ pub const DEFAULT_CLOUD_IO_PARALLELISM: usize = 64;
 
 const SERVER_SIDE_COPY_ENABLED_ENV: &str = "LANCE_IO_SERVER_SIDE_COPY_ENABLED";
 
+#[derive(Clone, Copy)]
+enum StreamCopyMode {
+    Overwrite,
+    Create,
+}
+
 const DEFAULT_LOCAL_BLOCK_SIZE: usize = 4 * 1024; // 4KB block size
 #[cfg(any(
     feature = "aws",
@@ -1247,6 +1253,63 @@ impl ObjectStore {
         source_path: &Path,
         destination_store: &Self,
         destination_path: &Path,
+    ) -> Result<WriteResult> {
+        self.copy_via_stream_with_mode(
+            source_path,
+            destination_store,
+            destination_path,
+            StreamCopyMode::Overwrite,
+        )
+        .await
+    }
+
+    /// Copy an object by streaming its bytes into a create-if-absent destination.
+    ///
+    /// This has the same transfer and validation behavior as [`Self::copy_via_stream`],
+    /// but atomically refuses to replace an existing destination. Cloud stores use
+    /// conditional single-part or multipart uploads instead of provider-native copy. An
+    /// existing destination with the same size is treated as an idempotent success, but is
+    /// never overwritten.
+    #[instrument(
+        name = "multipart_stream_copy_if_not_exists",
+        level = "info",
+        skip(self, source_path, destination_store, destination_path),
+        fields(
+            source = %source_path,
+            destination = %destination_path,
+            source_size = Empty,
+            read_chunk_size = Empty,
+            multipart_part_size = crate::object_writer::initial_upload_size(),
+            multipart_concurrency = crate::object_writer::max_upload_parallelism(),
+            part_count = Empty,
+            bytes_transferred = Empty,
+            destination_size = Empty,
+            validation = Empty,
+            elapsed_ms = Empty,
+        ),
+        err
+    )]
+    pub async fn copy_if_not_exists_via_stream(
+        &self,
+        source_path: &Path,
+        destination_store: &Self,
+        destination_path: &Path,
+    ) -> Result<WriteResult> {
+        self.copy_via_stream_with_mode(
+            source_path,
+            destination_store,
+            destination_path,
+            StreamCopyMode::Create,
+        )
+        .await
+    }
+
+    async fn copy_via_stream_with_mode(
+        &self,
+        source_path: &Path,
+        destination_store: &Self,
+        destination_path: &Path,
+        mode: StreamCopyMode,
     ) -> Result<WriteResult> {
         let started_at = Instant::now();
         if self.has_direct_local_paths() && destination_store.has_direct_local_paths() {
@@ -1271,7 +1334,14 @@ impl ObjectStore {
             Span::current().record("source_size", source_size as u64);
 
             let metrics = destination_store.io_tracker.begin_io("copy");
-            let result = super::local::copy_file(source_path, destination_path);
+            let result = match mode {
+                StreamCopyMode::Overwrite => super::local::copy_file(source_path, destination_path),
+                StreamCopyMode::Create => destination_store
+                    .inner
+                    .copy_if_not_exists(source_path, destination_path)
+                    .await
+                    .map_err(Error::from),
+            };
             metrics.record(&result, source_size as u64);
             result.map_err(|source| {
                 stream_copy_error(
@@ -1321,17 +1391,22 @@ impl ObjectStore {
         })?;
         Span::current().record("source_size", source_size as u64);
 
-        let mut writer = destination_store
-            .create(destination_path)
-            .await
-            .map_err(|source| {
-                stream_copy_error(
-                    "destination writer creation",
-                    source_path,
-                    destination_path,
-                    source,
-                )
-            })?;
+        let mut writer: Box<dyn Writer> = match mode {
+            StreamCopyMode::Overwrite => destination_store.create(destination_path).await,
+            StreamCopyMode::Create => {
+                ObjectWriter::new_with_mode(destination_store, destination_path, PutMode::Create)
+                    .await
+                    .map(|writer| Box::new(writer) as Box<dyn Writer>)
+            }
+        }
+        .map_err(|source| {
+            stream_copy_error(
+                "destination writer creation",
+                source_path,
+                destination_path,
+                source,
+            )
+        })?;
         let read_chunk_size = usize::try_from(self.max_iop_size())
             .unwrap_or(usize::MAX)
             .max(1);
@@ -1395,14 +1470,37 @@ impl ObjectStore {
         }
         Span::current().record("bytes_transferred", bytes_transferred as u64);
 
-        let write_result = Writer::shutdown(writer.as_mut()).await.map_err(|source| {
-            stream_copy_error(
-                "destination completion",
-                source_path,
-                destination_path,
-                source,
-            )
-        })?;
+        let write_result = match Writer::shutdown(writer.as_mut()).await {
+            Ok(write_result) => write_result,
+            Err(source) => {
+                let completion_error = stream_copy_error(
+                    "destination completion",
+                    source_path,
+                    destination_path,
+                    source,
+                );
+                if !matches!(mode, StreamCopyMode::Create) {
+                    return Err(completion_error);
+                }
+                let destination_size = match destination_store.size(destination_path).await {
+                    Ok(destination_size) => destination_size,
+                    Err(_) => return Err(completion_error),
+                };
+                Span::current().record("destination_size", destination_size);
+                if destination_size != source_size as u64 {
+                    Span::current().record("validation", "failed");
+                    return Err(Error::io(format!(
+                        "multipart_stream_copy existing destination size mismatch from \
+                         {source_path} to {destination_path}: source_size={source_size}, \
+                         destination_size={destination_size}"
+                    )));
+                }
+                WriteResult {
+                    size: source_size,
+                    e_tag: None,
+                }
+            }
+        };
         if write_result.size != source_size {
             Span::current().record("validation", "failed");
             return Err(Error::io(format!(
@@ -3032,6 +3130,81 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn test_copy_if_not_exists_via_stream_preserves_existing_destination() {
+        let mut store = ObjectStore::memory();
+        store.inner = Arc::new(CopyFailingStore {
+            inner: InMemory::new(),
+        });
+
+        let source = Path::from("source.bin");
+        let destination = Path::from("destination.bin");
+        store.put(&source, b"new contents").await.unwrap();
+        store.put(&destination, b"existing contents").await.unwrap();
+
+        store
+            .copy_if_not_exists_via_stream(&source, &store, &destination)
+            .await
+            .expect_err("conditional stream copy must not replace an existing object");
+
+        assert_eq!(
+            store.read_one_all(&destination).await.unwrap().as_ref(),
+            b"existing contents"
+        );
+        assert_eq!(
+            store.read_one_all(&source).await.unwrap().as_ref(),
+            b"new contents"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_copy_if_not_exists_via_stream_creates_destination() {
+        let mut store = ObjectStore::memory();
+        store.inner = Arc::new(CopyFailingStore {
+            inner: InMemory::new(),
+        });
+
+        let source = Path::from("source.bin");
+        let destination = Path::from("destination.bin");
+        let contents = b"stream into a previously absent destination";
+        store.put(&source, contents).await.unwrap();
+
+        let result = store
+            .copy_if_not_exists_via_stream(&source, &store, &destination)
+            .await
+            .unwrap();
+
+        assert_eq!(result.size, contents.len());
+        assert_eq!(
+            store.read_one_all(&destination).await.unwrap().as_ref(),
+            contents
+        );
+    }
+
+    #[tokio::test]
+    async fn test_copy_if_not_exists_via_stream_accepts_equal_size_retry() {
+        let mut store = ObjectStore::memory();
+        store.inner = Arc::new(CopyFailingStore {
+            inner: InMemory::new(),
+        });
+
+        let source = Path::from("source.bin");
+        let destination = Path::from("destination.bin");
+        store.put(&source, b"new").await.unwrap();
+        store.put(&destination, b"old").await.unwrap();
+
+        let result = store
+            .copy_if_not_exists_via_stream(&source, &store, &destination)
+            .await
+            .unwrap();
+
+        assert_eq!(result.size, 3);
+        assert_eq!(
+            store.read_one_all(&destination).await.unwrap().as_ref(),
+            b"old"
+        );
+    }
+
     #[tokio::test]
     async fn test_remove_stream_removes_every_path() {
         let store = ObjectStore::memory();
@@ -3382,6 +3555,47 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn test_copy_if_not_exists_via_stream_uses_conditional_multipart_completion() {
+        let mut source_store = ObjectStore::memory();
+        source_store.max_iop_size = 1024 * 1024;
+        let observations = Arc::new(MultipartObservations::default());
+        let mut destination_store = ObjectStore::memory();
+        destination_store.inner = Arc::new(ObservedMultipartStore {
+            inner: InMemory::new(),
+            observations: observations.clone(),
+            fail_parts: false,
+            destination_size_adjustment: 0,
+        });
+
+        let source = Path::from("source.bin");
+        let destination = Path::from("destination.bin");
+        let contents = vec![42; crate::object_writer::initial_upload_size() * 2 + 1];
+        source_store.put(&source, &contents).await.unwrap();
+        destination_store
+            .put(&destination, b"existing contents")
+            .await
+            .unwrap();
+
+        source_store
+            .copy_if_not_exists_via_stream(&source, &destination_store, &destination)
+            .await
+            .expect_err("conditional multipart completion must reject an existing object");
+
+        assert!(
+            observations.part_count.load(Ordering::SeqCst) >= 2,
+            "conditional stream copy should use multipart upload for a large destination"
+        );
+        assert_eq!(
+            destination_store
+                .read_one_all(&destination)
+                .await
+                .unwrap()
+                .as_ref(),
+            b"existing contents"
+        );
+    }
+
     #[tokio::test]
     async fn test_copy_via_stream_aborts_failed_upload_and_retains_source() {
         let source_store = ObjectStore::memory();
diff --git a/rust/lance-io/src/object_writer.rs b/rust/lance-io/src/object_writer.rs
index c5eaffde99a..e76a2cf43e4 100644
--- a/rust/lance-io/src/object_writer.rs
+++ b/rust/lance-io/src/object_writer.rs
@@ -12,7 +12,7 @@ use async_trait::async_trait;
 use bytes::Bytes;
 use futures::FutureExt;
 use futures::future::BoxFuture;
-use object_store::{MultipartUpload, ObjectStoreExt};
+use object_store::{MultipartUpload, ObjectStoreExt, PutMode, PutMultipartOptions, PutOptions};
 use object_store::{ObjectStore, path::Path};
 use tokio::io::{AsyncWrite, AsyncWriteExt};
 use tokio::task::JoinSet;
@@ -90,6 +90,7 @@ pub struct ObjectWriter {
     path: Arc<Path>,
     cursor: usize,
     buffer: Vec<u8>,
+    put_mode: PutMode,
     // TODO: use constant size to support R2
     use_constant_size_upload_parts: bool,
 }
@@ -223,7 +224,7 @@ enum UploadState {
 
 /// Methods for state transitions.
 impl UploadState {
-    fn started_to_putting_single(&mut self, path: Arc<Path>, buffer: Vec<u8>) {
+    fn started_to_putting_single(&mut self, path: Arc<Path>, buffer: Vec<u8>, put_mode: PutMode) {
         // To get owned self, we temporarily swap with Done.
         let this = std::mem::replace(self, Self::Done(WriteResult::default()));
         *self = match this {
@@ -232,16 +233,26 @@ impl UploadState {
                 let started_at = Instant::now();
                 let fut = async move {
                     let size = buffer.len();
-                    let res = store.put(&path, buffer.into()).await.map_err(|source| {
-                        UploadFailure::new(
-                            format!(
-                                "single PUT of {path} failed after {:?} ({size} bytes, {})",
-                                started_at.elapsed(),
-                                upload_settings()
-                            ),
-                            source,
+                    let res = store
+                        .put_opts(
+                            &path,
+                            buffer.into(),
+                            PutOptions {
+                                mode: put_mode,
+                                ..Default::default()
+                            },
                         )
-                    })?;
+                        .await
+                        .map_err(|source| {
+                            UploadFailure::new(
+                                format!(
+                                    "single PUT of {path} failed after {:?} ({size} bytes, {})",
+                                    started_at.elapsed(),
+                                    upload_settings()
+                                ),
+                                source,
+                            )
+                        })?;
                     Ok(WriteResult {
                         size,
                         e_tag: res.e_tag,
@@ -291,11 +302,20 @@ impl UploadState {
 
 impl ObjectWriter {
     pub async fn new(object_store: &LanceObjectStore, path: &Path) -> Result<Self> {
+        Self::new_with_mode(object_store, path, PutMode::Overwrite).await
+    }
+
+    pub(crate) async fn new_with_mode(
+        object_store: &LanceObjectStore,
+        path: &Path,
+        put_mode: PutMode,
+    ) -> Result<Self> {
         Ok(Self {
             state: UploadState::Started(object_store.inner.clone()),
             cursor: 0,
             path: Arc::new(path.clone()),
             buffer: Vec::with_capacity(initial_upload_size()),
+            put_mode,
             use_constant_size_upload_parts: object_store.use_constant_size_upload_parts,
         })
     }
@@ -472,18 +492,28 @@ impl AsyncWrite for ObjectWriter {
                 UploadState::Started(store) => {
                     let path = mut_self.path.clone();
                     let store = store.clone();
+                    let put_mode = mut_self.put_mode.clone();
                     let started_at = Instant::now();
                     let fut = Box::pin(async move {
-                        store.put_multipart(path.as_ref()).await.map_err(|source| {
-                            UploadFailure::new(
-                                format!(
-                                    "failed to create multipart upload for {path} after {:?} ({})",
-                                    started_at.elapsed(),
-                                    upload_settings()
-                                ),
-                                source,
+                        store
+                            .put_multipart_opts(
+                                path.as_ref(),
+                                PutMultipartOptions {
+                                    mode: put_mode,
+                                    ..Default::default()
+                                },
                             )
-                        })
+                            .await
+                            .map_err(|source| {
+                                UploadFailure::new(
+                                    format!(
+                                        "failed to create multipart upload for {path} after {:?} ({})",
+                                        started_at.elapsed(),
+                                        upload_settings()
+                                    ),
+                                    source,
+                                )
+                            })
                     });
                     self.state = UploadState::CreatingUpload(fut);
                 }
@@ -570,7 +600,8 @@ impl AsyncWrite for ObjectWriter {
                     // If we didn't start a multipart upload, we can just do a single put.
                     let part = std::mem::take(&mut mut_self.buffer);
                     let path = mut_self.path.clone();
-                    self.state.started_to_putting_single(path, part);
+                    let put_mode = mut_self.put_mode.clone();
+                    self.state.started_to_putting_single(path, part, put_mode);
                 }
                 UploadState::InProgress {
                     upload,
diff --git a/rust/lance/src/dataset/write.rs b/rust/lance/src/dataset/write.rs
index 20047b11a61..7ecc1f52494 100644
--- a/rust/lance/src/dataset/write.rs
+++ b/rust/lance/src/dataset/write.rs
@@ -2109,38 +2109,9 @@ pub(in crate::dataset) struct FilePromotion {
 
 impl FilePromotion {
     async fn promote(&self) -> Result<()> {
-        match self
-            .object_store
-            .inner
-            .copy_if_not_exists(&self.staging_path, &self.final_path)
-            .await
-        {
-            Ok(()) => {}
-            Err(
-                object_store::Error::AlreadyExists { .. }
-                | object_store::Error::Precondition { .. },
-            ) => {
-                let staged = self
-                    .object_store
-                    .inner
-                    .head(&self.staging_path)
-                    .await
-                    .map_err(|error| Error::io(error.to_string()))?;
-                let existing = self
-                    .object_store
-                    .inner
-                    .head(&self.final_path)
-                    .await
-                    .map_err(|error| Error::io(error.to_string()))?;
-                if staged.size != existing.size {
-                    return Err(Error::io(format!(
-                        "fixed data file {} already exists with size {}, but the staged retry has size {}",
-                        self.final_path, existing.size, staged.size
-                    )));
-                }
-            }
-            Err(error) => return Err(Error::io(error.to_string())),
-        }
+        self.object_store
+            .copy_if_not_exists_via_stream(&self.staging_path, &self.object_store, &self.final_path)
+            .await?;
         self.object_store
             .inner
             .delete(&self.staging_path)

From 6f3a946af760c57bb261125c9871c916fca20247 Mon Sep 17 00:00:00 2001
From: Jack Ye <yezhaoqin@gmail.com>
Date: Sat, 3 Oct 2026 18:25:54 -0700
Subject: [PATCH 06/14] fix(io): abort failed multipart completion

Clean up conditional multipart uploads when destination creation is rejected.
---
 rust/lance-io/src/object_store.rs  |  5 +++++
 rust/lance-io/src/object_writer.rs | 33 ++++++++++++++++++++----------
 2 files changed, 27 insertions(+), 11 deletions(-)

diff --git a/rust/lance-io/src/object_store.rs b/rust/lance-io/src/object_store.rs
index 9866b7bee78..b896851b345 100644
--- a/rust/lance-io/src/object_store.rs
+++ b/rust/lance-io/src/object_store.rs
@@ -3586,6 +3586,11 @@ mod tests {
             observations.part_count.load(Ordering::SeqCst) >= 2,
             "conditional stream copy should use multipart upload for a large destination"
         );
+        assert_eq!(
+            observations.abort_count.load(Ordering::SeqCst),
+            1,
+            "conditional completion failure should abort the multipart upload"
+        );
         assert_eq!(
             destination_store
                 .read_one_all(&destination)
diff --git a/rust/lance-io/src/object_writer.rs b/rust/lance-io/src/object_writer.rs
index e76a2cf43e4..605011cb2be 100644
--- a/rust/lance-io/src/object_writer.rs
+++ b/rust/lance-io/src/object_writer.rs
@@ -12,7 +12,7 @@ use async_trait::async_trait;
 use bytes::Bytes;
 use futures::FutureExt;
 use futures::future::BoxFuture;
-use object_store::{MultipartUpload, ObjectStoreExt, PutMode, PutMultipartOptions, PutOptions};
+use object_store::{MultipartUpload, PutMode, PutMultipartOptions, PutOptions};
 use object_store::{ObjectStore, path::Path};
 use tokio::io::{AsyncWrite, AsyncWriteExt};
 use tokio::task::JoinSet;
@@ -277,17 +277,28 @@ impl UploadState {
                 tracing::Span::current().record("part_count", part_idx as u64);
                 let started_at = Instant::now();
                 let fut = async move {
-                    let res = upload.complete().await.map_err(|source| {
-                        UploadFailure::new(
-                            format!(
-                                "completing multipart upload of {path} failed after {:?} \
+                    let res = match upload.complete().await {
+                        Ok(result) => result,
+                        Err(source) => {
+                            let elapsed = started_at.elapsed();
+                            if let Err(error) = upload.abort().await {
+                                tracing::warn!(
+                                    path = %path,
+                                    error = %error,
+                                    "Failed to abort multipart upload after completion error"
+                                );
+                            }
+                            return Err(UploadFailure::new(
+                                format!(
+                                    "completing multipart upload of {path} failed after {:?} \
                                  ({part_idx} parts, {bytes_written} bytes, {})",
-                                started_at.elapsed(),
-                                upload_settings()
-                            ),
-                            source,
-                        )
-                    })?;
+                                    elapsed,
+                                    upload_settings()
+                                ),
+                                source,
+                            ));
+                        }
+                    };
                     Ok(WriteResult {
                         size: 0, // This will be set properly later.
                         e_tag: res.e_tag,

From 43e7acc157e1d61a9c487b2c2b57c7d8ce83b450 Mon Sep 17 00:00:00 2001
From: Jack Ye <yezhaoqin@gmail.com>
Date: Sat, 3 Oct 2026 18:27:49 -0700
Subject: [PATCH 07/14] fix(io): preserve local promotion retries

Apply the same equal-size idempotency rule to conditional local copies as cloud multipart promotion.
---
 rust/lance-io/src/object_store.rs | 33 +++++++++++++++++++++++++++++--
 1 file changed, 31 insertions(+), 2 deletions(-)

diff --git a/rust/lance-io/src/object_store.rs b/rust/lance-io/src/object_store.rs
index b896851b345..2a70e810e12 100644
--- a/rust/lance-io/src/object_store.rs
+++ b/rust/lance-io/src/object_store.rs
@@ -1336,11 +1336,17 @@ impl ObjectStore {
             let metrics = destination_store.io_tracker.begin_io("copy");
             let result = match mode {
                 StreamCopyMode::Overwrite => super::local::copy_file(source_path, destination_path),
-                StreamCopyMode::Create => destination_store
+                StreamCopyMode::Create => match destination_store
                     .inner
                     .copy_if_not_exists(source_path, destination_path)
                     .await
-                    .map_err(Error::from),
+                {
+                    Ok(()) => Ok(()),
+                    Err(copy_error) => match destination_store.size(destination_path).await {
+                        Ok(destination_size) if destination_size == source_size as u64 => Ok(()),
+                        _ => Err(Error::from(copy_error)),
+                    },
+                },
             };
             metrics.record(&result, source_size as u64);
             result.map_err(|source| {
@@ -3205,6 +3211,29 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn test_copy_if_not_exists_via_stream_accepts_equal_size_local_retry() {
+        let directory = TempStdDir::default();
+        let (store, base_path) = ObjectStore::from_uri(directory.to_str().unwrap())
+            .await
+            .unwrap();
+        let source = base_path.clone().join("source.bin");
+        let destination = base_path.join("destination.bin");
+        store.put(&source, b"new").await.unwrap();
+        store.put(&destination, b"old").await.unwrap();
+
+        let result = store
+            .copy_if_not_exists_via_stream(&source, &store, &destination)
+            .await
+            .unwrap();
+
+        assert_eq!(result.size, 3);
+        assert_eq!(
+            store.read_one_all(&destination).await.unwrap().as_ref(),
+            b"old"
+        );
+    }
+
     #[tokio::test]
     async fn test_remove_stream_removes_every_path() {
         let store = ObjectStore::memory();

From 1afe65292c7fffd66a5485435c0267bab2f79e69 Mon Sep 17 00:00:00 2001
From: Jack Ye <yezhaoqin@gmail.com>
Date: Sat, 3 Oct 2026 18:31:56 -0700
Subject: [PATCH 08/14] style(io): use let-else for copy validation

---
 rust/lance-io/src/object_store.rs | 5 ++---
 1 file changed, 2 insertions(+), 3 deletions(-)

diff --git a/rust/lance-io/src/object_store.rs b/rust/lance-io/src/object_store.rs
index 2a70e810e12..9a196cd0da7 100644
--- a/rust/lance-io/src/object_store.rs
+++ b/rust/lance-io/src/object_store.rs
@@ -1488,9 +1488,8 @@ impl ObjectStore {
                 if !matches!(mode, StreamCopyMode::Create) {
                     return Err(completion_error);
                 }
-                let destination_size = match destination_store.size(destination_path).await {
-                    Ok(destination_size) => destination_size,
-                    Err(_) => return Err(completion_error),
+                let Ok(destination_size) = destination_store.size(destination_path).await else {
+                    return Err(completion_error);
                 };
                 Span::current().record("destination_size", destination_size);
                 if destination_size != source_size as u64 {

From 3cd17abcb648524706d65fa335c2f479add28585 Mon Sep 17 00:00:00 2001
From: Jack Ye <yezhaoqin@gmail.com>
Date: Sat, 3 Oct 2026 18:49:52 -0700
Subject: [PATCH 09/14] fix(io): gate conditional streaming by backend

---
 rust/lance-io/src/object_store.rs             | 98 +++++++++++++++++--
 .../src/object_store/providers/aws.rs         |  6 ++
 .../src/object_store/providers/azure.rs       |  1 +
 .../src/object_store/providers/gcp.rs         |  1 +
 .../src/object_store/providers/goosefs.rs     |  1 +
 .../src/object_store/providers/huggingface.rs |  1 +
 .../src/object_store/providers/local.rs       |  2 +
 .../src/object_store/providers/memory.rs      |  1 +
 .../src/object_store/providers/oss.rs         |  1 +
 .../src/object_store/providers/tencent.rs     |  1 +
 .../src/object_store/providers/tos.rs         |  1 +
 rust/lance/src/dataset/write.rs               |  2 +-
 12 files changed, 106 insertions(+), 10 deletions(-)

diff --git a/rust/lance-io/src/object_store.rs b/rust/lance-io/src/object_store.rs
index 9a196cd0da7..a3f64c4880a 100644
--- a/rust/lance-io/src/object_store.rs
+++ b/rust/lance-io/src/object_store.rs
@@ -217,6 +217,8 @@ pub struct ObjectStore {
     /// Whether to use constant size upload parts for multipart uploads. This
     /// is only necessary for Cloudflare R2.
     pub use_constant_size_upload_parts: bool,
+    /// Whether multipart uploads honor [`PutMode::Create`] at completion.
+    supports_conditional_multipart_put: bool,
     /// Whether we can assume that the list of files is lexically ordered. This
     /// is true for object stores, but not for local filesystems.
     pub list_is_lexically_ordered: bool,
@@ -246,6 +248,10 @@ impl std::fmt::Debug for ObjectStore {
                 "use_constant_size_upload_parts",
                 &self.use_constant_size_upload_parts,
             )
+            .field(
+                "supports_conditional_multipart_put",
+                &self.supports_conditional_multipart_put,
+            )
             .field("list_is_lexically_ordered", &self.list_is_lexically_ordered)
             .field("io_parallelism", &self.io_parallelism)
             .field("download_retry_count", &self.download_retry_count)
@@ -694,6 +700,7 @@ impl ObjectStore {
                 block_size: params.resolved_block_size()?.unwrap_or(64 * 1024),
                 max_iop_size: *DEFAULT_MAX_IOP_SIZE,
                 use_constant_size_upload_parts: params.use_constant_size_upload_parts,
+                supports_conditional_multipart_put: false,
                 list_is_lexically_ordered: params.list_is_lexically_ordered.unwrap_or_default(),
                 io_parallelism: DEFAULT_CLOUD_IO_PARALLELISM,
                 download_retry_count: DEFAULT_DOWNLOAD_RETRY_COUNT,
@@ -1263,13 +1270,53 @@ impl ObjectStore {
         .await
     }
 
-    /// Copy an object by streaming its bytes into a create-if-absent destination.
+    /// Copy within this store without replacing an existing destination.
     ///
-    /// This has the same transfer and validation behavior as [`Self::copy_via_stream`],
-    /// but atomically refuses to replace an existing destination. Cloud stores use
-    /// conditional single-part or multipart uploads instead of provider-native copy. An
-    /// existing destination with the same size is treated as an idempotent success, but is
-    /// never overwritten.
+    /// Backends with verified conditional multipart completion stream through
+    /// the client. Other backends retain their native conditional-copy path.
+    pub async fn copy_if_not_exists(
+        &self,
+        source_path: &Path,
+        destination_path: &Path,
+    ) -> Result<WriteResult> {
+        if self.supports_conditional_multipart_put {
+            return self
+                .copy_if_not_exists_via_stream(source_path, self, destination_path)
+                .await;
+        }
+
+        let source_size = self.size(source_path).await?;
+        match self
+            .inner
+            .copy_if_not_exists(source_path, destination_path)
+            .await
+        {
+            Ok(()) => {}
+            Err(
+                object_store::Error::AlreadyExists { .. }
+                | object_store::Error::Precondition { .. },
+            ) => {
+                let destination_size = self.size(destination_path).await?;
+                if source_size != destination_size {
+                    return Err(Error::io(format!(
+                        "destination {destination_path} already exists with size \
+                         {destination_size}, but source {source_path} has size {source_size}"
+                    )));
+                }
+            }
+            Err(error) => return Err(Error::io(error.to_string())),
+        }
+
+        Ok(WriteResult {
+            size: usize::try_from(source_size).map_err(|_| {
+                Error::io(format!(
+                    "source {source_path} size {source_size} does not fit in usize"
+                ))
+            })?,
+            e_tag: None,
+        })
+    }
+
     #[instrument(
         name = "multipart_stream_copy_if_not_exists",
         level = "info",
@@ -1289,7 +1336,7 @@ impl ObjectStore {
         ),
         err
     )]
-    pub async fn copy_if_not_exists_via_stream(
+    async fn copy_if_not_exists_via_stream(
         &self,
         source_path: &Path,
         destination_store: &Self,
@@ -1967,6 +2014,7 @@ impl ObjectStore {
             block_size,
             max_iop_size: *DEFAULT_MAX_IOP_SIZE,
             use_constant_size_upload_parts,
+            supports_conditional_multipart_put: false,
             list_is_lexically_ordered,
             io_parallelism,
             download_retry_count,
@@ -3211,7 +3259,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn test_copy_if_not_exists_via_stream_accepts_equal_size_local_retry() {
+    async fn test_copy_if_not_exists_accepts_equal_size_local_retry() {
         let directory = TempStdDir::default();
         let (store, base_path) = ObjectStore::from_uri(directory.to_str().unwrap())
             .await
@@ -3222,7 +3270,7 @@ mod tests {
         store.put(&destination, b"old").await.unwrap();
 
         let result = store
-            .copy_if_not_exists_via_stream(&source, &store, &destination)
+            .copy_if_not_exists(&source, &destination)
             .await
             .unwrap();
 
@@ -3233,6 +3281,38 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn test_copy_if_not_exists_uses_native_copy_without_conditional_multipart() {
+        let observations = Arc::new(MultipartObservations::default());
+        let mut store = ObjectStore::memory();
+        store.supports_conditional_multipart_put = false;
+        store.inner = Arc::new(ObservedMultipartStore {
+            inner: InMemory::new(),
+            observations: observations.clone(),
+            fail_parts: false,
+            destination_size_adjustment: 0,
+        });
+
+        let source = Path::from("source.bin");
+        let destination = Path::from("destination.bin");
+        let contents = vec![42; crate::object_writer::initial_upload_size() + 1];
+        store.put(&source, &contents).await.unwrap();
+        observations.part_count.store(0, Ordering::SeqCst);
+
+        let result = store
+            .copy_if_not_exists(&source, &destination)
+            .await
+            .unwrap();
+
+        assert_eq!(result.size, contents.len());
+        assert_eq!(observations.native_copy_count.load(Ordering::SeqCst), 1);
+        assert_eq!(observations.part_count.load(Ordering::SeqCst), 0);
+        assert_eq!(
+            store.read_one_all(&destination).await.unwrap().as_ref(),
+            contents.as_slice()
+        );
+    }
+
     #[tokio::test]
     async fn test_remove_stream_removes_every_path() {
         let store = ObjectStore::memory();
diff --git a/rust/lance-io/src/object_store/providers/aws.rs b/rust/lance-io/src/object_store/providers/aws.rs
index 3702e6be5ae..b638a914c32 100644
--- a/rust/lance-io/src/object_store/providers/aws.rs
+++ b/rust/lance-io/src/object_store/providers/aws.rs
@@ -258,6 +258,7 @@ impl ObjectStoreProvider for AwsStoreProvider {
             block_size,
             max_iop_size: *DEFAULT_MAX_IOP_SIZE,
             use_constant_size_upload_parts,
+            supports_conditional_multipart_put: !use_opendal,
             list_is_lexically_ordered: !is_s3_express,
             io_parallelism: DEFAULT_CLOUD_IO_PARALLELISM,
             download_retry_count,
@@ -1073,6 +1074,7 @@ mod tests {
             .await
             .unwrap();
         assert_eq!(store.scheme, "s3");
+        assert!(!store.supports_conditional_multipart_put);
     }
 
     /// S3 Express ignores `start-after` and does not list in key order, but it does hand back
@@ -1103,6 +1105,10 @@ mod tests {
 
         assert!(!store.list_is_lexically_ordered);
         assert_eq!(store.paginated_lister.is_some(), paginated);
+        assert_eq!(
+            store.supports_conditional_multipart_put,
+            use_opendal == "false"
+        );
     }
 
     #[derive(Debug)]
diff --git a/rust/lance-io/src/object_store/providers/azure.rs b/rust/lance-io/src/object_store/providers/azure.rs
index 794a717b31c..a86e7e2ff27 100644
--- a/rust/lance-io/src/object_store/providers/azure.rs
+++ b/rust/lance-io/src/object_store/providers/azure.rs
@@ -300,6 +300,7 @@ impl ObjectStoreProvider for AzureBlobStoreProvider {
             block_size,
             max_iop_size: *DEFAULT_MAX_IOP_SIZE,
             use_constant_size_upload_parts: false,
+            supports_conditional_multipart_put: false,
             list_is_lexically_ordered: true,
             io_parallelism: DEFAULT_CLOUD_IO_PARALLELISM,
             download_retry_count,
diff --git a/rust/lance-io/src/object_store/providers/gcp.rs b/rust/lance-io/src/object_store/providers/gcp.rs
index e3e4e4ee81c..4874d4b022c 100644
--- a/rust/lance-io/src/object_store/providers/gcp.rs
+++ b/rust/lance-io/src/object_store/providers/gcp.rs
@@ -344,6 +344,7 @@ impl ObjectStoreProvider for GcsStoreProvider {
             block_size,
             max_iop_size: *DEFAULT_MAX_IOP_SIZE,
             use_constant_size_upload_parts: false,
+            supports_conditional_multipart_put: false,
             list_is_lexically_ordered: true,
             io_parallelism: DEFAULT_CLOUD_IO_PARALLELISM,
             download_retry_count,
diff --git a/rust/lance-io/src/object_store/providers/goosefs.rs b/rust/lance-io/src/object_store/providers/goosefs.rs
index ea13c7d38b6..59217432a30 100644
--- a/rust/lance-io/src/object_store/providers/goosefs.rs
+++ b/rust/lance-io/src/object_store/providers/goosefs.rs
@@ -489,6 +489,7 @@ impl ObjectStoreProvider for GooseFsStoreProvider {
             block_size,
             max_iop_size: *DEFAULT_MAX_IOP_SIZE,
             use_constant_size_upload_parts: params.use_constant_size_upload_parts,
+            supports_conditional_multipart_put: false,
             list_is_lexically_ordered: params.list_is_lexically_ordered.unwrap_or(false),
             io_parallelism: DEFAULT_CLOUD_IO_PARALLELISM,
             download_retry_count: storage_options.download_retry_count(),
diff --git a/rust/lance-io/src/object_store/providers/huggingface.rs b/rust/lance-io/src/object_store/providers/huggingface.rs
index dac76d130ef..5e6caf84cf8 100644
--- a/rust/lance-io/src/object_store/providers/huggingface.rs
+++ b/rust/lance-io/src/object_store/providers/huggingface.rs
@@ -226,6 +226,7 @@ impl ObjectStoreProvider for HuggingfaceStoreProvider {
             block_size,
             max_iop_size: *DEFAULT_MAX_IOP_SIZE,
             use_constant_size_upload_parts: params.use_constant_size_upload_parts,
+            supports_conditional_multipart_put: false,
             list_is_lexically_ordered: params.list_is_lexically_ordered.unwrap_or(true),
             io_parallelism: DEFAULT_CLOUD_IO_PARALLELISM,
             download_retry_count,
diff --git a/rust/lance-io/src/object_store/providers/local.rs b/rust/lance-io/src/object_store/providers/local.rs
index 4dd834c9151..03b6437fdf9 100644
--- a/rust/lance-io/src/object_store/providers/local.rs
+++ b/rust/lance-io/src/object_store/providers/local.rs
@@ -132,6 +132,7 @@ impl ObjectStoreProvider for FileStoreProvider {
             block_size,
             max_iop_size: *DEFAULT_MAX_IOP_SIZE,
             use_constant_size_upload_parts: false,
+            supports_conditional_multipart_put: false,
             list_is_lexically_ordered: false,
             io_parallelism: DEFAULT_LOCAL_IO_PARALLELISM,
             download_retry_count,
@@ -198,6 +199,7 @@ mod tests {
             block_size: DEFAULT_LOCAL_BLOCK_SIZE,
             max_iop_size: *DEFAULT_MAX_IOP_SIZE,
             use_constant_size_upload_parts: false,
+            supports_conditional_multipart_put: false,
             list_is_lexically_ordered: false,
             io_parallelism: DEFAULT_LOCAL_IO_PARALLELISM,
             download_retry_count: 0,
diff --git a/rust/lance-io/src/object_store/providers/memory.rs b/rust/lance-io/src/object_store/providers/memory.rs
index f9b4a22e2cc..7d9f33ef565 100644
--- a/rust/lance-io/src/object_store/providers/memory.rs
+++ b/rust/lance-io/src/object_store/providers/memory.rs
@@ -28,6 +28,7 @@ impl ObjectStoreProvider for MemoryStoreProvider {
             block_size,
             max_iop_size: *DEFAULT_MAX_IOP_SIZE,
             use_constant_size_upload_parts: false,
+            supports_conditional_multipart_put: true,
             list_is_lexically_ordered: true,
             io_parallelism: DEFAULT_CLOUD_IO_PARALLELISM,
             download_retry_count,
diff --git a/rust/lance-io/src/object_store/providers/oss.rs b/rust/lance-io/src/object_store/providers/oss.rs
index 0adf52db0a2..d2d07792a4b 100644
--- a/rust/lance-io/src/object_store/providers/oss.rs
+++ b/rust/lance-io/src/object_store/providers/oss.rs
@@ -140,6 +140,7 @@ impl ObjectStoreProvider for OssStoreProvider {
             block_size,
             max_iop_size: *DEFAULT_MAX_IOP_SIZE,
             use_constant_size_upload_parts: params.use_constant_size_upload_parts,
+            supports_conditional_multipart_put: false,
             list_is_lexically_ordered: params.list_is_lexically_ordered.unwrap_or(true),
             io_parallelism: DEFAULT_CLOUD_IO_PARALLELISM,
             download_retry_count: storage_options.download_retry_count(),
diff --git a/rust/lance-io/src/object_store/providers/tencent.rs b/rust/lance-io/src/object_store/providers/tencent.rs
index 9ad0a91765d..144df51793c 100644
--- a/rust/lance-io/src/object_store/providers/tencent.rs
+++ b/rust/lance-io/src/object_store/providers/tencent.rs
@@ -96,6 +96,7 @@ impl ObjectStoreProvider for TencentStoreProvider {
             block_size,
             max_iop_size: *DEFAULT_MAX_IOP_SIZE,
             use_constant_size_upload_parts: params.use_constant_size_upload_parts,
+            supports_conditional_multipart_put: false,
             list_is_lexically_ordered: params.list_is_lexically_ordered.unwrap_or(true),
             io_parallelism: DEFAULT_CLOUD_IO_PARALLELISM,
             download_retry_count: storage_options.download_retry_count(),
diff --git a/rust/lance-io/src/object_store/providers/tos.rs b/rust/lance-io/src/object_store/providers/tos.rs
index 9f558e92550..ad296e0596f 100644
--- a/rust/lance-io/src/object_store/providers/tos.rs
+++ b/rust/lance-io/src/object_store/providers/tos.rs
@@ -146,6 +146,7 @@ impl ObjectStoreProvider for TosStoreProvider {
             block_size,
             max_iop_size: *DEFAULT_MAX_IOP_SIZE,
             use_constant_size_upload_parts: params.use_constant_size_upload_parts,
+            supports_conditional_multipart_put: false,
             list_is_lexically_ordered: params.list_is_lexically_ordered.unwrap_or(true),
             io_parallelism: DEFAULT_CLOUD_IO_PARALLELISM,
             download_retry_count: storage_options.download_retry_count(),
diff --git a/rust/lance/src/dataset/write.rs b/rust/lance/src/dataset/write.rs
index 7ecc1f52494..be74d0c1aba 100644
--- a/rust/lance/src/dataset/write.rs
+++ b/rust/lance/src/dataset/write.rs
@@ -2110,7 +2110,7 @@ pub(in crate::dataset) struct FilePromotion {
 impl FilePromotion {
     async fn promote(&self) -> Result<()> {
         self.object_store
-            .copy_if_not_exists_via_stream(&self.staging_path, &self.object_store, &self.final_path)
+            .copy_if_not_exists(&self.staging_path, &self.final_path)
             .await?;
         self.object_store
             .inner

From 0691ed747773ba6f7699b6bc74f419e207682bd2 Mon Sep 17 00:00:00 2001
From: Jack Ye <yezhaoqin@gmail.com>
Date: Sat, 3 Oct 2026 18:56:15 -0700
Subject: [PATCH 10/14] fix(io): clear multipart capability for wrappers

---
 rust/lance-io/src/object_store.rs           | 1 +
 rust/lance-io/src/object_store/providers.rs | 1 +
 2 files changed, 2 insertions(+)

diff --git a/rust/lance-io/src/object_store.rs b/rust/lance-io/src/object_store.rs
index a3f64c4880a..fdf00e94343 100644
--- a/rust/lance-io/src/object_store.rs
+++ b/rust/lance-io/src/object_store.rs
@@ -863,6 +863,7 @@ impl ObjectStore {
     /// [`Self::read_dir_page`] falls back through the (already-wrapped) `inner`.
     pub fn apply_wrapper(&mut self, wrapper: &dyn WrappingObjectStore) {
         self.inner = wrapper.wrap(&self.store_prefix, self.inner.clone());
+        self.supports_conditional_multipart_put = false;
         self.paginated_lister = self
             .paginated_lister
             .take()
diff --git a/rust/lance-io/src/object_store/providers.rs b/rust/lance-io/src/object_store/providers.rs
index 4982c2b97ee..12bdc868a67 100644
--- a/rust/lance-io/src/object_store/providers.rs
+++ b/rust/lance-io/src/object_store/providers.rs
@@ -551,6 +551,7 @@ mod tests {
             .unwrap();
 
         assert_eq!(store.paginated_lister.is_some(), keep_pushdown);
+        assert!(!store.supports_conditional_multipart_put);
         // Both halves of the store are labelled with the same prefix.
         assert_eq!(
             *wrapper.prefixes.lock().unwrap(),

From dd5c80026a71d3f414ba5c901d795c44b96f443a Mon Sep 17 00:00:00 2001
From: Jack Ye <yezhaoqin@gmail.com>
Date: Thu, 8 Oct 2026 20:06:18 -0700
Subject: [PATCH 11/14] fix(mem_wal): fence cancelled blob pack operations

---
 .../dataset/mem_wal/memtable/batch_store.rs   |  46 ++++-
 rust/lance/src/dataset/mem_wal/test_util.rs   |  29 +++
 rust/lance/src/dataset/mem_wal/write.rs       | 177 ++++++++++++++++--
 3 files changed, 232 insertions(+), 20 deletions(-)

diff --git a/rust/lance/src/dataset/mem_wal/memtable/batch_store.rs b/rust/lance/src/dataset/mem_wal/memtable/batch_store.rs
index 9f55a820bfb..ee2d530dcec 100644
--- a/rust/lance/src/dataset/mem_wal/memtable/batch_store.rs
+++ b/rust/lance/src/dataset/mem_wal/memtable/batch_store.rs
@@ -62,6 +62,9 @@ pub(crate) struct BlobPreprocessorState {
 }
 
 impl BlobPreprocessorState {
+    const CANCELLED_OPERATION: &'static str =
+        "shared Blob v2 pack operation was cancelled before completion";
+
     pub(crate) fn get_or_insert_with(
         &mut self,
         create: impl FnOnce() -> crate::Result<BlobPreprocessor>,
@@ -86,6 +89,29 @@ impl BlobPreprocessorState {
         }
         self.failure = Some(error.to_string());
     }
+
+    /// Mark a shared-pack mutation as incomplete until its future returns.
+    ///
+    /// The marker deliberately precedes the first await. If the caller drops
+    /// the future, the mutex guard is released while this terminal state stays
+    /// behind, so a later WAL flush cannot mistake the abandoned pack for a
+    /// successfully finalized one.
+    pub(crate) fn begin_operation(&mut self) -> crate::Result<&mut BlobPreprocessor> {
+        if let Some(failure) = &self.failure {
+            return Err(crate::Error::writer_poisoned(format!(
+                "Blob v2 pack is terminally failed: {failure}"
+            )));
+        }
+        self.failure = Some(Self::CANCELLED_OPERATION.to_string());
+        Ok(self
+            .preprocessor
+            .as_mut()
+            .expect("Blob preprocessor initialized before shared operation"))
+    }
+
+    pub(crate) fn complete_operation(&mut self) {
+        self.failure = None;
+    }
 }
 
 /// A batch stored in the lock-free store.
@@ -661,14 +687,18 @@ impl BatchStore {
                 "cannot finalize terminally failed Blob v2 pack: {failure}"
             )));
         }
-        if let Some(preprocessor) = state.preprocessor.as_mut()
-            && let Err(error) = preprocessor.finish().await
-        {
-            preprocessor.abort();
-            state.failure = Some(error.to_string());
-            return Err(crate::Error::writer_poisoned(format!(
-                "failed to finalize Blob v2 pack: {error}"
-            )));
+        if state.preprocessor.is_none() {
+            return Ok(());
+        }
+        let result = state.begin_operation()?.finish().await;
+        match result {
+            Ok(()) => state.complete_operation(),
+            Err(error) => {
+                state.fail(&error);
+                return Err(crate::Error::writer_poisoned(format!(
+                    "failed to finalize Blob v2 pack: {error}"
+                )));
+            }
         }
         Ok(())
     }
diff --git a/rust/lance/src/dataset/mem_wal/test_util.rs b/rust/lance/src/dataset/mem_wal/test_util.rs
index abafbb7c68f..54b045ab9d9 100644
--- a/rust/lance/src/dataset/mem_wal/test_util.rs
+++ b/rust/lance/src/dataset/mem_wal/test_util.rs
@@ -35,6 +35,12 @@ pub struct FailControls {
     simulate_lost_ack: AtomicBool,
     /// WAL-entry `put_opts` attempts observed, for assertions.
     wal_put_attempts: AtomicUsize,
+    /// Remaining Blob sidecar `put_opts` calls to fail.
+    blob_put_failures: AtomicUsize,
+    /// Block Blob sidecar PUTs forever so cancellation can drop the caller.
+    block_blob_puts: AtomicBool,
+    /// Blob sidecar `put_opts` attempts observed, for assertions.
+    blob_put_attempts: AtomicUsize,
     /// Every location written through this store.
     put_paths: StdMutex<Vec<String>>,
     /// Every location read through this store.
@@ -54,6 +60,15 @@ impl FailControls {
     pub fn attempts(&self) -> usize {
         self.wal_put_attempts.load(Ordering::SeqCst)
     }
+    pub fn fail_blob_puts(&self, n: usize) {
+        self.blob_put_failures.store(n, Ordering::SeqCst);
+    }
+    pub fn set_block_blob_puts(&self, value: bool) {
+        self.block_blob_puts.store(value, Ordering::SeqCst);
+    }
+    pub fn blob_attempts(&self) -> usize {
+        self.blob_put_attempts.load(Ordering::SeqCst)
+    }
 
     /// Did any write land on a path containing `needle`? An open that resolved
     /// its store from other params never reaches this store, so a `false` here
@@ -141,6 +156,20 @@ impl OSObjectStore for FailingObjectStore {
             .lock()
             .unwrap()
             .push(location.to_string());
+        if location.as_ref().ends_with(".blob") {
+            self.controls
+                .blob_put_attempts
+                .fetch_add(1, Ordering::SeqCst);
+            if self.controls.block_blob_puts.load(Ordering::SeqCst) {
+                std::future::pending::<()>().await;
+            }
+            if self.controls.blob_put_failures.load(Ordering::SeqCst) > 0 {
+                self.controls
+                    .blob_put_failures
+                    .fetch_sub(1, Ordering::SeqCst);
+                return Err(Self::injected_error());
+            }
+        }
         if Self::is_wal_entry(location) {
             self.controls
                 .wal_put_attempts
diff --git a/rust/lance/src/dataset/mem_wal/write.rs b/rust/lance/src/dataset/mem_wal/write.rs
index e2fe3c8045b..5435f9a58b9 100644
--- a/rust/lance/src/dataset/mem_wal/write.rs
+++ b/rust/lance/src/dataset/mem_wal/write.rs
@@ -2241,11 +2241,13 @@ impl SharedWriterState {
         // WalFlusher seals it before appending the corresponding descriptors.
         let batch_store = memtable.batch_store();
         let mut guard = batch_store.blob_preprocessor().lock().await;
-        let preprocessor = guard.get_or_insert_with(new_preprocessor)?;
-
-        for batch in &batches {
-            preprocessor.validate_batch(batch).await?;
+        {
+            let preprocessor = guard.get_or_insert_with(new_preprocessor)?;
+            for batch in &batches {
+                preprocessor.validate_batch(batch).await?;
+            }
         }
+        let preprocessor = guard.begin_operation()?;
         let prepared = async {
             let mut prepared = Vec::with_capacity(batches.len());
             for batch in batches {
@@ -2254,15 +2256,21 @@ impl SharedWriterState {
             Ok(prepared)
         }
         .await;
-        if let Err(error) = &prepared {
-            // The open pack may contain payloads from earlier unacknowledged
-            // puts. Latch a terminal pack failure so even a flush that was
-            // already queued cannot append their descriptors after the upload
-            // is abandoned.
-            guard.fail(error);
-            self.wal_flusher.poison(error);
+        match prepared {
+            Ok(batches) => {
+                guard.complete_operation();
+                Ok((batches, None, true))
+            }
+            Err(error) => {
+                // The open pack may contain payloads from earlier
+                // unacknowledged puts. Latch a terminal pack failure so even a
+                // flush that was already queued cannot append their descriptors
+                // after the upload is abandoned.
+                guard.fail(&error);
+                self.wal_flusher.poison(&error);
+                Err(error)
+            }
         }
-        prepared.map(|batches| (batches, None, true))
     }
 
     /// Ask the index-apply task to cover `[indexed, end_batch_position)` of this
@@ -5412,6 +5420,151 @@ mod tests {
         assert!(tailer.read_entry(1).await.unwrap().is_none());
     }
 
+    #[tokio::test]
+    async fn gate_repro_failed_pack_must_not_enter_wal_on_retry() {
+        let (store, base_path, controls) = failing_memory_store().await;
+        let shard_id = Uuid::new_v4();
+        let batch = create_blob_v2_batch(0, &[BlobTestValue::Bytes(b"first".to_vec())]);
+        let writer = ShardWriter::open(
+            store.clone(),
+            base_path.clone(),
+            "memory://".to_string(),
+            ShardWriterConfig {
+                shard_id,
+                durable_write: true,
+                max_wal_buffer_size: usize::MAX,
+                max_wal_flush_interval: Some(Duration::from_secs(60)),
+                ..Default::default()
+            },
+            batch.schema(),
+            vec![],
+        )
+        .await
+        .unwrap();
+
+        let (_, watcher) = writer.put_no_wait(vec![batch]).await.unwrap();
+        let batch_store = match &writer.mode {
+            WriterMode::MemTable { state, .. } => state.read().await.memtable.batch_store(),
+            WriterMode::WalOnly { .. } => unreachable!(),
+        };
+        let source = WalFlushSource::BatchStore {
+            batch_store: batch_store.clone(),
+        };
+        controls.fail_blob_puts(1);
+        let first_error = writer.wal_flusher.flush(&source, 1).await.unwrap_err();
+        assert!(
+            first_error
+                .to_string()
+                .contains("failed to finalize Blob v2 pack")
+        );
+        assert!(writer.wal_flusher.check_poisoned().is_err());
+        assert!(watcher.unwrap().wait().await.is_err());
+
+        assert!(writer.wal_flusher.flush(&source, 1).await.is_err());
+        assert!(
+            WalTailer::new(store.clone(), base_path.clone(), shard_id)
+                .read_entry(1)
+                .await
+                .unwrap()
+                .is_none()
+        );
+        let data_dir = batch_store
+            .target()
+            .unwrap()
+            .generation_path(&base_path, &shard_id)
+            .join(DATA_DIR);
+        assert!(blob_sidecars(store.as_ref(), &data_dir).await.is_empty());
+    }
+
+    #[tokio::test]
+    async fn gate_repro_cancelled_rollover_must_not_ack_missing_pack() {
+        let (store, base_path, controls) = failing_memory_store().await;
+        let shard_id = Uuid::new_v4();
+        let with_small_pack = |batch: RecordBatch| {
+            let schema = batch.schema();
+            let fields = schema
+                .fields()
+                .iter()
+                .map(|field| {
+                    let mut field = field.as_ref().clone();
+                    if field.name() == "blob" {
+                        let mut metadata = field.metadata().clone();
+                        metadata.insert(
+                            lance_arrow::BLOB_PACK_FILE_SIZE_THRESHOLD_META_KEY.to_string(),
+                            "8".to_string(),
+                        );
+                        field = field.with_metadata(metadata);
+                    }
+                    field
+                })
+                .collect::<Vec<_>>();
+            RecordBatch::try_new(Arc::new(ArrowSchema::new(fields)), batch.columns().to_vec())
+                .unwrap()
+        };
+        let first = with_small_pack(create_blob_v2_batch(
+            0,
+            &[BlobTestValue::Bytes(b"first".to_vec())],
+        ));
+        let second = with_small_pack(create_blob_v2_batch(
+            1,
+            &[BlobTestValue::Bytes(b"later".to_vec())],
+        ));
+        let writer = ShardWriter::open(
+            store.clone(),
+            base_path.clone(),
+            "memory://".to_string(),
+            ShardWriterConfig {
+                shard_id,
+                durable_write: true,
+                max_wal_buffer_size: usize::MAX,
+                max_wal_flush_interval: Some(Duration::from_secs(60)),
+                ..Default::default()
+            },
+            first.schema(),
+            vec![],
+        )
+        .await
+        .unwrap();
+
+        let (_, first_watcher) = writer.put_no_wait(vec![first]).await.unwrap();
+        controls.set_block_blob_puts(true);
+        let cancelled =
+            tokio::time::timeout(Duration::from_millis(50), writer.put_no_wait(vec![second])).await;
+        assert!(cancelled.is_err());
+        assert_eq!(controls.blob_attempts(), 1);
+        controls.set_block_blob_puts(false);
+
+        let batch_store = match &writer.mode {
+            WriterMode::MemTable { state, .. } => state.read().await.memtable.batch_store(),
+            WriterMode::WalOnly { .. } => unreachable!(),
+        };
+        let error = writer
+            .wal_flusher
+            .flush(
+                &WalFlushSource::BatchStore {
+                    batch_store: batch_store.clone(),
+                },
+                1,
+            )
+            .await
+            .unwrap_err();
+        assert!(error.to_string().contains("terminally failed Blob v2 pack"));
+        assert!(first_watcher.unwrap().wait().await.is_err());
+        assert!(
+            WalTailer::new(store.clone(), base_path.clone(), shard_id)
+                .read_entry(1)
+                .await
+                .unwrap()
+                .is_none()
+        );
+        let data_dir = batch_store
+            .target()
+            .unwrap()
+            .generation_path(&base_path, &shard_id)
+            .join(DATA_DIR);
+        assert!(blob_sidecars(store.as_ref(), &data_dir).await.is_empty());
+    }
+
     #[tokio::test]
     async fn test_blob_v2_target_is_reused_across_replay_and_flush() {
         let (store, base_path, base_uri, _temp_dir) = create_local_store().await;

From 84fd5817385bcc0846fc75f9dde5e2f053b2772c Mon Sep 17 00:00:00 2001
From: Jack Ye <yezhaoqin@gmail.com>
Date: Thu, 8 Oct 2026 20:06:25 -0700
Subject: [PATCH 12/14] fix(index): satisfy Rust 1.98 chunk lint

---
 rust/lance/src/index/vector/ivf.rs | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/rust/lance/src/index/vector/ivf.rs b/rust/lance/src/index/vector/ivf.rs
index 4859e1e82a3..2526be8bc4c 100644
--- a/rust/lance/src/index/vector/ivf.rs
+++ b/rust/lance/src/index/vector/ivf.rs
@@ -7300,7 +7300,12 @@ mod tests {
             assert_eq!(centroids.value_type(), DataType::UInt8);
             let values = centroids.values().as_primitive::<UInt8Type>().values();
             assert_eq!(
-                values.chunks_exact(DIMENSION).collect::<HashSet<_>>().len(),
+                values
+                    .as_chunks::<DIMENSION>()
+                    .0
+                    .iter()
+                    .collect::<HashSet<_>>()
+                    .len(),
                 NUM_PARTITIONS
             );
         }

From 9c0dcd606d02218f1ed8a54aaa46f71c41f7817e Mon Sep 17 00:00:00 2001
From: Jack Ye <yezhaoqin@gmail.com>
Date: Thu, 8 Oct 2026 21:19:07 -0700
Subject: [PATCH 13/14] perf(mem_wal): preserve sequential flush cadence

---
 rust/lance/src/dataset/mem_wal/write.rs | 120 ++++++++++++++++++------
 1 file changed, 91 insertions(+), 29 deletions(-)

diff --git a/rust/lance/src/dataset/mem_wal/write.rs b/rust/lance/src/dataset/mem_wal/write.rs
index 5435f9a58b9..8db24deb46e 100644
--- a/rust/lance/src/dataset/mem_wal/write.rs
+++ b/rust/lance/src/dataset/mem_wal/write.rs
@@ -1352,6 +1352,9 @@ struct WriterState {
     flush_requested: bool,
     /// Current memtable's claimed WAL-size threshold crossings.
     wal_flush_trigger_count: usize,
+    /// Last write-path interval trigger. Its boundary is resolved when handled,
+    /// so this recovers sequential-write latency without defeating group commit.
+    last_wal_flush_trigger_time: u64,
 }
 
 /// Capture a point-in-time handle to one in-memory memtable (active or frozen
@@ -2523,24 +2526,34 @@ impl SharedWriterState {
         )
     }
 
-    /// Check whether the current memtable crossed a WAL size threshold.
+    /// Check whether the current memtable crossed a WAL size or time threshold.
     ///
-    /// Time-based flushing is owned by [`WalFlushHandler::tickers`]. Keeping a
-    /// second clock here queues a fixed batch boundary after every slow remote
-    /// put, which defeats group commit. Size triggers are best-effort too: the
-    /// handler resolves the latest pending suffix when it drains the message.
-    /// Freeze and close use their own exact, completion-bearing triggers.
+    /// Both signals are best-effort and resolve the latest pending suffix when
+    /// the handler drains the message. This preserves group commit while the
+    /// write-path clock prevents a sequential durable client from waiting for a
+    /// fresh ticker period after every completed append. Freeze and close use
+    /// their own exact, completion-bearing triggers.
     ///
     /// Takes `&mut WriterState` directly since caller already holds the lock.
     fn maybe_trigger_wal_flush(&self, state: &mut WriterState) {
+        let now = now_millis();
+        let time_trigger = claim_wal_time_trigger(
+            &mut state.last_wal_flush_trigger_time,
+            now,
+            self.config.max_wal_flush_interval,
+        );
         let total_bytes = state.memtable.batch_store().row_bytes();
-        if !claim_wal_size_trigger(
+        let size_trigger = claim_wal_size_trigger(
             &mut state.wal_flush_trigger_count,
             total_bytes,
             self.config.max_wal_buffer_size,
-        ) {
+        );
+        if !time_trigger && !size_trigger {
             return;
         }
+        if size_trigger {
+            state.last_wal_flush_trigger_time = now;
+        }
 
         let _ = self.wal_flush_tx.send(TriggerWalFlush {
             source: WalFlushSource::NextPending,
@@ -2550,6 +2563,29 @@ impl SharedWriterState {
     }
 }
 
+/// Claim one write-path interval trigger without fixing its flush boundary.
+fn claim_wal_time_trigger(
+    last_trigger_ms: &mut u64,
+    now_ms: u64,
+    interval: Option<Duration>,
+) -> bool {
+    let Some(interval_ms) = interval
+        .filter(|interval| !interval.is_zero())
+        .and_then(|interval| u64::try_from(interval.as_millis()).ok())
+    else {
+        return false;
+    };
+    if *last_trigger_ms == 0 {
+        *last_trigger_ms = now_ms;
+        return false;
+    }
+    if now_ms.saturating_sub(*last_trigger_ms) < interval_ms {
+        return false;
+    }
+    *last_trigger_ms = now_ms;
+    true
+}
+
 /// Claim all WAL-size thresholds crossed by the current put as one trigger.
 ///
 /// A single append covers the whole pending suffix, so queueing one message per
@@ -2609,6 +2645,8 @@ struct WalOnlyTriggerState {
     /// fired. Resets to 0 when `pending_bytes` drops below it (drain
     /// happened since the last trigger).
     last_trigger_pending_bytes: usize,
+    /// Last write-path interval trigger. See [`claim_wal_time_trigger`].
+    last_wal_flush_trigger_time: u64,
 }
 
 /// Per-mode state for `ShardWriter`.
@@ -3025,6 +3063,7 @@ impl ShardWriter {
             frozen_memtables: VecDeque::new(),
             flush_requested: false,
             wal_flush_trigger_count: 0,
+            last_wal_flush_trigger_time: 0,
         };
         // Seed before the first freeze: replay above may already have filled the
         // memtable, and nothing else publishes until it seals.
@@ -3533,18 +3572,11 @@ impl ShardWriter {
             writer_state.trigger_index_apply(batch_store, indexes, batch_positions.end)?;
         }
 
-        // The WAL append is *not* triggered here. It happens on the background
-        // ticker (and on the size trigger, and at freeze/close), which is the only
-        // way the flush interval can mean anything: while every durable put
-        // triggered its own append, the interval could add a redundant trigger but
-        // never delay or batch one.
-        //
-        // The cost is real and accepted: a single client's sequential *durable*
-        // throughput drops from ~10 writes/sec (one PUT round-trip) to roughly one
-        // per tick. That is a policy choice — the interval should mean what it
-        // says, and S3 API cost should be bounded. Latency-sensitive callers want
-        // `durable_write: false`, which now costs them durability only, not
-        // visibility.
+        // The write-path size and interval signals above name no fixed boundary:
+        // the handler resolves the pending suffix when it receives them. That
+        // lets concurrent puts group while a sequential durable client can
+        // submit promptly once an interval has elapsed. The background ticker
+        // remains the backstop when no later put arrives to emit a signal.
 
         // The watcher is returned in both modes now. A non-durable put still
         // waits — for its index apply (~ms), not for an S3 PUT (~100ms).
@@ -3597,8 +3629,9 @@ impl ShardWriter {
             .durable_write
             .then(|| self.wal_flusher.track_batch(None, 0, batch_positions.end));
 
-        // Size-based triggers run on the write path for durable and non-durable
-        // puts alike. The background ticker is the sole interval trigger.
+        // Best-effort size and interval triggers run on the write path for
+        // durable and non-durable puts alike. The background ticker remains the
+        // idle-write backstop.
         self.maybe_trigger_wal_flush_wal_only(wal_flush_tx, trigger, state.queue_bytes());
 
         self.stats.record_put(start.elapsed());
@@ -3617,8 +3650,7 @@ impl ShardWriter {
         Ok(WriteResult { batch_positions })
     }
 
-    /// WAL-only-mode size trigger. Time-based flushing is owned by the handler
-    /// ticker, as in MemTable mode.
+    /// WAL-only-mode best-effort size and interval triggers.
     fn maybe_trigger_wal_flush_wal_only(
         &self,
         wal_flush_tx: &mpsc::UnboundedSender<TriggerWalFlush>,
@@ -3626,16 +3658,26 @@ impl ShardWriter {
         pending_bytes: usize,
     ) {
         let mut t = trigger.write().unwrap();
+        let now = now_millis();
+        let time_trigger = claim_wal_time_trigger(
+            &mut t.last_wal_flush_trigger_time,
+            now,
+            self.config.max_wal_flush_interval,
+        );
 
         // Claim every newly crossed `max_wal_buffer_size` boundary with one
         // dynamically resolved trigger. If the pending queue shrank below the
         // recorded baseline (a drain happened), reset the baseline first so the
         // next crossing fires correctly.
-        if claim_pending_wal_size_trigger(
+        let size_trigger = claim_pending_wal_size_trigger(
             &mut t.last_trigger_pending_bytes,
             pending_bytes,
             self.config.max_wal_buffer_size,
-        ) {
+        );
+        if time_trigger || size_trigger {
+            if size_trigger {
+                t.last_wal_flush_trigger_time = now;
+            }
             let _ = wal_flush_tx.send(TriggerWalFlush {
                 source: WalFlushSource::NextPending,
                 end_batch_position: 0,
@@ -9941,6 +9983,26 @@ mod tests {
         assert!(!claim_wal_size_trigger(&mut claimed, usize::MAX, 0));
     }
 
+    #[test]
+    fn test_wal_time_trigger_preserves_interval_without_fixed_boundary() {
+        let interval = Some(Duration::from_millis(100));
+        let mut last = 0;
+
+        assert!(!claim_wal_time_trigger(&mut last, 1_000, interval));
+        assert_eq!(last, 1_000);
+        assert!(!claim_wal_time_trigger(&mut last, 1_099, interval));
+        assert!(claim_wal_time_trigger(&mut last, 1_100, interval));
+        assert_eq!(last, 1_100);
+        assert!(!claim_wal_time_trigger(&mut last, 1_199, interval));
+        assert!(claim_wal_time_trigger(&mut last, 1_200, interval));
+        assert!(!claim_wal_time_trigger(&mut last, 1_300, None));
+        assert!(!claim_wal_time_trigger(
+            &mut last,
+            1_300,
+            Some(Duration::ZERO)
+        ));
+    }
+
     #[test]
     fn test_pending_wal_size_trigger_coalesces_and_resets_after_drain() {
         let threshold = 64;
@@ -9972,9 +10034,9 @@ mod tests {
         assert_eq!(claimed_bytes, threshold);
     }
 
-    /// A durable writer with no flush ticker cannot make progress in either
-    /// mode — the ticker is the only thing that drives the WAL append the put
-    /// waits on — so `open()` rejects it rather than letting a put block forever.
+    /// A durable writer with no flush interval cannot guarantee progress in
+    /// either mode: a small put may not cross the size threshold, and without an
+    /// interval neither the write path nor ticker drives its WAL append.
     #[rstest]
     #[case::memtable(true)]
     #[case::wal_only(false)]

From 1979a4d65362519b8347190c31144042db038253 Mon Sep 17 00:00:00 2001
From: Jack Ye <yezhaoqin@gmail.com>
Date: Fri, 9 Oct 2026 18:31:21 -0700
Subject: [PATCH 14/14] test(mem_wal): name blob pack regressions

---
 rust/lance/src/dataset/mem_wal/write.rs | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/rust/lance/src/dataset/mem_wal/write.rs b/rust/lance/src/dataset/mem_wal/write.rs
index 8db24deb46e..5118ddd1182 100644
--- a/rust/lance/src/dataset/mem_wal/write.rs
+++ b/rust/lance/src/dataset/mem_wal/write.rs
@@ -5463,7 +5463,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn gate_repro_failed_pack_must_not_enter_wal_on_retry() {
+    async fn test_failed_shared_blob_pack_retry_cannot_enter_wal() {
         let (store, base_path, controls) = failing_memory_store().await;
         let shard_id = Uuid::new_v4();
         let batch = create_blob_v2_batch(0, &[BlobTestValue::Bytes(b"first".to_vec())]);
@@ -5519,7 +5519,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn gate_repro_cancelled_rollover_must_not_ack_missing_pack() {
+    async fn test_cancelled_shared_blob_pack_rollover_cannot_ack_missing_payload() {
         let (store, base_path, controls) = failing_memory_store().await;
         let shard_id = Uuid::new_v4();
         let with_small_pack = |batch: RecordBatch| {