fix(mem_wal): stop WAL replay from re-loading already-compacted entries (lance-format#6767)

## Summary

After a writer flushed a memtable to L0 and an external compactor merged
that generation into the base table — legitimately draining
`flushed_generations` to empty — a subsequent restart re-replayed the
original WAL entries into the new active memtable, duplicating rows on
read.

Two bugs were interacting:

1. **Disambiguation:** `replay_memtable_from_wal` distinguished "fresh
shard" from "flushed and compacted" via
`flushed_generations.is_empty()`. That works in a closed-world
deployment but breaks the moment an external compactor enters the
picture — and the compactor is the *intended* consumer that drains that
vector, so the signal is structurally broken under OSS-WAL.

2. **Cursor never advanced:** `MemTableFlusher::flush` read
`covered_wal_entry_position` from
`memtable.last_flushed_wal_entry_position()`, but that field is only set
by the `mark_wal_flushed` test helper. In production it stayed at 0, so
`replay_after_wal_entry_position` never advanced past 0. Under 0-based
WAL positions this masked bug #1 — both "fresh" and "post-flush-of-0"
produced cursor=0.

## Fix

- **WAL positions are now 1-based** (`FIRST_WAL_ENTRY_POSITION = 1`). A
cursor of `0` unambiguously means "no flush has stamped this shard," so
replay collapses to `cursor.saturating_add(1)` without consulting
`flushed_generations`.
- **`WalFlushHandler::handle`** writes the just-appended position back
into `state.last_flushed_wal_entry_position` under the state lock before
signalling the completion cell.
- **`MemTableFlusher::flush` / `flush_with_indexes`** now take an
explicit `covered_wal_entry_position` arg. The production caller derives
it per-memtable from the `WalFlushResult` carried in the completion cell
— authoritative under concurrent flushes — falling back to
`memtable.frozen_at_wal_entry_position()` when freeze did not trigger a
flush.
- **State seed at open** uses the post-replay WAL tip, not
`manifest.wal_entry_position_last_seen` (the latter is bumped on every
tailer read and can sit above any flushed generation).
- Proto field docs on `ShardManifest.replay_after_wal_entry_position` /
`wal_entry_position_last_seen` updated to spell out the 1-based
convention and what default-0 means.

## Test plan

- [x] Added
`test_memtable_replay_skips_entries_after_external_compaction` in
`rust/lance/src/dataset/mem_wal/write.rs`: open writer, put rows, close
(flush), simulate the compactor by directly committing a manifest with
empty `flushed_generations`, reopen, assert the memtable is empty. Fails
on the pre-fix code; passes now.
- [x] `cargo test -p lance --lib dataset::mem_wal` — 236/236 pass
- [x] `cargo test -p lance --lib` — 1600/1600 pass
- [x] `cargo test -p lance-index --lib` — 302/302 pass
- [x] `cargo clippy --all --tests --benches -- -D warnings` — clean
- [x] `cargo fmt --all -- --check` — clean

## Compatibility

WAL position numbering changes from 0-based to 1-based. Existing on-disk
manifests / WAL files written by the prior `oss-wal-multiplex` code are
not migrated — coordinated with downstream consumers (sophon) to start
fresh.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: hamersaw

protos/table.proto

@@ -547,12 +547,16 @@ message ShardManifest {
   // A writer must increment this when claiming the shard.
   uint64 writer_epoch = 2;
 
-  // The most recent WAL entry position (0-based) that has been flushed to a MemTable.
+  // The most recent WAL entry position that has been flushed to a MemTable.
   // During recovery, replay starts from replay_after_wal_entry_position + 1.
+  // WAL positions are 1-based, so the default value 0 unambiguously means
+  // "no flush has ever stamped this shard" and recovery replays from 1.
   uint64 replay_after_wal_entry_position = 3;
 
-  // The most recent WAL entry position (0-based) at the time manifest was updated.
-  // This is a hint, not authoritative - recovery must list files to find actual state.
+  // The most recent WAL entry position observed at the time the manifest was
+  // updated. WAL positions are 1-based; default 0 means no entry has been
+  // written yet. This is a hint, not authoritative - recovery must list
+  // files to find actual state.
   uint64 wal_entry_position_last_seen = 4;
 
   // Next generation ID to create (incremented after each MemTable flush).

rust/lance-index/src/mem_wal.rs

@@ -159,10 +159,15 @@ pub struct ShardManifest {
     /// ShardField from the ShardSpec determines how to interpret each value.
     pub shard_field_values: HashMap<String, Vec<u8>>,
     pub writer_epoch: u64,
-    /// The most recent WAL entry position (0-based) flushed to a MemTable.
-    /// Recovery replays from `replay_after_wal_entry_position + 1`.
+    /// The most recent WAL entry position flushed to a MemTable.
+    /// Recovery replays from `replay_after_wal_entry_position + 1`. The
+    /// default value 0 means "no flush has ever stamped this shard" — WAL
+    /// positions themselves are 1-based, so 0 is never a valid covered
+    /// position.
     pub replay_after_wal_entry_position: u64,
-    /// The most recent WAL entry position (0-based) when manifest was updated.
+    /// The most recent WAL entry position observed at manifest write time.
+    /// Default 0 means "no entry has been written yet"; WAL positions are
+    /// 1-based.
     pub wal_entry_position_last_seen: u64,
     pub current_generation: u64,
     pub flushed_generations: Vec<FlushedGeneration>,

rust/lance/benches/mem_wal_index_micro.rs

@@ -396,9 +396,12 @@ async fn measure_flush(
     let (epoch, _) = manifest_store.claim_epoch(0).await?;
     let flusher = MemTableFlusher::new(store, base_path, uri, shard_id, manifest_store);
 
+    // total_batches WAL entries were stamped at positions 1..=total_batches
+    // by the mark_wal_flushed loop above (1-based positions).
+    let covered_wal_entry_position = total_batches as u64;
     let t = Instant::now();
     let _result = flusher
-        .flush_with_indexes(&memtable, epoch, index_configs)
+        .flush_with_indexes(&memtable, epoch, index_configs, covered_wal_entry_position)
         .await?;
     let elapsed = t.elapsed();
 

rust/lance/src/dataset/mem_wal/memtable/flush.rs

@@ -79,8 +79,19 @@ impl MemTableFlusher {
     }
 
     /// Flush the MemTable to storage (data files, indexes, bloom filter).
+    ///
+    /// `covered_wal_entry_position` is stamped into the manifest's
+    /// `replay_after_wal_entry_position` so post-restart replay skips the
+    /// WAL entries this generation captures. Pass 0 only for shards that
+    /// have not yet appended any WAL entry — non-zero positions are
+    /// 1-based (see `FIRST_WAL_ENTRY_POSITION`).
     #[instrument(name = "mt_flush_storage", level = "info", skip_all, fields(shard_id = %self.shard_id, epoch, generation = memtable.generation(), row_count = memtable.row_count()))]
-    pub async fn flush(&self, memtable: &MemTable, epoch: u64) -> Result<FlushResult> {
+    pub async fn flush(
+        &self,
+        memtable: &MemTable,
+        epoch: u64,
+        covered_wal_entry_position: u64,
+    ) -> Result<FlushResult> {
         self.manifest_store.check_fenced(epoch).await?;
 
         if memtable.row_count() == 0 {
@@ -113,9 +124,13 @@ impl MemTableFlusher {
         self.write_bloom_filter(&bloom_path, memtable.bloom_filter())
             .await?;
 
-        let last_wal_entry_position = memtable.last_flushed_wal_entry_position();
         let new_manifest = self
-            .update_manifest(epoch, generation, &gen_folder_name, last_wal_entry_position)
+            .update_manifest(
+                epoch,
+                generation,
+                &gen_folder_name,
+                covered_wal_entry_position,
+            )
             .await?;
 
         info!(
@@ -129,7 +144,7 @@ impl MemTableFlusher {
                 path: gen_folder_name,
             },
             rows_flushed,
-            covered_wal_entry_position: last_wal_entry_position,
+            covered_wal_entry_position,
         })
     }
 
@@ -184,12 +199,16 @@ impl MemTableFlusher {
     }
 
     /// Flush the MemTable to storage with indexes.
+    ///
+    /// See [`MemTableFlusher::flush`] for `covered_wal_entry_position`
+    /// semantics.
     #[instrument(name = "mt_flush_with_indexes", level = "info", skip_all, fields(shard_id = %self.shard_id, epoch, generation = memtable.generation(), row_count = memtable.row_count(), index_count = index_configs.len()))]
     pub async fn flush_with_indexes(
         &self,
         memtable: &MemTable,
         epoch: u64,
         index_configs: &[MemIndexConfig],
+        covered_wal_entry_position: u64,
     ) -> Result<FlushResult> {
         self.manifest_store.check_fenced(epoch).await?;
 
@@ -288,9 +307,13 @@ impl MemTableFlusher {
         self.write_bloom_filter(&bloom_path, memtable.bloom_filter())
             .await?;
 
-        let last_wal_entry_position = memtable.last_flushed_wal_entry_position();
         let new_manifest = self
-            .update_manifest(epoch, generation, &gen_folder_name, last_wal_entry_position)
+            .update_manifest(
+                epoch,
+                generation,
+                &gen_folder_name,
+                covered_wal_entry_position,
+            )
             .await?;
 
         info!(
@@ -304,7 +327,7 @@ impl MemTableFlusher {
                 path: gen_folder_name,
             },
             rows_flushed: memtable.row_count(),
-            covered_wal_entry_position: last_wal_entry_position,
+            covered_wal_entry_position,
         })
     }
 
@@ -883,7 +906,7 @@ mod tests {
         assert!(!memtable.all_flushed_to_wal());
 
         let flusher = MemTableFlusher::new(store, base_path, base_uri, shard_id, manifest_store);
-        let result = flusher.flush(&memtable, epoch).await;
+        let result = flusher.flush(&memtable, epoch, 0).await;
 
         assert!(result.is_err());
         assert!(
@@ -912,7 +935,7 @@ mod tests {
         let memtable = MemTable::new(schema, 1, vec![]).unwrap();
 
         let flusher = MemTableFlusher::new(store, base_path, base_uri, shard_id, manifest_store);
-        let result = flusher.flush(&memtable, epoch).await;
+        let result = flusher.flush(&memtable, epoch, 0).await;
 
         assert!(result.is_err());
         assert!(result.unwrap_err().to_string().contains("empty MemTable"));
@@ -950,7 +973,7 @@ mod tests {
             shard_id,
             manifest_store.clone(),
         );
-        let result = flusher.flush(&memtable, epoch).await.unwrap();
+        let result = flusher.flush(&memtable, epoch, 1).await.unwrap();
 
         assert_eq!(result.generation.generation, 1);
         assert_eq!(result.rows_flushed, 10);
@@ -1011,7 +1034,7 @@ mod tests {
             manifest_store.clone(),
         );
         let result = flusher
-            .flush_with_indexes(&memtable, epoch, &index_configs)
+            .flush_with_indexes(&memtable, epoch, &index_configs, 1)
             .await
             .unwrap();
 
@@ -1144,7 +1167,7 @@ mod tests {
             manifest_store.clone(),
         );
         let result = flusher
-            .flush_with_indexes(&memtable, epoch, &index_configs)
+            .flush_with_indexes(&memtable, epoch, &index_configs, 1)
             .await
             .unwrap();
 
@@ -1271,7 +1294,7 @@ mod tests {
             manifest_store.clone(),
         );
         let result = flusher
-            .flush_with_indexes(&memtable, epoch, &index_configs)
+            .flush_with_indexes(&memtable, epoch, &index_configs, 1)
             .await
             .unwrap();
 

rust/lance/src/dataset/mem_wal/wal.rs

@@ -94,7 +94,7 @@ impl std::fmt::Debug for BatchDurableWatcher {
 /// A single WAL entry representing a batch of batches.
 #[derive(Debug, Clone)]
 pub struct WalEntry {
-    /// WAL entry position (0-based, sequential).
+    /// WAL entry position (1-based, sequential — see `FIRST_WAL_ENTRY_POSITION`).
     pub position: u64,
     /// Writer epoch at the time of write.
     pub writer_epoch: u64,
@@ -664,7 +664,12 @@ impl WalEntryData {
 // Generic WAL Appender and Tailer
 // ============================================================================
 
-const FIRST_WAL_ENTRY_POSITION: u64 = 0;
+/// First valid WAL entry position. Positions are 1-based so that a
+/// `ShardManifest::replay_after_wal_entry_position` of 0 unambiguously means
+/// "no flush has ever stamped the cursor" — replay then starts at position 1
+/// without needing to consult `flushed_generations`, which an external
+/// compactor may legitimately drain back to empty.
+const FIRST_WAL_ENTRY_POSITION: u64 = 1;
 const MAX_APPEND_CREATE_CONFLICTS: usize = 1024;
 const APPEND_CONFLICT_REFRESH_INTERVAL: usize = 16;
 const MAX_CURSOR_PROBE: u64 = 4096;
@@ -1389,8 +1394,9 @@ mod tests {
         let source = batch_store_source(&batch_store);
         let result = buffer.flush(&source, batch_store.len()).await.unwrap();
         let entry = result.entry.unwrap();
-        // First entry from a freshly-discovered position is 0 (atomic-create
-        // path discovers the tip via list and starts at FIRST_WAL_ENTRY_POSITION).
+        // First entry from a freshly-discovered position lands at
+        // FIRST_WAL_ENTRY_POSITION (atomic-create path discovers the tip
+        // via list).
         assert_eq!(entry.position, FIRST_WAL_ENTRY_POSITION);
         assert_eq!(entry.writer_epoch, 1);
         assert_eq!(entry.num_batches, 2);
@@ -1629,6 +1635,7 @@ mod tests {
         assert!(hint >= 1, "cursor hint never updated, last={hint}");
 
         // next_position must still resolve to one past the last appended entry.
-        assert_eq!(tailer.next_position().await.unwrap(), 3);
+        // Three entries from a fresh shard land at 1, 2, 3, so next is 4.
+        assert_eq!(tailer.next_position().await.unwrap(), 4);
     }
 }