feat: unify WalAppender into ShardWriter via enable_memtable mode (#6675)

## Summary

- Add `ShardWriterConfig::enable_memtable` (default `true`); when
`false`, `ShardWriter` runs in a new WAL-only mode that keeps the async
batched WAL pipeline but skips MemTable allocation, in-memory indexes,
MemTable freezing, and MemTable-bytes backpressure (a separate WAL-only
backpressure budget reuses `max_unflushed_memtable_bytes`).
- `WalAppender` becomes the WAL write primitive used by both modes
inside `WalFlusher`, replacing the prior duplicate plain-`put` path.
MemTable mode now also gets atomic put-if-not-exists, conflict retry,
and fence-on-write.
- `WalAppender` stays public as the lowest-level synchronous-atomic
appender. New `pub(crate) WalAppender::with_claimed_epoch` lets
`ShardWriter::open` claim the epoch once and inject it.
- WAL-only mode uses a drainable `WalOnlyState` queue with
snapshot/commit semantics so a failed append leaves batches in the queue
for retry instead of dropping them silently.
- `memtable_stats()`, `scan()`, `active_memtable_ref()` now return
`Result<...>` and produce a clear `invalid_input` error in WAL-only
mode.

Behavior change to call out: first WAL entry on a fresh shard is now
position 0 instead of 1, matching the 0-based positions documented in
the [MemTable & WAL
spec](https://github.com/lance-format/lance/blob/main/docs/src/format/table/mem_wal.md).
The previous flusher seeded its counter from
`wal_entry_position_last_seen + 1` and so always skipped position 0.

Context: this realizes the layering discussed in
#6669 (comment) —
keep `WalAppender` as the low-level primitive and let users always use
`ShardWriter`, with a config switch to turn the MemTable on or off.

cc @jackye1995

---------

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

Author: touch-of-grey

python/src/mem_wal.rs

@@ -201,7 +201,7 @@ impl PyRegionWriter {
                 // Snapshot stats before close so the captured state reflects
                 // what was written, not any internal bookkeeping done by close().
                 let stats_snapshot = stats_handle.snapshot();
-                let memtable_stats_before_close = writer.memtable_stats().await;
+                let memtable_stats_before_close = writer.memtable_stats().await?;
                 writer.close().await?;
                 let closed_memtable_stats = closed_memtable_stats(memtable_stats_before_close);
                 let mut closed_guard = closed_state.lock().await;
@@ -255,7 +255,7 @@ impl PyRegionWriter {
             .block_on(Some(py), async move {
                 let guard = inner.lock().await;
                 match guard.as_ref() {
-                    Some(w) => Ok(w.memtable_stats().await),
+                    Some(w) => w.memtable_stats().await,
                     None => {
                         let closed_guard = closed_state.lock().await;
                         closed_guard
@@ -267,7 +267,7 @@ impl PyRegionWriter {
                     }
                 }
             })?
-            .map_err(|e| PyIOError::new_err(e.to_string()))?;
+            .map_err(|e: lance::Error| PyIOError::new_err(e.to_string()))?;
 
         memtable_stats_to_pydict(py, &stats)
     }
@@ -296,7 +296,7 @@ impl PyRegionWriter {
                 let guard = inner.lock().await;
                 match guard.as_ref() {
                     Some(w) => {
-                        let active_ref = w.active_memtable_ref().await;
+                        let active_ref = w.active_memtable_ref().await?;
                         let writer_snapshot = w
                             .manifest()
                             .await?

rust/lance/Cargo.toml

@@ -210,5 +210,9 @@ harness = false
 name = "mem_wal_read"
 harness = false
 
+[[bench]]
+name = "mem_wal_replay"
+harness = false
+
 [lints]
 workspace = true

rust/lance/benches/mem_wal_read.rs

@@ -307,7 +307,7 @@ async fn setup_benchmark(
     let manifest = writer.manifest().await.unwrap();
 
     // Get active memtable reference
-    let active_memtable_ref = writer.active_memtable_ref().await;
+    let active_memtable_ref = writer.active_memtable_ref().await.unwrap();
 
     // Build shard snapshot
     let mut shard_snapshot = ShardSnapshot::new(shard_id);
@@ -906,7 +906,7 @@ async fn setup_vector_benchmark(
     }
 
     let manifest = writer.manifest().await.unwrap();
-    let active_memtable_ref = writer.active_memtable_ref().await;
+    let active_memtable_ref = writer.active_memtable_ref().await.unwrap();
 
     let mut shard_snapshot = ShardSnapshot::new(shard_id);
     if let Some(ref m) = manifest {