From 1e7b2b2811110ed174abae2d90f4a43aacf4c92d Mon Sep 17 00:00:00 2001 From: Daniel Rammer Date: Tue, 14 Jul 2026 15:52:56 -0500 Subject: [PATCH 1/4] feat(mem_wal): append the WAL on a background ticker, resolved by cursor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `flush_interval_ms` was inert. It routed to a timer that was only ever evaluated *on the write path*, so it could add a redundant trigger but never delay or batch one — and since every durable put triggered its own append, tuning the knob did nothing at all. Give the WAL flusher a real ticker (`MessageHandler::tickers`, which the memtable flusher already used) and take the append off the put path. The append is the only thing on that schedule: it is an S3 PUT, billed per call, and bounding that cost is the entire reason a flush interval exists. The index apply is not on it — it is in-memory and free to batch, so it stays per-put on its own task. The tick names no store. `MessageFactory` is synchronous and cannot take the async state lock, and capturing an `Arc` at handler construction would pin the first memtable forever. So `WalFlushSource::NextPending` is resolved when the message is *handled* — by walking the live stores oldest-first and taking the first that still owes an append. Oldest-first, not "the active memtable", and this is load-bearing. WAL entry positions are assigned in append-call order; replay walks them ascending; row positions follow; and primary-key recency is "newest visible row position wins". So append order *is* dedup order. A tick enqueued before a freeze is handled after it, and resolving to the active memtable would append the incoming memtable's batches ahead of the outgoing one's tail — silently handing the key to the stale row on the next replay. It survives the crash that caused it, and a full scan cannot see it. Selecting by cursor makes the target a function of what is durable rather than of when the timer fired. Two starvation fixes in the dispatcher, both of which this makes reachable: - The interval used the default `MissedTickBehavior::Burst`, which replays every tick missed while `handle()` ran. A WAL append easily outlasts its own interval, so missed ticks accumulate, the ticker arm is always ready, and — being `biased` — it starves the channel. Now `Delay`. - The `biased` select polled the ticker *before* `rx.recv()`. A tick only ever adds an append a real trigger would have made anyway, whereas a message may be a freeze's completion cell or `close()`'s final append, which nothing else delivers. Messages now win. `durable_write: true` with no ticker is rejected at open: the put path no longer triggers its own append, so such a writer would block forever on an append that never comes. Better a clear error than a deadlock. Accepted cost: single-client sequential *durable* throughput drops from ~10 writes/sec (one PUT round-trip) to roughly one per tick. That is the policy choice — the interval should mean what it says, and API cost should be bounded. Latency-sensitive callers want `durable_write: false`, which now costs them durability only, not visibility. Co-Authored-By: Claude Opus 4.8 (1M context) --- rust/lance/src/dataset/mem_wal/wal.rs | 22 ++ rust/lance/src/dataset/mem_wal/write.rs | 356 +++++++++++++++++++++--- 2 files changed, 333 insertions(+), 45 deletions(-) diff --git a/rust/lance/src/dataset/mem_wal/wal.rs b/rust/lance/src/dataset/mem_wal/wal.rs index 5fe7eeed907..efce7b53d0a 100644 --- a/rust/lance/src/dataset/mem_wal/wal.rs +++ b/rust/lance/src/dataset/mem_wal/wal.rs @@ -317,6 +317,22 @@ pub enum WalFlushSource { /// `BatchStore` to the WAL. Append-only — the index apply runs on its own /// task, so a failed append can no longer publish rows through it. BatchStore { batch_store: Arc }, + /// Timer-driven: append whichever store still owes the WAL an append, + /// resolved when the message is *handled*, not when it is enqueued. + /// + /// Carries no store on purpose. `MessageFactory` is synchronous and cannot + /// take the async state lock, and capturing an `Arc` once at + /// handler construction would pin the first memtable forever. + /// + /// Resolution walks the live stores **oldest first** and takes the first with + /// `global_end() > durable`. It must not simply take "the active memtable": + /// a tick enqueued before a freeze is handled *after* it, would resolve to + /// the new memtable, and would append its batches ahead of the outgoing + /// memtable's tail. WAL entry positions are assigned in append-call order, + /// replay walks them ascending, row positions follow, and primary-key recency + /// is "newest visible row position wins" — so an out-of-order append silently + /// inverts dedup after a crash: the stale row wins. A full scan looks fine. + NextPending, /// WAL-only mode: drain all pending batches from the shared /// `WalOnlyState`. There are no in-memory indexes to update. WalOnly { state: Arc }, @@ -327,6 +343,7 @@ impl WalFlushSource { match self { Self::BatchStore { .. } => "BatchStore", Self::WalOnly { .. } => "WalOnly", + Self::NextPending => "NextPending", } } } @@ -773,6 +790,11 @@ impl WalFlusher { .await } WalFlushSource::WalOnly { state } => self.flush_from_wal_only(state).await, + // The handler resolves a tick to a concrete store before it gets + // here, because only it can take the async state lock. + WalFlushSource::NextPending => Err(Error::internal( + "WalFlushSource::NextPending must be resolved by the flush handler", + )), }; // A terminal failure means the watermark can never advance; latch the // poison so waiters wake with the typed error and later writes fail fast. diff --git a/rust/lance/src/dataset/mem_wal/write.rs b/rust/lance/src/dataset/mem_wal/write.rs index e03c801af63..63ff9ff9031 100644 --- a/rust/lance/src/dataset/mem_wal/write.rs +++ b/rust/lance/src/dataset/mem_wal/write.rs @@ -468,7 +468,13 @@ impl TaskDispatcher { let mut ticker_intervals: Vec<(Interval, MessageFactory)> = tickers .into_iter() .map(|(duration, factory)| { - let interval = interval_at(tokio::time::Instant::now() + duration, duration); + let mut interval = interval_at(tokio::time::Instant::now() + duration, duration); + // `Burst` (the default) replays every tick missed while `handle()` + // was running. A WAL append can easily outlast its own interval, so + // the missed ticks pile up, the ticker arm below is always ready, + // and — being `biased` — it starves `rx` indefinitely: freeze + // completion cells and `close()`'s final append never get handled. + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); (interval, factory) }) .collect(); @@ -511,12 +517,12 @@ impl TaskDispatcher { debug!("Task '{}' received cancellation", self.name); break Ok(()); } - _ = first_interval.tick() => { - let message = (ticker_intervals[0].1)(); - if let Err(e) = self.handler.handle(message).await { - error!("Task '{}' error handling ticker message: {}", self.name, e); - } - } + // Explicit messages outrank the ticker. A tick is a backstop + // — it only ever *adds* an append that a real trigger would + // have made anyway — whereas a message may be a freeze's + // completion cell or `close()`'s final append, which nothing + // else will deliver. Polling the ticker first (as this did) + // lets a ready tick starve them. msg = self.rx.recv() => { match msg { Some(message) => { @@ -530,6 +536,12 @@ impl TaskDispatcher { } } } + _ = first_interval.tick() => { + let message = (ticker_intervals[0].1)(); + if let Err(e) = self.handler.handle(message).await { + error!("Task '{}' error handling ticker message: {}", self.name, e); + } + } } } }; @@ -1820,8 +1832,12 @@ impl ShardWriter { let backpressure = BackpressureController::new(config.clone()); // Background WAL flush handler — parallel WAL I/O + index updates. - let wal_handler = - WalFlushHandler::new(wal_flusher.clone(), Some(state.clone()), stats.clone()); + let wal_handler = WalFlushHandler::new( + wal_flusher.clone(), + Some(state.clone()), + config.max_wal_flush_interval, + stats.clone(), + ); task_executor.add_handler( "wal_flusher".to_string(), Box::new(wal_handler), @@ -1893,7 +1909,7 @@ impl ShardWriter { ) -> Result { // Background WAL flush handler — no MemTable state to consult, so // pass `None` for the frozen-vs-active detection. - let wal_handler = WalFlushHandler::new(wal_flusher, None, stats); + let wal_handler = WalFlushHandler::new(wal_flusher, None, None, stats); task_executor.add_handler( "wal_flusher".to_string(), Box::new(wal_handler), @@ -2196,23 +2212,22 @@ impl ShardWriter { // (in-memory, ~ms), so there is no reason to batch it onto the WAL's // schedule — that schedule exists to bound S3 API cost, which an // in-memory index apply does not incur. - if let Some(indexes) = &indexes { - writer_state.trigger_index_apply( - batch_store.clone(), - indexes.clone(), - batch_positions.end, - )?; + if let Some(indexes) = indexes { + writer_state.trigger_index_apply(batch_store, indexes, batch_positions.end)?; } - // Trigger the WAL flush here (outside the lock) so the watcher can - // resolve; only the `wait()` is the caller's to schedule. - if self.config.durable_write { - self.wal_flusher.trigger_flush( - WalFlushSource::BatchStore { batch_store }, - batch_positions.end, - None, - )?; - } + // 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 watcher is returned in both modes now. A non-durable put still // waits — for its index apply (~ms), not for an S3 PUT (~100ms). @@ -2825,10 +2840,28 @@ pub struct WalStats { pub next_wal_entry_position: u64, } -/// Background handler for WAL flush operations. +/// The oldest store that still owes the WAL an append, or `None` when everything +/// is durable. /// -/// This handler does parallel WAL I/O + index updates during flush. -/// Indexes are passed through the TriggerWalFlush message. +/// Ordering is the whole point. WAL entry positions are assigned in append-call +/// order; replay walks them ascending; row positions follow; and primary-key +/// recency is "newest visible row position wins". So appending a newer memtable +/// ahead of an older one's tail silently inverts dedup after a crash. +/// +/// Taking "the active memtable" would do exactly that, because a timer tick +/// enqueued before a freeze is handled after it and would resolve to the incoming +/// memtable. Selecting by cursor instead makes the target a function of what is +/// actually durable, not of when the timer happened to fire. +fn next_pending_store( + frozen: impl Iterator>, + active: Arc, + durable: usize, +) -> Option> { + frozen + .chain(std::iter::once(active)) + .find(|store| store.global_end() > durable) +} + /// The index-apply task: one sequential consumer of the index-apply channel. /// /// Sequential consumption is the safety property. `HnswGraph::insert_batch` hard- @@ -2874,6 +2907,9 @@ struct WalFlushHandler { /// via Arc::ptr_eq on the active batch_store. `None` when running in /// WAL-only mode (no MemTable, no frozen-vs-active distinction). memtable_state: Option>>, + /// How often to append in the background. `None` disables the ticker, leaving + /// the append size-triggered (and freeze/close-triggered) only. + flush_interval: Option, stats: SharedWriteStats, } @@ -2881,11 +2917,13 @@ impl WalFlushHandler { fn new( wal_flusher: Arc, memtable_state: Option>>, + flush_interval: Option, stats: SharedWriteStats, ) -> Self { Self { wal_flusher, memtable_state, + flush_interval, stats, } } @@ -2893,6 +2931,33 @@ impl WalFlushHandler { #[async_trait] impl MessageHandler for WalFlushHandler { + /// Append periodically in the background. + /// + /// This is what the flush interval was always supposed to mean. It routed to + /// a timer that was only ever *evaluated on the write path*, so it could add + /// a redundant trigger but never delay or batch one — with every durable put + /// triggering its own append, tuning the knob did nothing at all. + /// + /// The ticker exists to bound S3 API cost: an append is a PUT, billed per + /// call, and it is the only thing on this schedule. The index apply is not — + /// it is in-memory and free to batch, so it runs per-put on its own task. + fn tickers(&mut self) -> Vec<(Duration, MessageFactory)> { + // No interval => no ticker. A zero interval would panic in tokio. + let Some(interval) = self.flush_interval.filter(|d| !d.is_zero()) else { + return vec![]; + }; + // The tick names no store: `MessageFactory` is synchronous and cannot take + // the async state lock. `handle()` resolves it against the cursor. + vec![( + interval, + Box::new(|| TriggerWalFlush { + source: WalFlushSource::NextPending, + end_batch_position: 0, + done: None, + }), + )] + } + async fn handle(&mut self, message: TriggerWalFlush) -> Result<()> { let TriggerWalFlush { source, @@ -2900,6 +2965,16 @@ impl MessageHandler for WalFlushHandler { done, } = message; + // A timer tick names no store — resolve it now, at handle time. + let (source, end_batch_position) = match source { + WalFlushSource::NextPending => match self.resolve_next_pending().await { + Some(resolved) => resolved, + // Everything is already durable; the tick has nothing to do. + None => return Ok(()), + }, + other => (other, end_batch_position), + }; + let result = self.do_flush(source, end_batch_position).await; // Propagate the just-appended WAL entry position back into the @@ -2930,6 +3005,44 @@ impl MessageHandler for WalFlushHandler { } impl WalFlushHandler { + /// Pick the store the WAL still owes an append, oldest first. + /// + /// **WAL entries must be appended in global batch-position order for the + /// writer's lifetime.** `WalAppender::append` assigns each entry's position + /// from its own counter, in call order; replay walks those positions + /// ascending and assigns row positions in that order; and primary-key recency + /// is "newest visible row position wins". So append order fixes dedup order. + /// Append two memtables out of order and replay silently hands the dedup to + /// the *stale* row — corruption that survives the crash that caused it, and + /// that a full scan cannot see. + /// + /// Resolving to "the active memtable" would break exactly that: a tick + /// enqueued before a freeze is handled after it, resolves to the incoming + /// memtable, and appends its batches ahead of the outgoing memtable's tail. + /// So the target is a function of `durable`, not of when the timer fired. + /// + /// Safe to walk the frozen list because a store that still owes an append + /// cannot be swept: its L0 flush is blocked on the completion cell that only + /// that append fires. + async fn resolve_next_pending(&self) -> Option<(WalFlushSource, usize)> { + let state_lock = self.memtable_state.as_ref()?; + let state = state_lock.read().await; + let durable = self.wal_flusher.durable(); + + next_pending_store( + state + .frozen_memtables + .iter() + .map(|frozen| frozen.memtable.batch_store()), + state.memtable.batch_store(), + durable, + ) + .map(|store| { + let end = store.len(); + (WalFlushSource::BatchStore { batch_store: store }, end) + }) + } + /// Unified flush method for both active and frozen memtables and for /// WAL-only mode. /// @@ -4205,7 +4318,7 @@ mod tests { durable_write: false, sync_indexed_write: false, max_wal_buffer_size: 1024 * 1024, - max_wal_flush_interval: None, + max_wal_flush_interval: Some(Duration::from_millis(10)), max_memtable_size: 64 * 1024 * 1024, manifest_scan_batch_size: 2, ..Default::default() @@ -4248,7 +4361,7 @@ mod tests { durable_write: true, sync_indexed_write: false, max_wal_buffer_size: 1024 * 1024, - max_wal_flush_interval: None, + max_wal_flush_interval: Some(Duration::from_millis(10)), max_memtable_size: 64 * 1024 * 1024, manifest_scan_batch_size: 2, ..Default::default() @@ -4285,7 +4398,7 @@ mod tests { durable_write: false, sync_indexed_write: false, max_wal_buffer_size: 1024 * 1024, - max_wal_flush_interval: None, + max_wal_flush_interval: Some(Duration::from_millis(10)), max_memtable_size: 64 * 1024 * 1024, manifest_scan_batch_size: 2, ..Default::default() @@ -4331,7 +4444,7 @@ mod tests { durable_write: false, sync_indexed_write: false, max_wal_buffer_size: 1024 * 1024, - max_wal_flush_interval: None, + max_wal_flush_interval: Some(Duration::from_millis(10)), max_memtable_size: 64 * 1024 * 1024, manifest_scan_batch_size: 2, ..Default::default() @@ -4366,7 +4479,7 @@ mod tests { durable_write: false, sync_indexed_write: true, max_wal_buffer_size: 1024 * 1024, - max_wal_flush_interval: None, + max_wal_flush_interval: Some(Duration::from_millis(10)), max_memtable_size: 64 * 1024 * 1024, manifest_scan_batch_size: 2, ..Default::default() @@ -4418,7 +4531,7 @@ mod tests { durable_write: false, sync_indexed_write: true, max_wal_buffer_size: 1024 * 1024, - max_wal_flush_interval: None, + max_wal_flush_interval: Some(Duration::from_millis(10)), max_memtable_size: 64 * 1024 * 1024, manifest_scan_batch_size: 2, ..Default::default() @@ -4507,7 +4620,7 @@ mod tests { durable_write: false, sync_indexed_write: false, max_wal_buffer_size: 1024 * 1024, - max_wal_flush_interval: None, + max_wal_flush_interval: Some(Duration::from_millis(10)), max_memtable_size: 1024, // Very small - will trigger flush quickly manifest_scan_batch_size: 2, ..Default::default() @@ -4614,7 +4727,7 @@ mod tests { durable_write: true, sync_indexed_write: false, max_wal_buffer_size: 1024 * 1024, - max_wal_flush_interval: None, + max_wal_flush_interval: Some(Duration::from_millis(10)), max_memtable_size: 64, manifest_scan_batch_size: 2, ..Default::default() @@ -4659,7 +4772,7 @@ mod tests { durable_write: true, sync_indexed_write: false, max_wal_buffer_size: 1024 * 1024, - max_wal_flush_interval: None, + max_wal_flush_interval: Some(Duration::from_millis(10)), // Tiny size threshold — every batch crosses it. max_memtable_size: 64, manifest_scan_batch_size: 2, @@ -4726,7 +4839,7 @@ mod tests { durable_write: true, sync_indexed_write: false, max_wal_buffer_size: 1024 * 1024, - max_wal_flush_interval: None, + max_wal_flush_interval: Some(Duration::from_millis(10)), max_memtable_size: usize::MAX, max_unflushed_memtable_bytes: usize::MAX, manifest_scan_batch_size: 2, @@ -4857,7 +4970,7 @@ mod tests { durable_write: true, sync_indexed_write: false, max_wal_buffer_size: 1024 * 1024, - max_wal_flush_interval: None, + max_wal_flush_interval: Some(Duration::from_millis(10)), // Tiny size threshold so a few batches cross it. max_memtable_size: 1024, manifest_scan_batch_size: 2, @@ -4998,7 +5111,7 @@ mod tests { durable_write: true, enable_memtable: false, max_wal_buffer_size: 1024 * 1024, - max_wal_flush_interval: None, + max_wal_flush_interval: Some(Duration::from_millis(10)), manifest_scan_batch_size: 2, ..Default::default() } @@ -5291,7 +5404,7 @@ mod tests { durable_write: true, sync_indexed_write: false, max_wal_buffer_size: 1024 * 1024, - max_wal_flush_interval: None, + max_wal_flush_interval: Some(Duration::from_millis(10)), max_memtable_size: 64 * 1024 * 1024, manifest_scan_batch_size: 2, ..Default::default() @@ -5902,6 +6015,159 @@ mod tests { ); } + /// A background tick must append the **oldest** store that still owes the WAL, + /// never "whatever memtable is active". + /// + /// A tick carries no store: it is enqueued by a timer and resolved when it is + /// handled. So a tick enqueued before a freeze is handled *after* it, and + /// resolving to the active memtable would append the incoming memtable's + /// batches ahead of the outgoing memtable's tail. WAL entry positions are + /// assigned in append-call order, replay walks them ascending, row positions + /// follow, and primary-key recency is "newest visible row position wins" — so + /// that inverts dedup after a crash, handing the key to the stale row. It + /// survives the crash that caused it, and a full scan cannot see it. + #[test] + fn test_next_pending_store_picks_the_oldest_owing_an_append() { + let schema = create_test_schema(); + + // A frozen store of 2 batches at coordinate 0, and the active store that + // rotated in behind it at coordinate 2. + let frozen = Arc::new(BatchStore::with_capacity(4)); + frozen.append(create_test_batch(&schema, 0, 1)).unwrap(); + frozen.append(create_test_batch(&schema, 1, 1)).unwrap(); + let active = Arc::new(BatchStore::with_capacity_at(4, 2)); + active.append(create_test_batch(&schema, 2, 1)).unwrap(); + + let frozen_list = || std::iter::once(Arc::clone(&frozen)); + + // Nothing durable: the frozen store owes the oldest append, so it wins — + // even though the active memtable also has un-appended batches. + let picked = next_pending_store(frozen_list(), Arc::clone(&active), 0).unwrap(); + assert!( + Arc::ptr_eq(&picked, &frozen), + "the outgoing memtable's tail must be appended before the incoming one's head" + ); + + // Still true partway through the frozen store. + let picked = next_pending_store(frozen_list(), Arc::clone(&active), 1).unwrap(); + assert!(Arc::ptr_eq(&picked, &frozen)); + + // Once the frozen store is fully durable, the active one is next. + let picked = next_pending_store(frozen_list(), Arc::clone(&active), 2).unwrap(); + assert!(Arc::ptr_eq(&picked, &active)); + + // Everything durable: nothing to do. + assert!(next_pending_store(frozen_list(), Arc::clone(&active), 3).is_none()); + } + + /// A durable writer with no flush ticker cannot make progress, so `open()` + /// rejects it rather than letting a put block forever. + #[tokio::test] + async fn test_open_rejects_durable_write_without_a_ticker() { + let (store, base_path, base_uri, _temp_dir) = create_local_store().await; + let schema = schema_with_pk(); + let shard_id = Uuid::new_v4(); + + let config = ShardWriterConfig { + durable_write: true, + max_wal_flush_interval: None, + ..memtable_config_with_pk(shard_id) + }; + + let Err(err) = ShardWriter::open(store, base_path, base_uri, config, schema, vec![]).await + else { + panic!("durable_write with no ticker must be rejected"); + }; + assert!( + err.to_string().contains("max_wal_flush_interval"), + "the error must name the knob, got: {err}" + ); + } + + /// WAL entries must be appended in global batch-position order across a + /// memtable rotation, because append order *is* primary-key recency order. + /// + /// `WalAppender::append` assigns each entry's position from its own counter, + /// in call order. Replay walks those positions ascending and assigns row + /// positions in that order. Primary-key recency is "newest visible row + /// position wins". So an out-of-order append silently inverts dedup after a + /// crash — the stale row wins — and a full scan cannot see it. + /// + /// The hazard is the background ticker. If it resolved to "whatever memtable + /// is active" rather than to the oldest store still owing an append, a tick + /// enqueued before a freeze but handled after it would append the *incoming* + /// memtable's batches ahead of the outgoing memtable's tail. So the target is + /// resolved from the durability cursor, not from wall-clock timing. + #[tokio::test] + async fn test_wal_append_order_preserves_pk_recency_across_rotation() { + use crate::dataset::mem_wal::wal::WalTailer; + + let (store, base_path, base_uri, _temp_dir) = create_local_store().await; + let schema = schema_with_pk(); + let shard_id = Uuid::new_v4(); + + // Two batches per memtable, so the second put fills it and rotates. + let config = ShardWriterConfig { + max_memtable_batches: 2, + ..memtable_config_with_pk(shard_id) + }; + + let writer = ShardWriter::open( + store.clone(), + base_path.clone(), + base_uri, + config, + schema.clone(), + vec![], + ) + .await + .unwrap(); + + // Memtable 1: ids 7 then 20 (the second fills it and triggers the freeze). + writer + .put(vec![create_test_batch(&schema, 7, 1)]) + .await + .unwrap(); + writer + .put(vec![create_test_batch(&schema, 20, 1)]) + .await + .unwrap(); + // Memtable 2 overwrites id=7 with a newer row. Its append must land in the + // WAL *after* memtable 1's, or replay would resolve id=7 to the stale copy. + writer + .put(vec![create_test_batch(&schema, 30, 1)]) + .await + .unwrap(); + writer.close().await.unwrap(); + + // Walk the WAL in entry order and collect the ids as replay would see them. + let tailer = WalTailer::new(store, base_path, shard_id); + let first = tailer.first_position().await.unwrap(); + let next = tailer.next_position().await.unwrap(); + let mut ids: Vec = Vec::new(); + for position in first..next { + let Some(entry) = tailer.read_entry(position).await.unwrap() else { + continue; + }; + for batch in &entry.batches { + let column = batch + .column_by_name("id") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + ids.extend((0..column.len()).map(|i| column.value(i))); + } + } + + assert_eq!( + ids, + vec![7, 20, 30], + "WAL entries must follow global batch-position order; memtable 1's rows must \ + precede memtable 2's, or replay inverts primary-key recency" + ); + } + /// An index config that disagrees with the schema fails `open()` outright. /// It must not be allowed to accept writes: the insert would fail /// deterministically on every batch, including batches replayed from the @@ -6451,7 +6717,7 @@ mod tests { durable_write: false, sync_indexed_write: false, max_wal_buffer_size: 64 * 1024 * 1024, - max_wal_flush_interval: None, + max_wal_flush_interval: Some(Duration::from_millis(10)), max_memtable_size: 64 * 1024 * 1024, manifest_scan_batch_size: 2, ..Default::default() @@ -6506,7 +6772,7 @@ mod tests { durable_write: false, sync_indexed_write: false, max_wal_buffer_size: 64 * 1024 * 1024, - max_wal_flush_interval: None, + max_wal_flush_interval: Some(Duration::from_millis(10)), max_memtable_size: 64 * 1024 * 1024, manifest_scan_batch_size: 2, ..Default::default() @@ -6561,7 +6827,7 @@ mod tests { durable_write: false, sync_indexed_write: false, max_wal_buffer_size: 64 * 1024 * 1024, - max_wal_flush_interval: None, + max_wal_flush_interval: Some(Duration::from_millis(10)), max_memtable_size: 64 * 1024 * 1024, manifest_scan_batch_size: 2, // Short grace so the sweep is observable without a slow test. @@ -6621,7 +6887,7 @@ mod tests { durable_write: false, sync_indexed_write: false, max_wal_buffer_size: 64 * 1024 * 1024, - max_wal_flush_interval: None, + max_wal_flush_interval: Some(Duration::from_millis(10)), max_memtable_size: 64 * 1024 * 1024, manifest_scan_batch_size: 2, frozen_memtable_grace: Duration::ZERO, From c8c16447446797dceb89b2e1501a99009458c3a0 Mon Sep 17 00:00:00 2001 From: Daniel Rammer Date: Tue, 14 Jul 2026 21:33:58 -0500 Subject: [PATCH 2/4] refactor(mem_wal): remove the inert sync_indexed_write config family MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `sync_indexed_write`, `async_index_buffer_rows`, and `async_index_interval` had no behavioral consumers — they were only serialized into the config-defaults metadata map. They existed to configure async index-update buffering, which the index-apply-task split replaced with unconditional per-put indexing: a write is now read-your-writes through the index in every mode, which is exactly what `sync_indexed_write` promised. The knobs no longer control anything. Remove all three across every surface: the Rust core field, builders, defaults, and metadata serialization; the write-throughput bench's now-meaningless sync/async index axis; and the Python and Java binding parameters. Co-Authored-By: Claude Opus 4.8 (1M context) --- java/lance-jni/src/mem_wal.rs | 9 -- .../org/lance/memwal/ShardWriterConfig.java | 41 ----- .../java/org/lance/memwal/MemWalTest.java | 1 - python/python/lance/dataset.py | 18 --- python/python/tests/test_mem_wal.py | 7 +- python/src/dataset.rs | 33 ---- .../mem_wal/fts/mem_wal_fineweb_fts.rs | 12 +- .../mem_wal/fts/mem_wal_fts_read_bench.rs | 1 - .../mem_wal/kv/mem_wal_kv_point_lookup.rs | 1 - .../mem_wal_point_lookup_bench.rs | 1 - .../vector/hnsw/mem_wal_recall_hnsw.rs | 1 - .../mem_wal/vector/mem_wal_index_micro.rs | 1 - .../mem_wal/vector/mem_wal_vector_bench.rs | 1 - .../benches/mem_wal/write/mem_wal_replay.rs | 1 - .../mem_wal_shard_writer_backpressure.rs | 10 -- .../benches/mem_wal/write/mem_wal_write.rs | 150 +++++++----------- rust/lance/src/dataset/mem_wal/api.rs | 12 -- rust/lance/src/dataset/mem_wal/write.rs | 93 +---------- 18 files changed, 69 insertions(+), 324 deletions(-) diff --git a/java/lance-jni/src/mem_wal.rs b/java/lance-jni/src/mem_wal.rs index 8c97d4aee2e..9047c6288b0 100644 --- a/java/lance-jni/src/mem_wal.rs +++ b/java/lance-jni/src/mem_wal.rs @@ -1265,9 +1265,6 @@ fn build_writer_config(env: &mut JNIEnv, config: &JObject) -> Result Result durableWrite = Optional.empty(); - private Optional syncIndexedWrite = Optional.empty(); private Optional maxWalBufferSize = Optional.empty(); private Optional maxWalFlushIntervalMs = Optional.empty(); private Optional maxMemtableSize = Optional.empty(); @@ -36,8 +35,6 @@ public class ShardWriterConfig { private Optional maxMemtableBatches = Optional.empty(); private Optional maxUnflushedMemtableBytes = Optional.empty(); private Optional manifestScanBatchSize = Optional.empty(); - private Optional asyncIndexBufferRows = Optional.empty(); - private Optional asyncIndexIntervalMs = Optional.empty(); private Optional backpressureLogIntervalMs = Optional.empty(); private Optional statsLogIntervalMs = Optional.empty(); private List hnswParams = Collections.emptyList(); @@ -48,12 +45,6 @@ public ShardWriterConfig withDurableWrite(boolean durableWrite) { return this; } - /** Whether indexed writes are applied synchronously. */ - public ShardWriterConfig withSyncIndexedWrite(boolean syncIndexedWrite) { - this.syncIndexedWrite = Optional.of(syncIndexedWrite); - return this; - } - /** Maximum size of the in-memory WAL buffer, in bytes. */ public ShardWriterConfig withMaxWalBufferSize(long maxWalBufferSize) { Preconditions.checkArgument( @@ -118,26 +109,6 @@ public ShardWriterConfig withManifestScanBatchSize(long manifestScanBatchSize) { return this; } - /** Number of rows buffered before an asynchronous index update is triggered. */ - public ShardWriterConfig withAsyncIndexBufferRows(long asyncIndexBufferRows) { - Preconditions.checkArgument( - asyncIndexBufferRows >= 0, - "asyncIndexBufferRows must not be negative, got %s", - asyncIndexBufferRows); - this.asyncIndexBufferRows = Optional.of(asyncIndexBufferRows); - return this; - } - - /** Interval between asynchronous index updates, in milliseconds. */ - public ShardWriterConfig withAsyncIndexIntervalMs(long asyncIndexIntervalMs) { - Preconditions.checkArgument( - asyncIndexIntervalMs >= 0, - "asyncIndexIntervalMs must not be negative, got %s", - asyncIndexIntervalMs); - this.asyncIndexIntervalMs = Optional.of(asyncIndexIntervalMs); - return this; - } - /** Interval between backpressure log messages, in milliseconds. */ public ShardWriterConfig withBackpressureLogIntervalMs(long backpressureLogIntervalMs) { Preconditions.checkArgument( @@ -172,10 +143,6 @@ public Optional durableWrite() { return durableWrite; } - public Optional syncIndexedWrite() { - return syncIndexedWrite; - } - public Optional maxWalBufferSize() { return maxWalBufferSize; } @@ -204,14 +171,6 @@ public Optional manifestScanBatchSize() { return manifestScanBatchSize; } - public Optional asyncIndexBufferRows() { - return asyncIndexBufferRows; - } - - public Optional asyncIndexIntervalMs() { - return asyncIndexIntervalMs; - } - public Optional backpressureLogIntervalMs() { return backpressureLogIntervalMs; } diff --git a/java/src/test/java/org/lance/memwal/MemWalTest.java b/java/src/test/java/org/lance/memwal/MemWalTest.java index 57d0b5b81ad..3ee998733ea 100644 --- a/java/src/test/java/org/lance/memwal/MemWalTest.java +++ b/java/src/test/java/org/lance/memwal/MemWalTest.java @@ -409,7 +409,6 @@ void testShardWriterDeleteMasksBaseRow(@TempDir Path tempDir) throws Exception { ShardWriterConfig config = new ShardWriterConfig() .withDurableWrite(true) - .withSyncIndexedWrite(true) .withMaxWalBufferSize(1) .withMaxWalFlushIntervalMs(10); diff --git a/python/python/lance/dataset.py b/python/python/lance/dataset.py index 7ba84e67017..34b014385fa 100644 --- a/python/python/lance/dataset.py +++ b/python/python/lance/dataset.py @@ -5150,7 +5150,6 @@ def initialize_mem_wal( identity_column: Optional[str] = None, unsharded: bool = False, durable_write: Optional[bool] = None, - sync_indexed_write: Optional[bool] = None, max_wal_buffer_size: Optional[int] = None, max_wal_flush_interval_ms: Optional[int] = None, max_memtable_size: Optional[int] = None, @@ -5158,8 +5157,6 @@ def initialize_mem_wal( max_memtable_batches: Optional[int] = None, max_unflushed_memtable_bytes: Optional[int] = None, manifest_scan_batch_size: Optional[int] = None, - async_index_buffer_rows: Optional[int] = None, - async_index_interval_ms: Optional[int] = None, backpressure_log_interval_ms: Optional[int] = None, stats_log_interval_ms: Optional[int] = None, hnsw_params: Optional[Dict[str, Dict[str, int]]] = None, @@ -5220,7 +5217,6 @@ def initialize_mem_wal( identity_column=identity_column, unsharded=unsharded, durable_write=durable_write, - sync_indexed_write=sync_indexed_write, max_wal_buffer_size=max_wal_buffer_size, max_wal_flush_interval_ms=max_wal_flush_interval_ms, max_memtable_size=max_memtable_size, @@ -5228,8 +5224,6 @@ def initialize_mem_wal( max_memtable_batches=max_memtable_batches, max_unflushed_memtable_bytes=max_unflushed_memtable_bytes, manifest_scan_batch_size=manifest_scan_batch_size, - async_index_buffer_rows=async_index_buffer_rows, - async_index_interval_ms=async_index_interval_ms, backpressure_log_interval_ms=backpressure_log_interval_ms, stats_log_interval_ms=stats_log_interval_ms, hnsw_params=hnsw_params, @@ -5252,7 +5246,6 @@ def mem_wal_writer( shard_id: str, *, durable_write: Optional[bool] = None, - sync_indexed_write: Optional[bool] = None, max_wal_buffer_size: Optional[int] = None, max_wal_flush_interval_ms: Optional[int] = None, max_memtable_size: Optional[int] = None, @@ -5260,8 +5253,6 @@ def mem_wal_writer( max_memtable_batches: Optional[int] = None, max_unflushed_memtable_bytes: Optional[int] = None, manifest_scan_batch_size: Optional[int] = None, - async_index_buffer_rows: Optional[int] = None, - async_index_interval_ms: Optional[int] = None, backpressure_log_interval_ms: Optional[int] = None, stats_log_interval_ms: Optional[int] = None, hnsw_params: Optional[Dict[str, Dict[str, int]]] = None, @@ -5279,8 +5270,6 @@ def mem_wal_writer( ``str(uuid.uuid4())``). durable_write : bool, optional Whether to fsync WAL writes (default: ``True``). - sync_indexed_write : bool, optional - Whether index updates are synchronous (default: ``True``). max_wal_buffer_size : int, optional Maximum WAL buffer size in bytes (default: 10 MB). max_wal_flush_interval_ms : int, optional @@ -5295,10 +5284,6 @@ def mem_wal_writer( Maximum unflushed bytes before backpressure (default: 1 GB). manifest_scan_batch_size : int, optional Batch size for manifest scans (default: 2). - async_index_buffer_rows : int, optional - Buffer rows for async index updates (default: 10 000). - async_index_interval_ms : int, optional - Interval for async index updates in milliseconds (default: 1000). backpressure_log_interval_ms : int, optional Interval for backpressure log messages in milliseconds (default: 30 000). @@ -5346,7 +5331,6 @@ def mem_wal_writer( name: val for name, val in [ ("durable_write", durable_write), - ("sync_indexed_write", sync_indexed_write), ("max_wal_buffer_size", max_wal_buffer_size), ("max_wal_flush_interval_ms", max_wal_flush_interval_ms), ("max_memtable_size", max_memtable_size), @@ -5354,8 +5338,6 @@ def mem_wal_writer( ("max_memtable_batches", max_memtable_batches), ("max_unflushed_memtable_bytes", max_unflushed_memtable_bytes), ("manifest_scan_batch_size", manifest_scan_batch_size), - ("async_index_buffer_rows", async_index_buffer_rows), - ("async_index_interval_ms", async_index_interval_ms), ("backpressure_log_interval_ms", backpressure_log_interval_ms), ("stats_log_interval_ms", stats_log_interval_ms), ("hnsw_params", hnsw_params), diff --git a/python/python/tests/test_mem_wal.py b/python/python/tests/test_mem_wal.py index f0825eb9a66..d93f5fa41dd 100644 --- a/python/python/tests/test_mem_wal.py +++ b/python/python/tests/test_mem_wal.py @@ -210,7 +210,6 @@ def test_shard_writer_delete_binding_masks_base_row(tmp_path): with ds.mem_wal_writer( shard_id, durable_write=True, - sync_indexed_write=True, max_wal_buffer_size=1, max_wal_flush_interval_ms=10, ) as writer: @@ -356,7 +355,6 @@ def test_shard_writer_e2e_correctness(tmp_path): writer = ds.mem_wal_writer( shard_id, durable_write=True, - sync_indexed_write=True, max_wal_buffer_size=10 * 1024, # 10 KB max_wal_flush_interval_ms=50, max_memtable_size=80, # flush after ~80 rows @@ -406,9 +404,7 @@ def test_shard_writer_e2e_correctness(tmp_path): # === New writer: write and read back via active MemTable scanner === ds2 = lance.dataset(ds_path) shard_id2 = str(uuid.uuid4()) - with ds2.mem_wal_writer( - shard_id2, durable_write=False, sync_indexed_write=True - ) as writer2: + with ds2.mem_wal_writer(shard_id2, durable_write=False) as writer2: verify_batch = _e2e_batch(schema, start_id=10000, num_rows=10) writer2.put(pa.Table.from_batches([verify_batch])) result = writer2.lsm_scanner().to_table() @@ -523,7 +519,6 @@ def test_initialize_mem_wal_writer_config_defaults(tmp_path): # Duration knobs are recorded in milliseconds with a `_ms` suffix. assert defaults["max_wal_flush_interval_ms"] == "250" # Every ShardWriterConfig tunable is recorded once any default is set. - assert "sync_indexed_write" in defaults assert "enable_memtable" in defaults diff --git a/python/src/dataset.rs b/python/src/dataset.rs index 298c033dd2e..3b4cf13bf0a 100644 --- a/python/src/dataset.rs +++ b/python/src/dataset.rs @@ -203,7 +203,6 @@ fn stats_log_interval_from_millis(ms: u64) -> Option { #[allow(clippy::too_many_arguments)] fn writer_config_from_kwargs( durable_write: Option, - sync_indexed_write: Option, max_wal_buffer_size: Option, max_wal_flush_interval_ms: Option, max_memtable_size: Option, @@ -211,8 +210,6 @@ fn writer_config_from_kwargs( max_memtable_batches: Option, max_unflushed_memtable_bytes: Option, manifest_scan_batch_size: Option, - async_index_buffer_rows: Option, - async_index_interval_ms: Option, backpressure_log_interval_ms: Option, stats_log_interval_ms: Option, hnsw_params: Option>>, @@ -225,10 +222,6 @@ fn writer_config_from_kwargs( config = config.with_durable_write(v); any = true; } - if let Some(v) = sync_indexed_write { - config = config.with_sync_indexed_write(v); - any = true; - } if let Some(v) = max_wal_buffer_size { config = config.with_max_wal_buffer_size(v); any = true; @@ -257,14 +250,6 @@ fn writer_config_from_kwargs( config = config.with_manifest_scan_batch_size(v); any = true; } - if let Some(v) = async_index_buffer_rows { - config = config.with_async_index_buffer_rows(v); - any = true; - } - if let Some(v) = async_index_interval_ms { - config = config.with_async_index_interval(Duration::from_millis(v)); - any = true; - } if let Some(v) = backpressure_log_interval_ms { config = config.with_backpressure_log_interval(Duration::from_millis(v)); any = true; @@ -3586,7 +3571,6 @@ impl Dataset { identity_column=None, unsharded=false, durable_write=None, - sync_indexed_write=None, max_wal_buffer_size=None, max_wal_flush_interval_ms=None, max_memtable_size=None, @@ -3594,8 +3578,6 @@ impl Dataset { max_memtable_batches=None, max_unflushed_memtable_bytes=None, manifest_scan_batch_size=None, - async_index_buffer_rows=None, - async_index_interval_ms=None, backpressure_log_interval_ms=None, stats_log_interval_ms=None, hnsw_params=None, @@ -3609,7 +3591,6 @@ impl Dataset { identity_column: Option, unsharded: bool, durable_write: Option, - sync_indexed_write: Option, max_wal_buffer_size: Option, max_wal_flush_interval_ms: Option, max_memtable_size: Option, @@ -3617,8 +3598,6 @@ impl Dataset { max_memtable_batches: Option, max_unflushed_memtable_bytes: Option, manifest_scan_batch_size: Option, - async_index_buffer_rows: Option, - async_index_interval_ms: Option, backpressure_log_interval_ms: Option, stats_log_interval_ms: Option, hnsw_params: Option>>, @@ -3646,7 +3625,6 @@ impl Dataset { let writer_config = writer_config_from_kwargs( durable_write, - sync_indexed_write, max_wal_buffer_size, max_wal_flush_interval_ms, max_memtable_size, @@ -3654,8 +3632,6 @@ impl Dataset { max_memtable_batches, max_unflushed_memtable_bytes, manifest_scan_batch_size, - async_index_buffer_rows, - async_index_interval_ms, backpressure_log_interval_ms, stats_log_interval_ms, hnsw_params, @@ -3737,7 +3713,6 @@ impl Dataset { shard_id, *, durable_write=None, - sync_indexed_write=None, max_wal_buffer_size=None, max_wal_flush_interval_ms=None, max_memtable_size=None, @@ -3745,8 +3720,6 @@ impl Dataset { max_memtable_batches=None, max_unflushed_memtable_bytes=None, manifest_scan_batch_size=None, - async_index_buffer_rows=None, - async_index_interval_ms=None, backpressure_log_interval_ms=None, stats_log_interval_ms=None, hnsw_params=None, @@ -3756,7 +3729,6 @@ impl Dataset { py: Python<'_>, shard_id: String, durable_write: Option, - sync_indexed_write: Option, max_wal_buffer_size: Option, max_wal_flush_interval_ms: Option, max_memtable_size: Option, @@ -3764,8 +3736,6 @@ impl Dataset { max_memtable_batches: Option, max_unflushed_memtable_bytes: Option, manifest_scan_batch_size: Option, - async_index_buffer_rows: Option, - async_index_interval_ms: Option, backpressure_log_interval_ms: Option, stats_log_interval_ms: Option, hnsw_params: Option>>, @@ -3777,7 +3747,6 @@ impl Dataset { let config = writer_config_from_kwargs( durable_write, - sync_indexed_write, max_wal_buffer_size, max_wal_flush_interval_ms, max_memtable_size, @@ -3785,8 +3754,6 @@ impl Dataset { max_memtable_batches, max_unflushed_memtable_bytes, manifest_scan_batch_size, - async_index_buffer_rows, - async_index_interval_ms, backpressure_log_interval_ms, stats_log_interval_ms, hnsw_params, diff --git a/rust/lance/benches/mem_wal/fts/mem_wal_fineweb_fts.rs b/rust/lance/benches/mem_wal/fts/mem_wal_fineweb_fts.rs index 63f9368c3c5..188e2c73ce2 100644 --- a/rust/lance/benches/mem_wal/fts/mem_wal_fineweb_fts.rs +++ b/rust/lance/benches/mem_wal/fts/mem_wal_fineweb_fts.rs @@ -111,11 +111,6 @@ impl Mode { fn durable_write(self) -> bool { matches!(self, Self::SyncNoIndex | Self::SyncIndexed) } - - /// Index update happens inline in `put` (only meaningful when indexed). - fn sync_indexed_write(self) -> bool { - matches!(self, Self::SyncIndexed) - } } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -428,7 +423,6 @@ fn shard_writer_config(args: &Args, shard_id: Uuid, disable_auto_flush: bool) -> }; let mut config = ShardWriterConfig::new(shard_id) .with_durable_write(args.mode.durable_write()) - .with_sync_indexed_write(args.mode.sync_indexed_write()) .with_max_memtable_size(args.max_memtable_size) .with_max_unflushed_memtable_bytes(args.max_unflushed_memtable_bytes) .with_max_memtable_rows(max_rows) @@ -638,7 +632,7 @@ async fn run_read(args: &Args, uri: &str, corpus: &[String]) -> Result Result Result { wal_persist_retry_base_delay: std::time::Duration::from_millis(50), shard_spec_id: 0, durable_write: false, - sync_indexed_write: false, max_memtable_size: args.max_memtable_rows * row_bytes * 100, max_memtable_rows: args.max_memtable_rows, max_memtable_batches: batches_per_gen, diff --git a/rust/lance/benches/mem_wal/kv/mem_wal_kv_point_lookup.rs b/rust/lance/benches/mem_wal/kv/mem_wal_kv_point_lookup.rs index 757c2539642..e5c6cf234e1 100644 --- a/rust/lance/benches/mem_wal/kv/mem_wal_kv_point_lookup.rs +++ b/rust/lance/benches/mem_wal/kv/mem_wal_kv_point_lookup.rs @@ -792,7 +792,6 @@ async fn run_lance( wal_persist_retry_base_delay: std::time::Duration::from_millis(50), shard_spec_id: 0, durable_write: true, - sync_indexed_write: true, max_memtable_size: big, max_memtable_rows: args.rows * 4 + 1_000_000, max_memtable_batches: args.rows / args.batch_rows + 1_000_000, diff --git a/rust/lance/benches/mem_wal/point_lookup/mem_wal_point_lookup_bench.rs b/rust/lance/benches/mem_wal/point_lookup/mem_wal_point_lookup_bench.rs index cb9e6413ac1..d723cfde9c8 100644 --- a/rust/lance/benches/mem_wal/point_lookup/mem_wal_point_lookup_bench.rs +++ b/rust/lance/benches/mem_wal/point_lookup/mem_wal_point_lookup_bench.rs @@ -338,7 +338,6 @@ async fn run_lookup(args: &Args) -> Result { wal_persist_retry_base_delay: std::time::Duration::from_millis(50), shard_spec_id: 0, durable_write: false, - sync_indexed_write: false, max_memtable_size: max_memtable_rows * 200, max_memtable_rows, max_wal_flush_interval: Some(Duration::from_secs(60)), diff --git a/rust/lance/benches/mem_wal/vector/hnsw/mem_wal_recall_hnsw.rs b/rust/lance/benches/mem_wal/vector/hnsw/mem_wal_recall_hnsw.rs index 85dd5e7ac14..a4968161647 100644 --- a/rust/lance/benches/mem_wal/vector/hnsw/mem_wal_recall_hnsw.rs +++ b/rust/lance/benches/mem_wal/vector/hnsw/mem_wal_recall_hnsw.rs @@ -389,7 +389,6 @@ async fn run_checkpoint( wal_persist_retry_base_delay: std::time::Duration::from_millis(50), shard_spec_id: 0, durable_write: false, - sync_indexed_write: true, max_memtable_size: cp.saturating_mul(row_size_estimate).saturating_mul(4), max_memtable_rows: cp.saturating_mul(2), max_memtable_batches: total_batches_max.saturating_mul(2).max(8_000), diff --git a/rust/lance/benches/mem_wal/vector/mem_wal_index_micro.rs b/rust/lance/benches/mem_wal/vector/mem_wal_index_micro.rs index 812535ce053..c6fb42834a3 100644 --- a/rust/lance/benches/mem_wal/vector/mem_wal_index_micro.rs +++ b/rust/lance/benches/mem_wal/vector/mem_wal_index_micro.rs @@ -214,7 +214,6 @@ async fn main() -> lance_core::Result<()> { wal_persist_retry_base_delay: std::time::Duration::from_millis(50), shard_spec_id: 0, durable_write, - sync_indexed_write: true, max_memtable_size: max_rows.saturating_mul(row_size_estimate).saturating_mul(4), max_memtable_rows: max_rows.saturating_mul(2), max_memtable_batches: total_batches_max.saturating_mul(2).max(8_000), diff --git a/rust/lance/benches/mem_wal/vector/mem_wal_vector_bench.rs b/rust/lance/benches/mem_wal/vector/mem_wal_vector_bench.rs index 632bd37626c..a8fe8e81f29 100644 --- a/rust/lance/benches/mem_wal/vector/mem_wal_vector_bench.rs +++ b/rust/lance/benches/mem_wal/vector/mem_wal_vector_bench.rs @@ -520,7 +520,6 @@ async fn run_search(args: &Args) -> Result { wal_persist_retry_base_delay: std::time::Duration::from_millis(50), shard_spec_id: 0, durable_write: false, - sync_indexed_write: false, max_memtable_size: args.max_memtable_rows * row_bytes * 2, max_memtable_rows: args.max_memtable_rows, max_unflushed_memtable_bytes: args.max_memtable_rows * row_bytes * 6, diff --git a/rust/lance/benches/mem_wal/write/mem_wal_replay.rs b/rust/lance/benches/mem_wal/write/mem_wal_replay.rs index 14ec406e5e5..6fda0748006 100644 --- a/rust/lance/benches/mem_wal/write/mem_wal_replay.rs +++ b/rust/lance/benches/mem_wal/write/mem_wal_replay.rs @@ -179,7 +179,6 @@ async fn populate_shard_wal( let dataset = Dataset::open(dataset_uri).await.unwrap(); let mut config = ShardWriterConfig::new(shard_id); config.durable_write = true; - config.sync_indexed_write = false; let writer = dataset.mem_wal_writer(shard_id, config).await.unwrap(); for i in 0..num_entries { diff --git a/rust/lance/benches/mem_wal/write/mem_wal_shard_writer_backpressure.rs b/rust/lance/benches/mem_wal/write/mem_wal_shard_writer_backpressure.rs index 56915fb205d..79298daa872 100644 --- a/rust/lance/benches/mem_wal/write/mem_wal_shard_writer_backpressure.rs +++ b/rust/lance/benches/mem_wal/write/mem_wal_shard_writer_backpressure.rs @@ -99,10 +99,6 @@ impl Mode { fn durable_write(self) -> bool { matches!(self, Self::SyncNoIndex | Self::SyncIndexed) } - - fn sync_indexed_write(self) -> bool { - matches!(self, Self::SyncIndexed) - } } /// Which index the MemTable maintains in the indexed (`*_idx`) modes. @@ -192,7 +188,6 @@ struct Args { max_memtable_batches: Option, max_wal_buffer_size: usize, max_wal_flush_interval_ms: u64, - async_index_buffer_rows: usize, sample_interval_ms: u64, target_rows_per_sec: Option, num_partitions: usize, @@ -223,7 +218,6 @@ impl Default for Args { max_memtable_batches: None, max_wal_buffer_size: 10 * 1024 * 1024, max_wal_flush_interval_ms: 100, - async_index_buffer_rows: 10_000, sample_interval_ms: 500, target_rows_per_sec: None, num_partitions: 1, @@ -343,11 +337,9 @@ async fn run(args: Args) -> Result<()> { let shard_id = Uuid::new_v4(); let mut config = ShardWriterConfig::new(shard_id) .with_durable_write(args.mode.durable_write()) - .with_sync_indexed_write(args.mode.sync_indexed_write()) .with_max_memtable_size(args.max_memtable_size) .with_max_unflushed_memtable_bytes(args.max_unflushed_memtable_bytes) .with_max_wal_buffer_size(args.max_wal_buffer_size) - .with_async_index_buffer_rows(args.async_index_buffer_rows) .with_max_memtable_rows(memtable_limits.rows) .with_max_memtable_batches(memtable_limits.batches); if args.max_wal_flush_interval_ms == 0 { @@ -544,7 +536,6 @@ async fn run(args: Args) -> Result<()> { "max_memtable_batches": memtable_limits.batches, "max_wal_buffer_size": args.max_wal_buffer_size, "max_wal_flush_interval_ms": args.max_wal_flush_interval_ms, - "async_index_buffer_rows": args.async_index_buffer_rows, "sample_interval_ms": args.sample_interval_ms, "skip_close": args.skip_close, "setup_seconds": setup_s, @@ -980,7 +971,6 @@ fn parse_args() -> Result { "--max-wal-flush-interval-ms" => { args.max_wal_flush_interval_ms = parse(&flag, &value)?; } - "--async-index-buffer-rows" => args.async_index_buffer_rows = parse(&flag, &value)?, "--sample-interval-ms" => args.sample_interval_ms = parse(&flag, &value)?, "--target-rows-per-sec" => args.target_rows_per_sec = Some(parse(&flag, &value)?), "--num-partitions" => args.num_partitions = parse(&flag, &value)?, diff --git a/rust/lance/benches/mem_wal/write/mem_wal_write.rs b/rust/lance/benches/mem_wal/write/mem_wal_write.rs index db28c5b33ab..b5018fe09f1 100644 --- a/rust/lance/benches/mem_wal/write/mem_wal_write.rs +++ b/rust/lance/benches/mem_wal/write/mem_wal_write.rs @@ -30,7 +30,6 @@ //! - `BATCH_SIZE`: Number of rows per write batch (default: 20) //! - `NUM_BATCHES`: Total number of batches to write (default: 1000) //! - `DURABLE_WRITE`: yes/no/both (default: no) - whether writes wait for WAL flush -//! - `INDEXED_WRITE`: yes/no/both (default: no) - whether writes update indexes synchronously //! - `MAX_WAL_BUFFER_SIZE`: WAL buffer size in bytes (default: 1MB from ShardWriterConfig) //! - `MAX_FLUSH_INTERVAL_MS`: WAL flush interval in milliseconds, 0 to disable (default: 1000ms) //! - `MAX_MEMTABLE_SIZE`: MemTable size threshold in bytes (default: 64MB from ShardWriterConfig) @@ -44,8 +43,7 @@ //! `no` uses WAL-only mode (no MemTable, no indexes, no Lance flushes; pure WAL throughput). //! `both` runs each combination twice, once per mode, side-by-side. //! When `no` or `both`, the WAL-only branch always runs with -//! `MEMWAL_MAINTAINED_INDEXES=none` and skips `INDEXED_WRITE=yes` -//! (sync-indexed writes require a MemTable). +//! `MEMWAL_MAINTAINED_INDEXES=none`. //! - `SAMPLE_SIZE`: Number of benchmark iterations (default: 10, minimum: 10) #![allow(clippy::print_stdout, clippy::print_stderr)] @@ -119,11 +117,6 @@ fn get_durable_write_options() -> Vec { parse_yes_no_both("DURABLE_WRITE", "no") } -/// Get indexed write settings from environment. -fn get_indexed_write_options() -> Vec { - parse_yes_no_both("INDEXED_WRITE", "no") -} - /// Get enable_memtable settings from environment. Default `yes` keeps /// existing benchmark behavior; `no` runs WAL-only mode; `both` runs both /// modes side-by-side for comparison. @@ -442,31 +435,26 @@ fn build_label( num_batches: usize, batch_size: usize, durable: bool, - indexed: bool, enable_memtable: bool, storage: &str, ) -> String { let durable_str = if durable { "durable" } else { "nondurable" }; - // sync_indexed_write controls sync vs async index updates - let indexed_str = if indexed { "sync_idx" } else { "async_idx" }; let mode_str = if enable_memtable { "memtable" } else { "wal_only" }; format!( - "{}x{} {} {} {} ({})", - num_batches, batch_size, mode_str, durable_str, indexed_str, storage + "{}x{} {} {} ({})", + num_batches, batch_size, mode_str, durable_str, storage ) } /// Build dataset name prefix from config options. -fn build_name_prefix(durable: bool, indexed: bool, enable_memtable: bool) -> String { +fn build_name_prefix(durable: bool, enable_memtable: bool) -> String { let d = if durable { "d" } else { "nd" }; - // sync_indexed_write: sync (si) vs async (ai) - let i = if indexed { "si" } else { "ai" }; let m = if enable_memtable { "mt" } else { "wo" }; - format!("{}_{}_{}", m, d, i) + format!("{}_{}", m, d) } /// Benchmark Lance MemWAL write throughput. @@ -490,7 +478,6 @@ fn bench_lance_memwal_write(c: &mut Criterion) { let maintained_indexes = get_maintained_indexes(); let durable_options = get_durable_write_options(); - let indexed_options = get_indexed_write_options(); let enable_memtable_options = get_enable_memtable_options(); let max_wal_buffer_size = get_max_wal_buffer_size(); let max_flush_interval = get_max_flush_interval(); @@ -550,70 +537,56 @@ fn bench_lance_memwal_write(c: &mut Criterion) { // Generate benchmarks for all combinations for &enable_memtable in &enable_memtable_options { for &durable in &durable_options { - for &indexed in &indexed_options { - if !enable_memtable && indexed { - eprintln!( - "Skipping wal_only + sync_idx (sync_indexed_write requires a MemTable)" - ); - continue; - } - - let label = build_label( - num_batches, - batch_size, - durable, - indexed, - enable_memtable, - storage_label, - ); - let name_prefix = build_name_prefix(durable, indexed, enable_memtable); - - // WAL-only mode never uses indexes; force the dataset - // setup to skip the MemWAL index list. - let effective_indexes: Vec = if enable_memtable { - maintained_indexes.clone() - } else { - Vec::new() - }; - - // Create dataset ONCE before benchmark iterations - // Each iteration will use a different shard on the same dataset - let dataset = rt.block_on(create_dataset( - &schema, - &name_prefix, - vector_dim, - &effective_indexes, - &dataset_prefix, - )); - let dataset_uri = dataset.uri().to_string(); - - // Pre-generate all batches before timing (outside iter_custom) - let batches: Arc> = Arc::new( - (0..num_batches) - .map(|i| { - create_test_batch( - &schema, - (i * batch_size) as i64, - batch_size, - vector_dim, - ) - }) - .collect(), - ); - - println!("Running: {}", label); - - // Track if we've printed stats (only print once across all samples) - let stats_printed = Arc::new(AtomicBool::new(false)); - - group.bench_with_input( - BenchmarkId::new("Lance MemWAL", &label), - &(batch_size, num_batches, durable, indexed, row_size_bytes), - |b, &(_batch_size, _num_batches, durable, indexed, row_size_bytes)| { - let dataset_uri = dataset_uri.clone(); - let batches = batches.clone(); - let stats_printed = stats_printed.clone(); - b.to_async(&rt).iter_custom(|iters| { + let label = build_label( + num_batches, + batch_size, + durable, + enable_memtable, + storage_label, + ); + let name_prefix = build_name_prefix(durable, enable_memtable); + + // WAL-only mode never uses indexes; force the dataset + // setup to skip the MemWAL index list. + let effective_indexes: Vec = if enable_memtable { + maintained_indexes.clone() + } else { + Vec::new() + }; + + // Create dataset ONCE before benchmark iterations + // Each iteration will use a different shard on the same dataset + let dataset = rt.block_on(create_dataset( + &schema, + &name_prefix, + vector_dim, + &effective_indexes, + &dataset_prefix, + )); + let dataset_uri = dataset.uri().to_string(); + + // Pre-generate all batches before timing (outside iter_custom) + let batches: Arc> = Arc::new( + (0..num_batches) + .map(|i| { + create_test_batch(&schema, (i * batch_size) as i64, batch_size, vector_dim) + }) + .collect(), + ); + + println!("Running: {}", label); + + // Track if we've printed stats (only print once across all samples) + let stats_printed = Arc::new(AtomicBool::new(false)); + + group.bench_with_input( + BenchmarkId::new("Lance MemWAL", &label), + &(batch_size, num_batches, durable, row_size_bytes), + |b, &(_batch_size, _num_batches, durable, row_size_bytes)| { + let dataset_uri = dataset_uri.clone(); + let batches = batches.clone(); + let stats_printed = stats_printed.clone(); + b.to_async(&rt).iter_custom(|iters| { let dataset_uri = dataset_uri.clone(); let batches = batches.clone(); let stats_printed = stats_printed.clone(); @@ -631,10 +604,10 @@ fn bench_lance_memwal_write(c: &mut Criterion) { shard_id, shard_spec_id: 0, max_wal_persist_retries: 3, - wal_persist_retry_base_delay: - std::time::Duration::from_millis(50), + wal_persist_retry_base_delay: std::time::Duration::from_millis( + 50, + ), durable_write: durable, - sync_indexed_write: indexed, max_wal_buffer_size: max_wal_buffer_size .unwrap_or(default_config.max_wal_buffer_size), max_wal_flush_interval: max_flush_interval @@ -643,8 +616,6 @@ fn bench_lance_memwal_write(c: &mut Criterion) { .unwrap_or(default_config.max_memtable_size), max_memtable_rows: default_config.max_memtable_rows, max_memtable_batches: default_config.max_memtable_batches, - async_index_buffer_rows: default_config.async_index_buffer_rows, - async_index_interval: default_config.async_index_interval, manifest_scan_batch_size: default_config .manifest_scan_batch_size, max_unflushed_memtable_bytes: default_config @@ -706,9 +677,8 @@ fn bench_lance_memwal_write(c: &mut Criterion) { total_duration } }) - }, - ); - } + }, + ); } } diff --git a/rust/lance/src/dataset/mem_wal/api.rs b/rust/lance/src/dataset/mem_wal/api.rs index 1bd95092e7c..4351ecc95b7 100644 --- a/rust/lance/src/dataset/mem_wal/api.rs +++ b/rust/lance/src/dataset/mem_wal/api.rs @@ -392,10 +392,6 @@ fn writer_config_to_defaults(config: &ShardWriterConfig) -> HashMap HashMap Self { - self.sync_indexed_write = indexed; - self - } - /// Set maximum WAL buffer size. pub fn with_max_wal_buffer_size(mut self, size: usize) -> Self { self.max_wal_buffer_size = size; @@ -383,18 +345,6 @@ impl ShardWriterConfig { self } - /// Set async index buffer rows. - pub fn with_async_index_buffer_rows(mut self, rows: usize) -> Self { - self.async_index_buffer_rows = rows; - self - } - - /// Set async index interval. - pub fn with_async_index_interval(mut self, interval: Duration) -> Self { - self.async_index_interval = interval; - self - } - /// Set stats logging interval. Use None to disable periodic stats logging. pub fn with_stats_log_interval(mut self, interval: Option) -> Self { self.stats_log_interval = interval; @@ -3991,7 +3941,6 @@ mod tests { ShardWriterConfig { shard_id, durable_write: false, - sync_indexed_write: true, manifest_scan_batch_size: 2, ..Default::default() } @@ -4316,7 +4265,6 @@ mod tests { shard_id: Uuid::new_v4(), shard_spec_id: 0, durable_write: false, - sync_indexed_write: false, max_wal_buffer_size: 1024 * 1024, max_wal_flush_interval: Some(Duration::from_millis(10)), max_memtable_size: 64 * 1024 * 1024, @@ -4359,7 +4307,6 @@ mod tests { shard_id: Uuid::new_v4(), shard_spec_id: 0, durable_write: true, - sync_indexed_write: false, max_wal_buffer_size: 1024 * 1024, max_wal_flush_interval: Some(Duration::from_millis(10)), max_memtable_size: 64 * 1024 * 1024, @@ -4396,7 +4343,6 @@ mod tests { shard_id: Uuid::new_v4(), shard_spec_id: 0, durable_write: false, - sync_indexed_write: false, max_wal_buffer_size: 1024 * 1024, max_wal_flush_interval: Some(Duration::from_millis(10)), max_memtable_size: 64 * 1024 * 1024, @@ -4442,7 +4388,6 @@ mod tests { shard_id: Uuid::new_v4(), shard_spec_id: 0, durable_write: false, - sync_indexed_write: false, max_wal_buffer_size: 1024 * 1024, max_wal_flush_interval: Some(Duration::from_millis(10)), max_memtable_size: 64 * 1024 * 1024, @@ -4477,7 +4422,6 @@ mod tests { shard_id: Uuid::new_v4(), shard_spec_id: 0, durable_write: false, - sync_indexed_write: true, max_wal_buffer_size: 1024 * 1024, max_wal_flush_interval: Some(Duration::from_millis(10)), max_memtable_size: 64 * 1024 * 1024, @@ -4529,7 +4473,6 @@ mod tests { shard_id, shard_spec_id: 0, durable_write: false, - sync_indexed_write: true, max_wal_buffer_size: 1024 * 1024, max_wal_flush_interval: Some(Duration::from_millis(10)), max_memtable_size: 64 * 1024 * 1024, @@ -4618,7 +4561,6 @@ mod tests { shard_id: Uuid::new_v4(), shard_spec_id: 0, durable_write: false, - sync_indexed_write: false, max_wal_buffer_size: 1024 * 1024, max_wal_flush_interval: Some(Duration::from_millis(10)), max_memtable_size: 1024, // Very small - will trigger flush quickly @@ -4725,7 +4667,6 @@ mod tests { shard_id: Uuid::new_v4(), shard_spec_id: 0, durable_write: true, - sync_indexed_write: false, max_wal_buffer_size: 1024 * 1024, max_wal_flush_interval: Some(Duration::from_millis(10)), max_memtable_size: 64, @@ -4770,7 +4711,6 @@ mod tests { shard_id: Uuid::new_v4(), shard_spec_id: 0, durable_write: true, - sync_indexed_write: false, max_wal_buffer_size: 1024 * 1024, max_wal_flush_interval: Some(Duration::from_millis(10)), // Tiny size threshold — every batch crosses it. @@ -4837,7 +4777,6 @@ mod tests { shard_id, shard_spec_id: 0, durable_write: true, - sync_indexed_write: false, max_wal_buffer_size: 1024 * 1024, max_wal_flush_interval: Some(Duration::from_millis(10)), max_memtable_size: usize::MAX, @@ -4917,7 +4856,6 @@ mod tests { shard_spec_id: 0, durable_write: false, enable_memtable, - sync_indexed_write: false, max_wal_buffer_size: usize::MAX, max_wal_flush_interval: None, max_wal_persist_retries: 0, @@ -4968,7 +4906,6 @@ mod tests { shard_id: Uuid::new_v4(), shard_spec_id: 0, durable_write: true, - sync_indexed_write: false, max_wal_buffer_size: 1024 * 1024, max_wal_flush_interval: Some(Duration::from_millis(10)), // Tiny size threshold so a few batches cross it. @@ -5402,7 +5339,6 @@ mod tests { shard_id, shard_spec_id: 0, durable_write: true, - sync_indexed_write: false, max_wal_buffer_size: 1024 * 1024, max_wal_flush_interval: Some(Duration::from_millis(10)), max_memtable_size: 64 * 1024 * 1024, @@ -6715,7 +6651,6 @@ mod tests { shard_id: Uuid::new_v4(), shard_spec_id: 0, durable_write: false, - sync_indexed_write: false, max_wal_buffer_size: 64 * 1024 * 1024, max_wal_flush_interval: Some(Duration::from_millis(10)), max_memtable_size: 64 * 1024 * 1024, @@ -6770,7 +6705,6 @@ mod tests { shard_id: Uuid::new_v4(), shard_spec_id: 0, durable_write: false, - sync_indexed_write: false, max_wal_buffer_size: 64 * 1024 * 1024, max_wal_flush_interval: Some(Duration::from_millis(10)), max_memtable_size: 64 * 1024 * 1024, @@ -6825,7 +6759,6 @@ mod tests { shard_id: Uuid::new_v4(), shard_spec_id: 0, durable_write: false, - sync_indexed_write: false, max_wal_buffer_size: 64 * 1024 * 1024, max_wal_flush_interval: Some(Duration::from_millis(10)), max_memtable_size: 64 * 1024 * 1024, @@ -6885,7 +6818,6 @@ mod tests { shard_id: Uuid::new_v4(), shard_spec_id: 0, durable_write: false, - sync_indexed_write: false, max_wal_buffer_size: 64 * 1024 * 1024, max_wal_flush_interval: Some(Duration::from_millis(10)), max_memtable_size: 64 * 1024 * 1024, @@ -7196,9 +7128,7 @@ mod shard_writer_tests { Some("100") ); // Every tunable field is present. - assert!(defaults.contains_key("sync_indexed_write")); assert!(defaults.contains_key("enable_memtable")); - assert!(defaults.contains_key("async_index_interval_ms")); // add_writer_config_default records arbitrary keys. assert_eq!( defaults.get("custom_knob").map(String::as_str), @@ -7410,9 +7340,7 @@ mod shard_writer_tests { .expect("Failed to initialize MemWAL"); let shard_id = Uuid::new_v4(); - let config = ShardWriterConfig::new(shard_id) - .with_durable_write(true) - .with_sync_indexed_write(true); + let config = ShardWriterConfig::new(shard_id).with_durable_write(true); let writer = dataset .mem_wal_writer(shard_id, config) .await @@ -7750,9 +7678,7 @@ mod shard_writer_tests { // Create shard writer let shard_id = Uuid::new_v4(); - let config = ShardWriterConfig::new(shard_id) - .with_durable_write(false) - .with_sync_indexed_write(false); + let config = ShardWriterConfig::new(shard_id).with_durable_write(false); let writer = dataset .mem_wal_writer(shard_id, config) @@ -7810,9 +7736,7 @@ mod shard_writer_tests { .expect("Failed to initialize MemWAL"); let shard_id = Uuid::new_v4(); - let config = ShardWriterConfig::new(shard_id) - .with_durable_write(true) - .with_sync_indexed_write(true); + let config = ShardWriterConfig::new(shard_id).with_durable_write(true); let writer = dataset .mem_wal_writer(shard_id, config) @@ -7951,9 +7875,7 @@ mod shard_writer_tests { // Create shard writer with default config let shard_id = Uuid::new_v4(); - let config = ShardWriterConfig::new(shard_id) - .with_durable_write(false) - .with_sync_indexed_write(false); + let config = ShardWriterConfig::new(shard_id).with_durable_write(false); let writer = dataset .mem_wal_writer(shard_id, config) @@ -8030,7 +7952,6 @@ mod shard_writer_tests { let shard_id = Uuid::new_v4(); let config = ShardWriterConfig::new(shard_id) .with_durable_write(true) // Ensure WAL files are written - .with_sync_indexed_write(true) .with_max_memtable_size(50 * 1024) // 50KB - triggers flush after ~8 batches .with_max_wal_buffer_size(10 * 1024) // 10KB WAL buffer .with_max_wal_flush_interval(Duration::from_millis(50)); // Fast flush @@ -8160,9 +8081,7 @@ mod shard_writer_tests { // rows stay invisible until the next flush. This test used to pass with // `durable_write(false)` only because an un-advanced cursor of 0 was // misread as "batch 0 is visible" — it was asserting the dirty read. - let new_config = ShardWriterConfig::new(new_shard_id) - .with_durable_write(true) - .with_sync_indexed_write(true); + let new_config = ShardWriterConfig::new(new_shard_id).with_durable_write(true); let new_writer = dataset .mem_wal_writer(new_shard_id, new_config) From be6e6fa4f37179ebc664f2629814d32ee2571080 Mon Sep 17 00:00:00 2001 From: Daniel Rammer Date: Tue, 21 Jul 2026 16:09:04 -0500 Subject: [PATCH 3/4] fix(mem_wal): reject durable memtable writes with no flush ticker MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `test_open_rejects_durable_write_without_a_ticker` asserted that `ShardWriter::open` rejects `durable_write: true` with no `max_wal_flush_interval`, but the validation was never added — the test failed on every platform (windows-build and linux-arm in CI), Linux just reported later. Since memtable puts no longer self-trigger a WAL append, a durable put becomes visible only once the background ticker drives the append. With no interval (or a zero one, which tokio can't schedule) a small put that never fills the size-triggered buffer blocks until close. Validate this at open, before claiming the epoch, matching the neighboring index/enable_memtable check. WAL-only mode is exempt: there a durable put flushes inline. Co-Authored-By: Claude Opus 4.8 (1M context) --- rust/lance/src/dataset/mem_wal/write.rs | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/rust/lance/src/dataset/mem_wal/write.rs b/rust/lance/src/dataset/mem_wal/write.rs index 082e888035d..2ee7f2e4654 100644 --- a/rust/lance/src/dataset/mem_wal/write.rs +++ b/rust/lance/src/dataset/mem_wal/write.rs @@ -1505,6 +1505,25 @@ impl ShardWriter { )); } + // A durable memtable writer needs a flush ticker to make progress. With + // `durable_write` on, a put becomes visible only once its WAL append + // lands, and memtable puts no longer self-trigger that append — the + // background ticker drives it. Without an interval (or with a zero one, + // which tokio cannot schedule), a small put that never fills the + // size-triggered buffer would block until close. Reject the config here + // rather than let a put hang. WAL-only mode is exempt: there a durable + // put triggers its own flush inline. + if config.enable_memtable + && config.durable_write + && config.max_wal_flush_interval.is_none_or(|d| d.is_zero()) + { + return Err(Error::invalid_input( + "durable_write requires a positive max_wal_flush_interval: with no \ + flush ticker a durable memtable put has nothing to drive its WAL \ + append and would block until close", + )); + } + // Callers pass the base schema; lance owns the `_tombstone` column and // appends it here so the memtable/generation schema = base + tombstone. // Idempotent, so a reopen that already extended the schema is a no-op. From c64cd784dd6680e45fc609a7b9395047cb94e675 Mon Sep 17 00:00:00 2001 From: Daniel Rammer Date: Tue, 21 Jul 2026 16:44:22 -0500 Subject: [PATCH 4/4] feat(mem_wal): drive WAL-only durable writes on the flush ticker MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit WAL-only mode still flushed the WAL inline on every durable put — one S3 PUT per put — while MemTable mode had already moved durable appends onto the background ticker. That asymmetry was an incomplete migration, not a design choice: at #6675 both modes flushed per durable put, and the ticker commit only migrated MemTable mode, leaving WAL-only on a per-trigger done cell. The done cell existed solely to propagate flush errors back to the caller; since the two-cursor work, BatchDurableWatcher::wait() surfaces the typed fence/persistence error via the poison latch, so the done cell is no longer needed. Make WAL-only consistent: - open_wal_only_mode wires max_wal_flush_interval into the flush handler and shares the pending WalOnlyState so a tick resolves against it. - flush_from_wal_only advances the writer-global durability cursor after a successful append (the FIFO queue's front sits at the watermark, so the new watermark is durable + count). - put_wal_only drops the inline done-cell flush and instead parks durable puts on the durability watermark, triggering the size/time flush for all puts like MemTable mode. A terminal append failure poisons the writer and wakes every waiter with the typed error. - The open() rejection of durable_write with no ticker now applies to both modes (the WAL-only carve-out is gone). Concurrent durable puts on a fenced writer now all wake via the poison latch rather than each racing its own flush. Adds a focused test that a durable WAL-only put is durable the instant put() returns, and parametrizes the no-ticker rejection over both modes. Co-Authored-By: Claude Opus 4.8 (1M context) --- rust/lance/src/dataset/mem_wal/wal.rs | 14 +- rust/lance/src/dataset/mem_wal/write.rs | 250 ++++++++++++++++-------- 2 files changed, 176 insertions(+), 88 deletions(-) diff --git a/rust/lance/src/dataset/mem_wal/wal.rs b/rust/lance/src/dataset/mem_wal/wal.rs index efce7b53d0a..cb2b609d71f 100644 --- a/rust/lance/src/dataset/mem_wal/wal.rs +++ b/rust/lance/src/dataset/mem_wal/wal.rs @@ -890,13 +890,15 @@ impl WalFlusher { let append_result = self.wal_appender.append(snapshot.batches).await?; let wal_io_duration = start.elapsed(); - // Append succeeded — remove the flushed batches from the front of - // the queue. Note: WAL-only mode does not use the global durability - // watermark (`durable_watermark_tx`) — `put_wal_only` waits on the - // per-trigger `done` cell instead — so we don't advance it here. - // Same for the wal-flush-completion cell, which is only consulted - // by MemTable-mode backpressure waiters. + // Append succeeded — remove the flushed batches from the front of the + // queue, then advance the writer-global durability cursor so durable + // `put_wal_only` callers waiting on their `BatchDurableWatcher` wake. + // The queue is strict FIFO with contiguous positions and its front + // always sits at the current watermark, so the newly-durable suffix + // ends at `durable + count`. Flushes are serialized on the single + // handler task, so this read-then-advance cannot race another flush. state.commit_flushed(snapshot.count); + self.advance_durable(self.durable() + snapshot.count); Ok(WalFlushResult { entry: Some(WalEntry { diff --git a/rust/lance/src/dataset/mem_wal/write.rs b/rust/lance/src/dataset/mem_wal/write.rs index 2ee7f2e4654..50a0932be49 100644 --- a/rust/lance/src/dataset/mem_wal/write.rs +++ b/rust/lance/src/dataset/mem_wal/write.rs @@ -1505,22 +1505,18 @@ impl ShardWriter { )); } - // A durable memtable writer needs a flush ticker to make progress. With - // `durable_write` on, a put becomes visible only once its WAL append - // lands, and memtable puts no longer self-trigger that append — the + // A durable writer needs a flush ticker to make progress, in either + // mode. With `durable_write` on, a put becomes durable only once its WAL + // append lands, and neither mode self-triggers that append per put — the // background ticker drives it. Without an interval (or with a zero one, // which tokio cannot schedule), a small put that never fills the // size-triggered buffer would block until close. Reject the config here - // rather than let a put hang. WAL-only mode is exempt: there a durable - // put triggers its own flush inline. - if config.enable_memtable - && config.durable_write - && config.max_wal_flush_interval.is_none_or(|d| d.is_zero()) - { + // rather than let a put hang. + if config.durable_write && config.max_wal_flush_interval.is_none_or(|d| d.is_zero()) { return Err(Error::invalid_input( "durable_write requires a positive max_wal_flush_interval: with no \ - flush ticker a durable memtable put has nothing to drive its WAL \ - append and would block until close", + flush ticker a durable put has nothing to drive its WAL append and \ + would block until close", )); } @@ -1804,6 +1800,7 @@ impl ShardWriter { let wal_handler = WalFlushHandler::new( wal_flusher.clone(), Some(state.clone()), + None, config.max_wal_flush_interval, stats.clone(), ); @@ -1876,9 +1873,25 @@ impl ShardWriter { stats: SharedWriteStats, task_executor: &Arc, ) -> Result { + // The pending queue is shared with the flush handler so a background + // tick can resolve to the batches still owed an append — the WAL-only + // analog of resolving a tick against the durability cursor in MemTable + // mode. A durable put waits on the durability cursor the handler's + // append advances, so the ticker must run (`open` rejects durable + + // no-interval); a non-durable writer may pass no interval and rely on + // the size/close triggers alone. + let state = Arc::new(WalOnlyState::default()); + // Background WAL flush handler — no MemTable state to consult, so - // pass `None` for the frozen-vs-active detection. - let wal_handler = WalFlushHandler::new(wal_flusher, None, None, stats); + // pass `None` for the frozen-vs-active detection; the pending queue and + // the flush interval drive the background append instead. + let wal_handler = WalFlushHandler::new( + wal_flusher, + None, + Some(state.clone()), + config.max_wal_flush_interval, + stats, + ); task_executor.add_handler( "wal_flusher".to_string(), Box::new(wal_handler), @@ -1893,7 +1906,7 @@ impl ShardWriter { let backpressure = BackpressureController::new(config.clone()); Ok(WriterMode::WalOnly { - state: Arc::new(WalOnlyState::default()), + state, wal_flush_tx, trigger: StdRwLock::new(WalOnlyTriggerState::default()), backpressure, @@ -2231,54 +2244,48 @@ impl ShardWriter { // Push batches into the pending queue and capture the assigned // [start, end) range. `next_batch_position` is monotonic across the // writer's lifetime; positions are not BatchStore indices but they - // are used the same way for durability tracking. + // are used the same way for durability tracking. Because the queue is + // strict FIFO with contiguous positions and its front always sits at + // the durability cursor, `end` is exactly the writer-global durable + // count this put reaches once its append lands — no globalizing offset + // as in MemTable mode, which restarts positions per generation. let batch_positions = state.push(batches); - // Time- and size-based triggers, mirroring MemTable mode but reading - // pending bytes from `WalOnlyState` instead of an active MemTable. - // Only fires for non-durable writes; durable writes go through the - // explicit done-cell path below so flush errors (e.g., fence) reach - // the caller. - if !self.config.durable_write { - let target_position = batch_positions.end; - let pending_bytes = state.estimated_size(); - self.maybe_trigger_wal_flush_wal_only( - state, - wal_flush_tx, - trigger, - target_position, - pending_bytes, - ); - } + // Under `durable_write` the put becomes durable only once its WAL + // append lands. Track it on the writer-global durability cursor *before* + // triggering, so a flush that completes between here and the wait is not + // missed: the watcher recomputes visibility from the cursor rather than + // latching a one-shot wake. WAL-only mode has no indexes, so the + // index-visibility half of the watcher is a no-op (`None`, target 0). + let durable_watcher = self + .config + .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.estimated_size(), + ); self.stats.record_put(start.elapsed()); - // For durable writes, trigger an immediate flush and wait for the - // done cell. Using the done cell instead of the durability watermark - // watcher ensures flush errors (e.g., the WalAppender returning a - // fence error) propagate back to `put` instead of hanging. - if self.config.durable_write { - let done = WatchableOnceCell::new(); - let reader = done.reader(); - self.wal_flusher.trigger_flush( - WalFlushSource::WalOnly { - state: state.clone(), - }, - batch_positions.end, - Some(done), - )?; - let mut reader = reader; - match reader.await_value().await { - Some(Ok(_)) => {} - // Rebuild the typed error (peer fence vs. persistence-failure - // self-fence) so a WAL-only durable caller can tell them apart. - Some(Err(failure)) => return Err(failure.into_error()), - None => { - return Err(Error::io( - "WAL flush handler exited before reporting durability", - )); - } - } + // Durable writes wait on the durability cursor, advanced by the append + // the ticker (or the trigger above) drives — the same watermark path + // MemTable mode uses. A terminal flush failure (a peer fence or an + // exhausted-retry persistence self-fence) poisons the writer and wakes + // the waiter with that typed error instead of leaving it to hang. This + // is why `open` rejects `durable_write` with no flush ticker: nothing + // else would advance the cursor a small put is parked on. + if let Some(mut watcher) = durable_watcher { + watcher.wait().await?; } Ok(WriteResult { batch_positions }) @@ -2876,6 +2883,10 @@ struct WalFlushHandler { /// via Arc::ptr_eq on the active batch_store. `None` when running in /// WAL-only mode (no MemTable, no frozen-vs-active distinction). memtable_state: Option>>, + /// WAL-only-mode pending queue, so a background tick can resolve to the + /// batches still owed an append. `None` in MemTable mode. Exactly one of + /// `memtable_state` / `wal_only_state` is `Some`. + wal_only_state: Option>, /// How often to append in the background. `None` disables the ticker, leaving /// the append size-triggered (and freeze/close-triggered) only. flush_interval: Option, @@ -2886,12 +2897,14 @@ impl WalFlushHandler { fn new( wal_flusher: Arc, memtable_state: Option>>, + wal_only_state: Option>, flush_interval: Option, stats: SharedWriteStats, ) -> Self { Self { wal_flusher, memtable_state, + wal_only_state, flush_interval, stats, } @@ -2993,23 +3006,43 @@ impl WalFlushHandler { /// Safe to walk the frozen list because a store that still owes an append /// cannot be swept: its L0 flush is blocked on the completion cell that only /// that append fires. + /// + /// In WAL-only mode there is a single FIFO pending queue and no memtable + /// rotation, so the ordering hazard above cannot arise: the tick resolves to + /// the queue whenever it holds un-appended batches. async fn resolve_next_pending(&self) -> Option<(WalFlushSource, usize)> { - let state_lock = self.memtable_state.as_ref()?; - let state = state_lock.read().await; - let durable = self.wal_flusher.durable(); + if let Some(state_lock) = self.memtable_state.as_ref() { + let state = state_lock.read().await; + let durable = self.wal_flusher.durable(); - next_pending_store( - state - .frozen_memtables - .iter() - .map(|frozen| frozen.memtable.batch_store()), - state.memtable.batch_store(), - durable, - ) - .map(|store| { - let end = store.len(); - (WalFlushSource::BatchStore { batch_store: store }, end) - }) + return next_pending_store( + state + .frozen_memtables + .iter() + .map(|frozen| frozen.memtable.batch_store()), + state.memtable.batch_store(), + durable, + ) + .map(|store| { + let end = store.len(); + (WalFlushSource::BatchStore { batch_store: store }, end) + }); + } + + let state = self.wal_only_state.as_ref()?; + if state.batch_count() == 0 { + // Everything already appended; the tick has nothing to do. + return None; + } + // `flush_from_wal_only` snapshots the whole queue, so the end position + // is informational here; carry the next position for symmetry. + let end = state.next_batch_position(); + Some(( + WalFlushSource::WalOnly { + state: Arc::clone(state), + }, + end, + )) } /// Unified flush method for both active and frozen memtables and for @@ -5092,8 +5125,9 @@ mod tests { .await .unwrap(); - // Two durable puts → two WAL entries (durable_write triggers an - // explicit flush per put). + // Two durable puts → two WAL entries. Each `put` awaits its own append + // (driven by the background ticker) before returning, so the second + // batch is only pushed after the first is durable — they never coalesce. let r1 = writer .put(vec![create_test_batch(&schema, 0, 4)]) .await @@ -5124,6 +5158,51 @@ mod tests { assert!(e0.writer_epoch >= 1); } + /// A durable WAL-only put is driven by the background ticker, not an inline + /// per-put flush: `put().await` must not return until the ticker's append + /// advances the durability watermark it waits on. With a long interval and a + /// buffer the write cannot cross, the ticker is the *only* thing that can + /// make the put durable — so if the WAL entry is present the instant `put` + /// returns (before any `close()`), the watermark-wait is doing its job. + #[tokio::test] + async fn test_wal_only_durable_put_waits_for_ticker_append() { + use crate::dataset::mem_wal::wal::WalTailer; + + let (store, base_path, base_uri, _temp_dir) = create_local_store().await; + let schema = create_test_schema(); + let shard_id = Uuid::new_v4(); + + let mut config = wal_only_config(shard_id); + config.max_wal_flush_interval = Some(Duration::from_millis(100)); + config.max_wal_buffer_size = 100 * 1024 * 1024; // never crossed + + let writer = ShardWriter::open( + store.clone(), + base_path.clone(), + base_uri, + config, + schema.clone(), + vec![], + ) + .await + .unwrap(); + + writer + .put(vec![create_test_batch(&schema, 0, 4)]) + .await + .unwrap(); + + // `put` returned, so the batch must already be durable — read the WAL + // directly, without closing the writer. + let tailer = WalTailer::new(store, base_path, shard_id); + assert_eq!(tailer.next_position().await.unwrap(), 2); + let entry = tailer.read_entry(1).await.unwrap().unwrap(); + assert_eq!(entry.batches.len(), 1); + assert_eq!(entry.batches[0].num_rows(), 4); + + writer.close().await.unwrap(); + } + #[tokio::test] async fn test_wal_only_rejects_index_configs() { let (store, base_path, base_uri, _temp_dir) = create_local_store().await; @@ -6015,16 +6094,21 @@ mod tests { assert!(next_pending_store(frozen_list(), Arc::clone(&active), 3).is_none()); } - /// A durable writer with no flush ticker cannot make progress, so `open()` - /// rejects it rather than letting a put block forever. + /// 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. + #[rstest] + #[case::memtable(true)] + #[case::wal_only(false)] #[tokio::test] - async fn test_open_rejects_durable_write_without_a_ticker() { + async fn test_open_rejects_durable_write_without_a_ticker(#[case] enable_memtable: bool) { let (store, base_path, base_uri, _temp_dir) = create_local_store().await; let schema = schema_with_pk(); let shard_id = Uuid::new_v4(); let config = ShardWriterConfig { durable_write: true, + enable_memtable, max_wal_flush_interval: None, ..memtable_config_with_pk(shard_id) }; @@ -6498,8 +6582,9 @@ mod tests { /// fence), the drained batches were dropped — the next concurrent put /// would then see an empty pending queue and spuriously return Ok, /// hiding the data loss. With the snapshot/commit fix, the failed flush - /// leaves the batches in the queue, and the concurrent put gets a clean - /// fence error too (when its own flush attempts the same WAL position). + /// leaves the batches in the queue for retry, and — because the flush is + /// terminal — it poisons the writer, waking *both* parked durability + /// waiters with the typed fence error instead of either one hanging. #[tokio::test] async fn test_wal_only_fenced_concurrent_puts_do_not_silently_succeed() { use std::sync::Arc; @@ -6550,9 +6635,10 @@ mod tests { // the destructive-drain bug, the first flush would consume both // pending batches into a failing append; the second flush would // see an empty queue and return spurious success, silently losing - // the second put's data. With the snapshot/commit fix, the failed - // append leaves both batches in the queue and the second flush - // also fails with the fence error. + // the second put's data. Now both puts park on the durability + // watermark; the ticker's append fails with the fence, poisons the + // writer, and both waiters wake with the fence error — batches intact + // in the queue. let a1 = writer_a.clone(); let a2 = writer_a.clone(); let schema1 = schema.clone();