[upstream/lance] Provide a first-class MemWAL shard compaction API (dataset.compact_mem_wal_shard)

[beinan]

[upstream/lance] — this tracks a change that ideally lands in Lance (the MemWAL owner). Filed here for internal triage; move upstream when ready.

Summary

Lance's MemWAL design explicitly anticipates an "external compactor" that folds flushed generations back into the base table, but ships no API to do it. Every consumer must hand-roll the full sequence: read each flushed generation dataset, append its rows to the base table, claim the shard epoch, surgically drain the merged generations from the shard manifest, and delete the merged generations' blob directories — in a crash-safe order. This is subtle, easy to get wrong (data loss, storage leaks, epoch-fence bugs), and identical for every MemWAL user. It belongs in Lance as dataset.compact_mem_wal_shard(shard_id) (or similar).

Motivation

lance-context implements this today as ~120 lines in RolloutStore::merge_own_shard (crates/lance-context-core/src/rollout_store.rs:667). The full logic, with the non-obvious correctness requirements it encodes:

1. Read every flushed generation. For each flushed in manifest.flushed_generations, open its dataset at {base}/_mem_wal/{shard}/{path}, scan all rows, and record which generation ids and paths were merged.

2. Append merged rows to the base table via Dataset::append (Append mode). Append vs a concurrent Rewrite is non-conflicting in Lance's matrix, so this is safe alongside compaction.

3. Surgical drain — NOT a blanket clear. Claim the shard epoch, then commit_update a new manifest that retains every generation except the exact ids just merged:

   let (epoch, _) = manifest_store.claim_epoch(manifest.shard_spec_id).await?;
   manifest_store.commit_update(epoch, |current| ShardManifest {
       version: current.version + 1,
       flushed_generations: current.flushed_generations.iter()
           .filter(|fg| !merged_generations.contains(&fg.generation))
           .cloned().collect(),
       ..current.clone()
   }).await?;

   A blanket flushed_generations = [] would silently discard any generation that landed between reading the manifest and committing the drain — data loss. commit_update re-reads the latest manifest and applies the closure to current state, so the retain filter must run against current state, not the stale snapshot.

4. Delete blob directories second, never first. Only after the manifest no longer references a generation may its _mem_wal/{shard}/{path}/ directory be removed. Deleting before the drain would let a reader resolve a manifest entry whose data is gone. If the process dies between drain and delete, rows are already in the base table and the manifest no longer lists them, so nothing reads them — the directory is just reclaimable garbage. Skipping this deletion is a permanent storage leak (every merged generation's directory left behind forever); a delete failure must be best-effort (warn, don't fail the merge).

5. replay_after_wal_entry_position must be left untouched so a reopened writer does not re-replay already-merged WAL entries.

6. Crash recovery relies on caller-side row identity. If a crash interrupts between step 2 and step 3 (appended to base, manifest not yet drained), a subsequent read sees the rows via both the base table and the still-listed generation. The downstream dedups by a primary key, so no double count. Lance's API should either handle idempotency internally or document this contract precisely.

7. Epoch-claim fencing. claim_epoch bumps the writer epoch, fencing any other live writer of the shard. This is only safe under a single-writer-per-shard model where the compactor owns the shard. If a resident writer for the shard exists in the same process, it must be closed before the claim (else it gets fenced). The API should make this ordering explicit or manage it internally.

Goal

A single Lance call — e.g. dataset.compact_mem_wal_shard(shard_id) -> Result<usize> returning generations reclaimed — that performs steps 1–5 correctly, with 4/6/7's crash-safety and ordering guaranteed inside Lance, so downstreams stop re-deriving them.

Proposed API

impl Dataset {
    /// Fold this shard's flushed MemWAL generations into the base table and
    /// drain exactly those generations from the shard manifest, deleting their
    /// on-storage directories. Returns the number of generations reclaimed
    /// (0 if none pending). Caller must own/write only this shard (single-writer
    /// invariant); any resident ShardWriter for `shard_id` is closed first.
    pub async fn compact_mem_wal_shard(&mut self, shard_id: Uuid) -> Result<usize>;

    /// Optional: threshold variant — only compact when at least `min_generations`
    /// are pending, so callers don't read the manifest just to decide.
    pub async fn compact_mem_wal_shard_if_ready(
        &mut self, shard_id: Uuid, min_generations: usize,
    ) -> Result<usize>;
}

Where to look (Lance side)

• ShardManifestStore (read_latest, claim_epoch, commit_update) — the manifest primitives this must sit on top of.
• ShardManifest / flushed_generations / shard_spec_id / replay_after_wal_entry_position.
• Dataset::append, Dataset::mem_wal_writer, and the _mem_wal/{shard}/{generation} on-storage layout.
• ObjectStore::remove_dir_all for the blob-directory cleanup.

Acceptance criteria

• One public Dataset method performs read → append → surgical drain → blob delete, in a crash-safe order, for a given shard.
• Drain removes only the generations actually merged; a generation flushed concurrently is retained, not dropped (regression test: inject a new generation between read and drain, assert it survives and is not lost).
• Blob directories of merged generations are deleted after the drain; a delete failure is best-effort and does not fail the compaction (test: fault-inject a delete error, assert compaction still reports success).
• replay_after_wal_entry_position unchanged; a reopened writer does not re-replay merged entries (test).
• Interrupting between append and drain does not lose or double-count rows under the documented identity contract (test).
• Returns count of reclaimed generations; 0 when nothing pending or no manifest.

Non-goals

• Base-table small-fragment compaction (compact_files) — that's separate and already exists.
• Deciding when / by whom compaction runs (scheduling stays with the caller).
• Cross-shard compaction — this is per-shard by design.

Downstream follow-up (not part of this issue)

lance-context deletes its merge_own_shard (~120 lines) and calls dataset.compact_mem_wal_shard(shard); its merge_own_shard_if_ready / cleanup_own_shard become thin wrappers over the threshold variant.