Skip to content

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

Merged
jackye1995 merged 4 commits into
lance-format:mainfrom
touch-of-grey:jack/mem-wal-primitives
May 3, 2026
Merged

feat: add write ahead log appender and tailer primitives#6669
jackye1995 merged 4 commits into
lance-format:mainfrom
touch-of-grey:jack/mem-wal-primitives

Conversation

@touch-of-grey

@touch-of-grey touch-of-grey commented May 2, 2026

Copy link
Copy Markdown
Contributor

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.

@claude claude Bot left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Claude Code Review

This pull request is from a fork — automated review is disabled. A repository maintainer can comment @claude review to run a one-time review.

@github-actions github-actions Bot added the enhancement New feature or request label May 2, 2026
@touch-of-grey
touch-of-grey force-pushed the jack/mem-wal-primitives branch from 0a9b43f to 708866d Compare May 2, 2026 23:47
@touch-of-grey touch-of-grey changed the title feat: add memory write ahead log appender and tailer primitives feat: add write ahead log appender and tailer primitives May 2, 2026
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.
@touch-of-grey
touch-of-grey force-pushed the jack/mem-wal-primitives branch from 708866d to 82e07f2 Compare May 2, 2026 23:56
@jackye1995

Copy link
Copy Markdown
Contributor

@claude review

Comment thread docs/src/format/table/mem_wal.md Outdated
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.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 jackye1995 left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agreed, will do as a follow-up PR. Filed as a TODO.

Comment on lines +838 to +850
"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!(

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔴 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.

  1. Tailer: read_entry(P) succeeds where P > X, spawns tokio::spawn(best_effort_cursor_update(&ms, P)).
  2. Tailer task runs manifest_store.read_latest() -> returns manifest at version N.
  3. Concurrently, WalAppender::open -> claim_epoch runs read_latest() -> also returns manifest at version N.
  4. Tailer task prepares updated with version = N+1, same writer_epoch = E, new wal_entry_position_last_seen = P.
  5. claim_epoch prepares new manifest with version = N+1, writer_epoch = E+1.
  6. Both target manifest_filename(N+1) via PUT-IF-NOT-EXISTS in ShardManifestStore::write.
  7. Tailer wins (line 873-879 of wal.rs: let _ = manifest_store.write(&updated).await; succeeds).
  8. claim_epoch's self.write(&new_manifest).await returns AlreadyExists -> wrapped at manifest.rs:447-451 as "Failed to claim shard X (version N+1): another writer may have claimed it".
  9. claim_epoch does 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.
  10. WalAppender::open returns 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_update writes 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_hint enables it). Each successful read_entry spawns one fire-and-forget tokio::spawn task, so even a single tailer reading N entries produces up to N concurrent updates contending against any claim_epoch window.
  • Both code paths are new in this PR (WalAppender::open and WalTailer::with_cursor_updates), so this is a regression introduced by this PR's interaction with the pre-existing claim_epoch policy.

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:

  1. Have claim_epoch retry on AlreadyExists when the conflicting manifest's writer_epoch is unchanged (i.e. only a cursor update bumped the version), preserving the no-epoch-war guarantee for real claimants.
  2. Have best_effort_cursor_update re-read after preparing and skip the write if a newer version landed in the meantime.
  3. Store the cursor hint in a separate (non-versioned) namespace so it never contends with claim writes.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Comment thread docs/src/format/table/mem_wal.md Outdated
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.

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 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, 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.

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_entriesshard_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_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).

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

codecov Bot commented May 3, 2026

Copy link
Copy Markdown

@jackye1995
jackye1995 merged commit db02fc9 into lance-format:main May 3, 2026
29 checks passed
touch-of-grey added a commit to touch-of-grey/lance that referenced this pull request May 4, 2026
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]>
jackye1995 pushed a commit that referenced this pull request May 5, 2026
)

## 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]>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

enhancement New feature or request

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants