feat: add write ahead log appender and tailer primitives (#6669)

## Summary

Surface a generic read/write surface for MemWAL shards so callers can
drive a shard directly without going through the existing flusher.

- **Shard self-description.** `ShardManifest` now records the
`(shard_spec_id, field_id -> bytes)` assignment that produced the shard,
so a manifest found on disk can be mapped back to a shard spec without
consulting the inline index. Proto adds `ShardFieldEntry`; manifest
carries `shard_field_values: HashMap<String, Vec<u8>>`.
- **Idempotent shard initialization.**
`ShardManifestStore::initialize_shard()` writes manifest v1 at writer
epoch 0 and treats `AlreadyExists` as success when the existing manifest
matches.
- **Generic WAL primitives.**
- `WalAppender::open()` claims a writer epoch via the manifest store;
`append(batches)` serializes Arrow IPC and writes with
put-if-not-exists, retrying on conflict and fencing on PUT failure.
- `WalTailer::read_entry`, `next_position`, `first_position` use
`wal_entry_position_last_seen` as a probe hint with a listing fallback.
`WalReadEntry` includes the `writer_epoch` recorded in the entry so
callers can fence-check on replay.
- **Index discovery helpers.** New `Dataset` methods
`mem_wal_index_details()` and `list_mem_wal_latest_shard_ids()`
(object-storage directory listing). The inline shard snapshot is
positioned as a stale read-optimization rather than the source of truth;
`docs/src/format/table/mem_wal.md` is updated to match.

## Tests

Added unit tests for:
- `ShardManifestStore::initialize_shard` happy path, idempotent on
match, rejects mismatched conflict.
- `WalAppender` / `WalTailer` round-trip including `writer_epoch`
propagation; position increment.
- Writer-epoch fencing: a stale appender hits the conflict path on
append and surfaces the fence error.
- Input validation: empty batch list and zero-row batches are rejected.
- `WalTailer::with_cursor_updates(true)` updates
`wal_entry_position_last_seen` asynchronously, and `next_position()`
still resolves correctly.
- `mem_wal_path()` helper.

cc @jackye1995 for review.

Author: touch-of-grey

docs/src/format/table/mem_wal.md

@@ -45,13 +45,14 @@ A table has at most one MemWAL index. It stores:
 - **Configuration**: Shard specs defining how rows map to shards, and which indexes to maintain
 - **Merge progress**: Last generation merged to base table for each shard
 - **Index catchup progress**: Which merged generation each base table index has been rebuilt to cover
-- **Shard snapshots**: Snapshot of all shard states for read optimization
+- **Shard snapshots**: Point-in-time snapshot of shard states for read optimization
 
 The index is the source of truth for **configuration**, **merge progress** and **index catchup progress**
 Writers and mergers read the MemWAL index to get these configurations before writing.
 
 Each [shard's manifest](#shard-manifest) is authoritative for its own state.
-Readers use **shard snapshots** is a read-only optimization to see a point-in-time view of all shards without the need to open each shard manifest.
+Readers may use **shard snapshots** as a read-only optimization to see a point-in-time view of shards without opening each shard manifest.
+Readers that need the latest shard set must discover shard directories in storage and read each shard's latest manifest.
 
 See [MemWAL Index Details](#memwal-index-details) for the complete structure.
 
@@ -173,6 +174,7 @@ Each shard has a manifest file. This is the source of truth for the state of a s
 The manifest contains:
 
 - **Fencing state**: `writer_epoch` as the latest writer fencing token, see [Writer Fencing](#writer-fencing) for more details.
+- **Shard assignment**: `shard_spec_id` and `shard_field_values` record how this shard maps to its shard spec. `shard_field_values` is a map from shard field id to the raw Arrow scalar bytes of the computed value; the matching `ShardField.result_type` in the `ShardSpec` determines how to interpret each entry (e.g., 4 little-endian bytes for int32, raw UTF-8 bytes for utf8).
 - **WAL pointers**: `replay_after_wal_entry_position` (last entry position flushed to MemTable, 0-based), `wal_entry_position_last_seen` (last entry position seen at manifest update, 0-based)
 - **Generation trackers**: `current_generation` (next generation to flush), `flushed_generations` list of generation number and directory path pairs (e.g., generation 1 at `a1b2c3d4_gen_1`)
 
@@ -258,8 +260,17 @@ The `index_details` field in `IndexMetadata` contains a `MemWalIndexDetails` pro
 
 ### Shard Identifier
 
-Each shard has a unique identifier across all shards following UUID v4 standard.
-When a new shard is created, it is assigned a new identifier.
+Each shard has a unique UUID identifier within the table.
+When a new shard is created, implementations may assign either a random UUID or
+a deterministic UUID derived from the shard assignment when deterministic
+writer fencing is required.
+
+### Shard Discovery
+
+The MemWAL index can store shard snapshots for read optimization, but those snapshots may lag the latest shard set.
+Implementations that need to discover the current shard set should list `_mem_wal/` shard directories and read each shard's latest [shard manifest](#shard-manifest).
+
+Each shard manifest records the shard UUID, shard spec ID, and computed shard field values needed to map the shard back to a shard spec assignment.
 
 ### Shard Spec
 
@@ -342,28 +353,25 @@ This file uses standard Lance format with the shard snapshot schema, enabling ef
 ### Shard Snapshot Arrow Schema
 
 Shard snapshots are stored as a Lance file with one row per shard.
-The schema has one column per `ShardManifest` field plus shard spec columns:
-
-| Column                            | Type                                             | Description                                              |
-| --------------------------------- | ------------------------------------------------ | -------------------------------------------------------- |
-| `shard_id`                       | `fixed_size_binary(16)`                          | Shard UUID bytes                                        |
-| `version`                         | `uint64`                                         | Shard manifest version                                  |
-| `shard_spec_id`                  | `uint32`                                         | Shard spec ID (0 if manual)                             |
-| `writer_epoch`                    | `uint64`                                         | Writer fencing token                                     |
-| `replay_after_wal_entry_position` | `uint64`                                         | Last WAL entry position (0-based) flushed to MemTable    |
-| `wal_entry_position_last_seen`    | `uint64`                                         | Last WAL entry position (0-based) seen (hint)            |
-| `current_generation`              | `uint64`                                         | Next generation to flush                                 |
-| `flushed_generations`             | `list<struct<generation: uint64, path: string>>` | Flushed MemTable paths                                   |
-| `region_field_{field_id}`         | varies                                           | Shard field value (one column per field in shard spec) |
+The snapshot schema is optimized for shard discovery. Full mutable shard state
+remains in the authoritative shard manifest files.
+
+| Column                   | Type          | Description                                                                                                |
+| ------------------------ | ------------- | ---------------------------------------------------------------------------------------------------------- |
+| `shard_id`               | `utf8`        | Shard UUID string                                                                                          |
+| `shard_spec_id`          | `uint32`      | Shard spec ID (0 if manual)                                                                                |
+| `shard_field_{field_id}` | varies        | One column per shard field defined in the shard spec, typed to match the field's `ShardField.result_type`. |
 
 For example, with a shard spec containing a field `user_bucket` of type `int32`:
 
 | Column                     | Type    | Description                  |
 | -------------------------- | ------- | ---------------------------- |
 | ...                        | ...     | (base columns above)         |
-| `region_field_user_bucket` | `int32` | Bucket value for this shard |
+| `shard_field_user_bucket`  | `int32` | Bucket value for this shard |
 
-This schema directly corresponds to the fields in the `ShardManifest` protobuf message plus the computed shard field values.
+This schema records the fields needed to map each shard back to its shard spec
+assignment. Readers that need fencing epochs, WAL positions, or flushed
+generation state must read the latest shard manifests directly.
 
 ## Storage Layout
 

protos/table.proto

@@ -537,6 +537,12 @@ message ShardManifest {
   // A value of 0 indicates a manually-created shard not governed by any spec.
   uint32 shard_spec_id = 10;
 
+  // Computed shard field values as raw Arrow scalar bytes, keyed by shard
+  // field id. The byte encoding follows Arrow's little-endian convention:
+  // int32 is 4 LE bytes, utf8 is raw UTF-8 bytes, etc. The receiver looks
+  // up the result_type from the ShardSpec to interpret each value.
+  repeated ShardFieldEntry shard_field_entries = 14;
+
   // Writer fencing token - monotonically increasing.
   // A writer must increment this when claiming the shard.
   uint64 writer_epoch = 2;
@@ -559,6 +565,16 @@ message ShardManifest {
   repeated FlushedGeneration flushed_generations = 8;
 }
 
+// A shard field value stored as raw Arrow scalar bytes.
+message ShardFieldEntry {
+  // Shard field id (matches ShardField.field_id in the ShardSpec).
+  string field_id = 1;
+
+  // Raw Arrow scalar value bytes in little-endian encoding.
+  // The data type is determined by the result_type of the matching ShardField.
+  bytes value = 2;
+}
+
 // A flushed MemTable generation and its storage location.
 message FlushedGeneration {
   // Generation number.
@@ -596,20 +612,17 @@ message IndexCatchupProgress {
 // - Shard state snapshots
 //
 // Writers read this index to get configuration before writing.
-// Readers read this index to discover shards and their state.
+// Readers may use shard snapshots in this index as a point-in-time
+// optimization. Readers that need the latest shard set should list shard
+// directories in storage and read each shard's latest manifest.
 // A background process updates the index periodically to keep shard snapshots current.
 //
 // Shard snapshots are stored as a Lance file with one row per shard.
-// The schema has one column per ShardManifest field, with shard fields as columns:
-//   shard_id: fixed_size_binary(16)  -- UUID bytes
-//   version: uint64
+// The schema records shard discovery fields. Full mutable shard state remains
+// authoritative in the shard manifest files.
+//   shard_id: utf8
 //   shard_spec_id: uint32
-//   writer_epoch: uint64
-//   replay_after_wal_entry_position: uint64
-//   wal_entry_position_last_seen: uint64
-//   current_generation: uint64
-//   merged_generation: uint64
-//   flushed_generations: list<struct<generation: uint64, path: string>>
+//   shard_field_{field_id}: typed per the matching ShardField.result_type
 message MemWalIndexDetails {
   // Snapshot timestamp (Unix timestamp in milliseconds).
   int64 snapshot_ts_millis = 1;
@@ -690,4 +703,3 @@ message ShardField {
   // Transform parameters (e.g., num_buckets for bucket transform).
   map<string, string> parameters = 6;
 }
-

rust/lance-index/src/mem_wal.rs

@@ -153,6 +153,11 @@ pub struct ShardManifest {
     pub shard_id: Uuid,
     pub version: u64,
     pub shard_spec_id: u32,
+    /// Computed shard field values as raw Arrow scalar bytes, keyed by field id.
+    /// The byte encoding follows Arrow's little-endian convention: int32 is 4 LE
+    /// bytes, utf8 is raw UTF-8 bytes, etc. The result_type in the corresponding
+    /// 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`.
@@ -165,7 +170,8 @@ pub struct ShardManifest {
 
 impl DeepSizeOf for ShardManifest {
     fn deep_size_of_children(&self, context: &mut deepsize::Context) -> usize {
-        self.flushed_generations.deep_size_of_children(context)
+        self.shard_field_values.deep_size_of_children(context)
+            + self.flushed_generations.deep_size_of_children(context)
     }
 }
 
@@ -175,6 +181,14 @@ impl From<&ShardManifest> for pb::ShardManifest {
             shard_id: Some((&rm.shard_id).into()),
             version: rm.version,
             shard_spec_id: rm.shard_spec_id,
+            shard_field_entries: rm
+                .shard_field_values
+                .iter()
+                .map(|(k, v)| pb::ShardFieldEntry {
+                    field_id: k.clone(),
+                    value: v.clone(),
+                })
+                .collect(),
             writer_epoch: rm.writer_epoch,
             replay_after_wal_entry_position: rm.replay_after_wal_entry_position,
             wal_entry_position_last_seen: rm.wal_entry_position_last_seen,
@@ -193,10 +207,16 @@ impl TryFrom<pb::ShardManifest> for ShardManifest {
             .as_ref()
             .map(Uuid::try_from)
             .ok_or_else(|| Error::invalid_input("Missing shard_id in ShardManifest"))??;
+        let shard_field_values = rm
+            .shard_field_entries
+            .into_iter()
+            .map(|e| (e.field_id, e.value))
+            .collect();
         Ok(Self {
             shard_id,
             version: rm.version,
             shard_spec_id: rm.shard_spec_id,
+            shard_field_values,
             writer_epoch: rm.writer_epoch,
             replay_after_wal_entry_position: rm.replay_after_wal_entry_position,
             wal_entry_position_last_seen: rm.wal_entry_position_last_seen,

rust/lance/src/dataset/mem_wal.rs

@@ -41,9 +41,10 @@ mod util;
 mod wal;
 pub mod write;
 
-pub use api::{DatasetMemWalExt, MemWalConfig};
+pub use api::{DatasetMemWalExt, MemWalConfig, MemWalShardConfig};
 pub use manifest::ShardManifestStore;
 pub use memtable::scanner::MemTableScanner;
 pub use scanner::{LsmDataSource, LsmGeneration, LsmScanner, ShardSnapshot};
+pub use wal::{WalAppendResult, WalAppender, WalReadEntry, WalTailer};
 pub use write::ShardWriter;
 pub use write::ShardWriterConfig;