feat: add write ahead log appender and tailer primitives#6669
Conversation
0a9b43f to
708866d
Compare
Surface a generic read/write surface for MemWAL shards so callers can drive a shard directly without going through the existing flusher. Shard manifests now record their (shard_spec_id, field_id -> bytes) assignment so a shard found on disk can be mapped back to a shard spec without reading the inline index. ShardManifestStore::initialize_shard writes manifest v1 at epoch 0 idempotently. WalAppender claims a writer epoch via the manifest store and appends Arrow IPC entries with put-if-not-exists, retrying on conflict and fencing on PUT failure. WalTailer reads entries (returning the writer epoch alongside the batches so callers can fence-check) and uses wal_entry_position_last_seen as a probe hint with a listing fallback. Dataset gains list_mem_wal_latest_shard_ids (object-storage directory listing) and mem_wal_index_details accessors. The inline shard snapshot is now treated as a stale read-optimization rather than the source of truth; the spec doc is updated accordingly.
708866d to
82e07f2
Compare
|
@claude review |
| 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`, `shard_field_values`, and `shard_field_string_values` record how this shard maps to its shard spec. `shard_field_values` stores int32-valued shard fields and `shard_field_string_values` stores UTF-8 shard fields. |
There was a problem hiding this comment.
These descriptions are not matching the implementation which is using arrow standard to serialize and deserialize value of any type. Also a few edits in later parts of the spec are not right in the same way
There was a problem hiding this comment.
Fixed in e5e74e5. Reworded the bullet to describe the single shard_field_values map (raw Arrow scalar bytes, interpreted via ShardField.result_type), and fixed the snapshot Arrow schema table to show shard_field_{field_id} as a single column typed per the matching result_type. Also fixed the proto comment on MemWalIndexDetails.
jackye1995
left a comment
There was a problem hiding this comment.
Mostly looks good to me, some errors in the spec changes that should be corrected.
| /// Writes Arrow IPC entries atomically using put-if-not-exists. On conflict, | ||
| /// retries with the next position. Checks fencing on conflict or PUT failure. | ||
| #[derive(Debug)] | ||
| pub struct WalAppender { |
There was a problem hiding this comment.
Ideally we merge this with shard writer so that we don't have 2 ways of appending to the wal. Maybe the writer can initialize an appender. But can be a separated PR
There was a problem hiding this comment.
Agreed, will do as a follow-up PR. Filed as a TODO.
| "cannot append an empty batch list to WAL", | ||
| )); | ||
| } | ||
| let schema = batches[0].schema(); | ||
| for (idx, batch) in batches.iter().enumerate() { | ||
| if batch.num_rows() == 0 { | ||
| return Err(Error::invalid_input(format!( | ||
| "cannot append empty batch at index {} to WAL", | ||
| idx | ||
| ))); | ||
| } | ||
| if batch.schema_ref().fields() != schema.fields() { | ||
| return Err(Error::invalid_input(format!( |
There was a problem hiding this comment.
🔴 Tailer cursor updates and claim_epoch race on the same versioned manifest filename via PUT-IF-NOT-EXISTS, and claim_epoch is fail-fast by design. When the tailer (with with_cursor_updates(true)) wins, WalAppender::open returns the misleading error "another writer may have claimed it" even though no writer was involved. Fix by either retrying claim_epoch when the conflicting manifest only bumped the cursor (same writer_epoch), making the cursor update re-check the version before writing, or moving the cursor hint out of the versioned namespace.
Extended reasoning...
What
best_effort_cursor_update (wal.rs:838-850) and ShardManifestStore::claim_epoch (manifest.rs:413-454) both perform read-modify-write against the same shard manifest version namespace using PUT-IF-NOT-EXISTS, but only claim_epoch is fail-fast. When a tailer with with_cursor_updates(true) wins the version race, claim_epoch returns the misleading error "another writer may have claimed it" and WalAppender::open fails despite no writer ever having claimed the shard.
Step-by-step proof
Initial manifest at version N, writer_epoch E, wal_entry_position_last_seen = X.
- Tailer:
read_entry(P)succeeds whereP > X, spawnstokio::spawn(best_effort_cursor_update(&ms, P)). - Tailer task runs
manifest_store.read_latest()-> returns manifest at versionN. - Concurrently, WalAppender::open ->
claim_epochrunsread_latest()-> also returns manifest at versionN. - Tailer task prepares
updatedwithversion = N+1, samewriter_epoch = E, newwal_entry_position_last_seen = P. claim_epochprepares new manifest withversion = N+1,writer_epoch = E+1.- Both target
manifest_filename(N+1)via PUT-IF-NOT-EXISTS inShardManifestStore::write. - Tailer wins (line 873-879 of wal.rs:
let _ = manifest_store.write(&updated).await;succeeds). claim_epoch'sself.write(&new_manifest).awaitreturnsAlreadyExists-> wrapped at manifest.rs:447-451 as"Failed to claim shard X (version N+1): another writer may have claimed it".claim_epochdoes not retry — manifest.rs:400-402 documents this is intentional ("prevents epoch wars"), but that intent assumes only fencing claimants compete for the version slot.WalAppender::openreturns the error to the caller.
Why existing safeguards don't prevent this
- The no-retry rule was designed to prevent epoch wars between competing writers.
best_effort_cursor_updatewrites the same epoch — it is not fencing — but it shares the version namespace with claimants. with_cursor_updates(true)is a public, tested API surface (test_wal_tailer_uses_manifest_cursor_hintenables it). Each successfulread_entryspawns one fire-and-forgettokio::spawntask, so even a single tailer reading N entries produces up to N concurrent updates contending against anyclaim_epochwindow.- Both code paths are new in this PR (
WalAppender::openandWalTailer::with_cursor_updates), so this is a regression introduced by this PR's interaction with the pre-existingclaim_epochpolicy.
Impact
Whenever a tailer with cursor updates is active alongside a new writer, WalAppender::open becomes flaky and surfaces a misleading message that points operators at "another writer" when none exists. Not a corruption bug — callers can retry — but it produces silent, hard-to-diagnose failures in the documented happy-path configuration of these new primitives.
Suggested fix
Any of:
- Have
claim_epochretry onAlreadyExistswhen the conflicting manifest'swriter_epochis unchanged (i.e. only a cursor update bumped the version), preserving the no-epoch-war guarantee for real claimants. - Have
best_effort_cursor_updatere-read after preparing and skip the write if a newer version landed in the meantime. - Store the cursor hint in a separate (non-versioned) namespace so it never contends with claim writes.
There was a problem hiding this comment.
Good point. We should make sure when claim epoch fails, we recheck epoch, id it is not changed (for example in tailer's case) it is still fine and appender can retry claim.
There was a problem hiding this comment.
Done in e5e74e5. claim_epoch now retries on AlreadyExists when the conflicting manifest's writer_epoch is unchanged (or strictly less than our target epoch). Added test_claim_epoch_retries_through_benign_cursor_update covering exactly the tailer scenario.
There was a problem hiding this comment.
Tightened in c2de308: equal-or-lower writer_epoch retries (covers cursor updates, concurrent initialize_shard, and another claimant racing us at the same target epoch); strictly higher errors out as a real fence-ahead. Also added test_concurrent_claim_epoch_both_succeed_with_distinct_epochs to exercise the retry through real concurrency.
There was a problem hiding this comment.
Corrected in eccee0c — I had conflated the fencing rule (post-claim, equal-or-less is fine because we still hold the epoch) with the claim rule (during claim, equal means our target was just claimed by another writer, so we fail). claim_epoch now fails on latest_epoch >= next_epoch and only retries on strictly less. The benign cursor update / initialize_shard cases still retry as expected. Removed the concurrent-claim test that became non-deterministic under the corrected rule.
| 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`, `shard_field_values`, and `shard_field_string_values` record how this shard maps to its shard spec. `shard_field_values` stores int32-valued shard fields and `shard_field_string_values` stores UTF-8 shard fields. |
There was a problem hiding this comment.
🟡 Docs at docs/src/format/table/mem_wal.md line 177 describe the manifest as having both shard_field_values (int32) and shard_field_string_values (UTF-8) fields, but the implementation only defines a single map. The proto (ShardManifest.shard_field_entries) and Rust struct (shard_field_values: HashMap<String, Vec<u8>>) hold raw Arrow scalar bytes for both int32 and utf8 values, disambiguated by result_type from the ShardSpec. Update the docs to describe the single map and how result_type is used to interpret the bytes.
Extended reasoning...
The mismatch. This PR adds a Shard Assignment bullet to the Shard Manifest section in docs/src/format/table/mem_wal.md:
shard_spec_id,shard_field_values, andshard_field_string_valuesrecord how this shard maps to its shard spec.shard_field_valuesstores int32-valued shard fields andshard_field_string_valuesstores UTF-8 shard fields.
But neither shard_field_string_values nor a separate string-valued map exists in the implementation introduced by this same PR.
What the implementation actually contains. The proto in protos/table.proto adds exactly one new field to ShardManifest:
// 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;The Rust ShardManifest in rust/lance-index/src/mem_wal.rs has a single map:
pub shard_field_values: HashMap<String, Vec<u8>>,Each entry holds raw Arrow scalar bytes (int32 LE bytes, raw UTF-8 bytes, etc.), and the receiver looks up the matching ShardField.result_type from the ShardSpec to interpret the bytes. There is no second map.
Proof — grep across the entire repo. shard_field_string_values appears only at docs/src/format/table/mem_wal.md:177. Zero hits in protos/, zero hits in rust/, zero hits in tests. The conversion code in rust/lance-index/src/mem_wal.rs (From<&ShardManifest> for pb::ShardManifest and TryFrom<pb::ShardManifest> for ShardManifest) only maps shard_field_entries ↔ shard_field_values, with no string-typed counterpart. The unit test test_initialize_shard_writes_v1_with_epoch_zero in manifest.rs writes an int32 value into shard_field_values (7i32.to_le_bytes()) and reads it back from the same single map.
Where the confusion likely came from. The PR also rewrites the Shard Snapshot Arrow Schema section, which legitimately splits string vs int columns at the Lance file level using two prefixes:
| shard_field_{field_id} | int32 | one column per int shard field |
| shard_string_field_{field_id} | utf8 | one column per string shard field |
But that is the snapshot Arrow file's column-naming convention, not a ShardManifest protobuf field. The doc bullet conflates the two by inventing a shard_field_string_values field on the manifest that doesn't exist.
Impact. Docs only, no runtime effect. But this section is presented as the authoritative description of the Shard Manifest, and an implementer reading it would either look for a missing protobuf field or implement two maps that do not match the wire format.
Suggested fix. Replace the line at docs/src/format/table/mem_wal.md:177 with something like:
Shard assignment:
shard_spec_idandshard_field_valuesrecord how this shard maps to its shard spec.shard_field_valuesis a map from shard field id to the raw Arrow scalar bytes of the computed value; the matchingShardField.result_typein theShardSpecdetermines how to interpret each entry (e.g., 4 little-endian bytes for int32, raw UTF-8 bytes for utf8).
There was a problem hiding this comment.
Fixed in e5e74e5 along with the matching jackye1995 thread. Replaced the bullet at mem_wal.md:177 with a single shard_field_values description (raw Arrow scalar bytes interpreted via ShardField.result_type), and rewrote the snapshot Arrow schema table so shard_field_{field_id} is one column per shard field typed per the matching result_type — no separate string namespace. Also fixed the proto comment on MemWalIndexDetails.
- claim_epoch now retries on benign version conflicts (cursor updates,
initialize_shard) and only fails when another writer claimed an
equal-or-higher epoch. This fixes a race where a tailer with
with_cursor_updates(true) racing a new appender would surface a
misleading "another writer may have claimed it" error.
- Spec docs and proto comment for the shard manifest and shard snapshot
schema were inaccurate: the manifest carries a single
shard_field_values map keyed by field id with raw Arrow scalar bytes
(interpreted via ShardField.result_type), not separate int and
string maps. The snapshot schema's shard_field_{field_id} columns
are typed per the matching ShardField.result_type, with no separate
string-prefixed namespace.
- Added a regression test for the cursor-update vs claim_epoch race.
Per review, treat equal-or-lower writer_epoch on conflict as a benign race (cursor update, initialize_shard, or another claimant racing us at the same target — all of which are safely resolved by retry). Only a strictly-higher epoch indicates another writer pulled ahead of us. Adds test_concurrent_claim_epoch_both_succeed_with_distinct_epochs to exercise the retry path through real concurrency.
…trictly less Per follow-up review, distinguish the fencing case from the claim case. In the claim path, when post-conflict latest_epoch is equal to or greater than our target next_epoch, the target was already claimed by another writer and our claim has failed. Only strictly-less-than indicates a benign version bump (cursor update or initialize_shard writing epoch 0) that we can retry through. The fencing path (check_fenced) continues to use strict greater-than: an equal stored epoch means we are still the current writer. Removes the concurrent-claim test that became non-deterministic under the corrected rule (it intentionally now produces one success + one "another writer claimed" failure per the design's no-epoch-war policy).
Codecov Report❌ Patch coverage is 📢 Thoughts on this report? Let us know! |
ShardWriter is now the single high-level entry point; a new `ShardWriterConfig::enable_memtable` flag (default true) selects between the existing MemTable + indexes + Lance-file flushing pipeline and 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. WalAppender stays as the lower-level synchronous-atomic primitive and is now also the WAL write engine inside `WalFlusher`, replacing the previous duplicate plain-`put` implementation. As a consequence, MemTable mode also gains atomic put-if-not-exists, conflict retry, and fence-on-write for free; first WAL entry on a fresh shard is now position 0 (spec-aligned) instead of 1. Discussion: lance-format#6669 (comment) Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
) ## 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) <[email protected]>
Summary
Surface a generic read/write surface for MemWAL shards so callers can drive a shard directly without going through the existing flusher.
ShardManifestnow 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 addsShardFieldEntry; manifest carriesshard_field_values: HashMap<String, Vec<u8>>.ShardManifestStore::initialize_shard()writes manifest v1 at writer epoch 0 and treatsAlreadyExistsas success when the existing manifest matches.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_positionusewal_entry_position_last_seenas a probe hint with a listing fallback.WalReadEntryincludes thewriter_epochrecorded in the entry so callers can fence-check on replay.Datasetmethodsmem_wal_index_details()andlist_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.mdis updated to match.Tests
Added unit tests for:
ShardManifestStore::initialize_shardhappy path, idempotent on match, rejects mismatched conflict.WalAppender/WalTailerround-trip includingwriter_epochpropagation; position increment.WalTailer::with_cursor_updates(true)updateswal_entry_position_last_seenasynchronously, andnext_position()still resolves correctly.mem_wal_path()helper.cc @jackye1995 for review.