fix(merge): make demand folding a per-dimension join by beinan · Pull Request #352 · lance-format/lance-context

beinan · 2026-10-08T21:03:50Z

Part of #333. Fixes review findings 2, 3 [P1] and 4, 5 [P2] on #344/#345 (credit: production review). Stacked on #350; merge after it.

Problems (all reproduced)

Case	Before: forward	Before: reverse
old-epoch merge + new-epoch flush	20	60
flush + scan(older sealed, newer merged)	20	10

And at the storage layer: stored sealed=10, merged=8, writer publishes sealed=12, merged=None → merged lost, rebuild reads pending 12 instead of 4.

Root cause: the fold treated an event as a unit (replace or reject whole) and let one dimension's comparison decide the other's fate.

Fix

sealed and merged are independent Mark { epoch, value }, each the max over every event carrying it → the fold is a join-semilattice. Source is irrelevant to ordering (a lower scan value just loses the max). Pending = sealed − merged only at equal epoch; older-epoch merged ⇒ all pending; newer-epoch merged ⇒ none.
publish_demand_event and demand_release_changes share join_events: writer without merged keeps the executor's mark; executor with older sealed still advances merged.
Finding 4: oldest_pending_ms = flush time of the lowest unmerged generation, from a bounded per-shard map (SEALED_TIMES_CAP=64) pruned as merged advances. Newer flushes never reset it.
Finding 5: TableDemand::default() carries v=1; unknown versions → Ignored::UnknownVersion; v=0 reads as 1.
Spec §1.1 and test 1 updated.

Validation

Both review counterexamples; writer-never-forgets-merged (fold + etcd); scan cannot lower; lower epoch ignored; oldest_pending does not reset; beyond-cap counts exact; schema version enforced.
Randomized property test: 200 × 5-event streams (every permutation, ×2 duplicated) + 50 × 200-event streams (reverse, shuffle) across 2 shards, 3 epochs, 3 sources, with/without merged — marks, pending, bytes agree; age agrees below cap.
14 demand tests, resident two-master merge, 8 sweeper tests green against real etcd; clippy -D warnings, fmt, typos clean.

🤖 Generated with Claude Code

Third P1 step (docs/design/scheduler-p1-data-model.md §1.1, §2.2). A successful WAL catch-up now writes `merged_through` for every shard it merged into the same etcd transaction that releases table ownership, so a planner never observes a released table whose demand still counts the generations just merged. - core: `rollout_append::merged_watermarks(uri)` reads the base table's MemWAL index for the per-shard merged generation. - merge: `Coordinator::release_with_demand` extends `release` with compare/op pairs from `demand_release_changes`, which only advances a shard's `merged_through`, keeps the writer's sealed/epoch/bytes fields, creates a fresh executor-sourced record for an unknown shard, and emits nothing for a shard already at or past the new watermark. `release` itself is unchanged and delegates with an empty list. - master: `run_as` reads watermarks after a successful Catchup outcome and releases with them. A watermark read failure logs and releases without demand rather than stranding a completed merge. Other maintenance kinds release as before. Metric `master_demand_merged_watermarks_total`. Validation: merge-crate etcd test for advance/keep/skip/fresh-shard and a fold to the expected pending count; the resident two-master merge test now asserts two executor-sourced events per table whose merged watermarks equal the base table's and fold to zero pending. 7 demand tests and 9 master etcd tests pass; clippy -D warnings, fmt, typos clean. Refs lance-format#333. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Review of lance-format#344/lance-format#345 found that `TableDemand::fold` was not order independent and that `publish_demand_event` discarded stored merged watermarks. Both reproduced: old-epoch merge + new-epoch flush forward 20, reverse 60 flush + scan(older sealed, newer merged) forward 20, reverse 10 stored sealed=10 merged=8, writer sealed=12 -> merged lost Root cause: the fold treated an event as a unit that replaces or is rejected as a whole, and let one dimension's comparison decide the other's fate. sealed and merged are now independent (epoch, value) marks, each the max over every event that carried it, so the fold is a join-semilattice: any permutation or duplication yields the same marks, pending counts and bytes. Source no longer affects ordering; a scan simply loses the max if its value is lower. Pending is sealed - merged only when both marks share an epoch; a merged mark from an older epoch says nothing about current generations (all pending), one from a newer epoch means the shard was replaced (none pending). Storage layer: publish_demand_event and demand_release_changes share join_events, so a writer event without merged keeps the executor's mark and an executor event with an older sealed still advances merged. Age clock (finding 4): oldest_pending_ms is the flush time of the lowest unmerged generation from a bounded per-shard map pruned as merged advances; a newer flush adds an entry and never resets it. Versioning (finding 5): TableDemand::default() carries v=1; unknown versions are rejected with Ignored::UnknownVersion; v=0 reads as 1. Tests: both review counterexamples; writer-never-forgets-merged at fold and etcd layers; scan cannot lower anything; lower epoch ignored; oldest_pending does not reset; beyond-cap exactness; schema version enforced; randomized property test (200 x 5-event streams, every permutation; 50 x 200-event streams, reverse and shuffle) over two shards, three epochs, all sources, with and without merged. 14 demand tests, resident two-master merge test and 8 sweeper tests pass against real etcd. Clippy -D warnings, fmt, typos clean. Spec §1.1 and test 1 updated. Refs lance-format#333. Review findings 2, 3, 4, 5 on lance-format#344/lance-format#345. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

beinan · 2026-10-08T21:48:54Z

Superseded by #359. Round-4 review showed this ordered marks by (epoch, value), treating a writer epoch change as a numbering restart; Lance keeps generation numbering across writer claims, so the cross-epoch case was wrong in both orders. #359 orders by generation, carries the age clock in the stored record, and refuses to overwrite a newer-schema record.

P1.4 (docs/design/scheduler-p1-data-model.md §1.3). The scan is the reconciliation source for demand: it covers shards whose writer never ran the server flush path (retired writers, pre-upgrade shards) and corrects any writer event that was dropped. - core: all_shard_watermarks() reads every shard's latest manifest for (shard_id, writer_epoch, max flushed generation, pending count). Unlike wal_shard_snapshots it does not enforce the read cap: the scheduler must see an over-cap shard, not be refused by it. - master: demand_publish::spawn_scan_demand runs after each successful observation, detached and bounded (15 s timeout, 4 slots, drop and count), publishing one scan-sourced event per shard through the same per-dimension join as writers and executors, so a scan can only raise a watermark and never lowers a merged mark it does not know about. Metric master_scan_demand_events_total{result}. The scan's existing enqueue and retirement behaviour is untouched. Validation: etcd test with two writer shards and no server flush path: one scan event per shard, fold gives six pending; after an executor reports merged=2 on one shard, a second scan leaves merged intact and does not rewrite the record. 15 scanner unit tests pass; clippy -D warnings, fmt, typos clean. Refs lance-format#333. Stacked on lance-format#352. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

…ut (#357) Part of #333. Fixes review finding **2 [P1]** (round 4) on #350 (credit: production review). Independent of #352. ## Problems in #350 1. Watermark read after the watchdog exited and before `release`, with **no deadline** → a stalled storage read holds a completed merge's ownership. 2. Release txn compared every shard's demand key by `mod_revision` → **a normal writer flush on any shard fails the whole release**. Reproduced. 3. 129+ shards → etcd default `--max-txn-ops=128` exceeded. ## Fix - `release` is the plain ownership txn again; `release_with_demand` / `demand_release_changes` removed. - After a successful `Catchup` release: `demand_publish::spawn_merged_demand` — detached, 15 s timeout, 4 slots, drop-and-count, **one monotonic put per shard**, bounded CAS retry on a lost race. Loss/lateness cannot misstate demand (per-dimension join; scan reconciles). - Both publish paths share `join_events` + retry, so the writer path no longer silently returns `Ok(false)` after losing a race either. - `master_demand_merged_watermarks_total{result}`. ## Validation - New etcd test: writer `sealed=25` and executor `merged=18` published **concurrently** on one shard → both succeed, stored `(25, Some(18))`; lower merged no-op; unknown shard → fresh epoch-0 record. - Resident two-master merge test polls for the detached publish; both shards fold to zero pending. - 7 demand, 9 maintenance/resident, 9 sweeper tests green against etcd; clippy `-D warnings`, fmt, typos clean. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

…wer schema Supersedes lance-format#352, which review round 4 showed was still wrong: it ordered marks by (epoch, value), treating a writer epoch change as a numbering restart. Lance keeps generation numbering when a new writer claims a shard (ShardManifest { writer_epoch: next, ..base }), so a shard merged to 40 whose new writer flushes to 60 has 20 pending, not 60. Marks now order by (value, epoch); epoch is a fence and tiebreak only. The stored record is the per-dimension join (sealed, merged, bytes each a max; sealed_times a min-per-generation union pruned above merged and to SEALED_TIMES_CAP) and TableDemand::fold applies the same join, so a rebuild from etcd equals the in-memory fold - including the age clock, which the stored record now carries (round-4 finding 4: in-memory 10 vs rebuilt 20 before). merged_epoch preserves provenance when a merged mark was reported under a different epoch than the record's writer. Publishers refuse to overwrite a stored record from a newer schema (round-4 finding 5: v=999 was overwritten before). Both publish paths share one read-join-CAS with bounded retry. Tests (15, all pass against real etcd): round-3 and round-4 counterexamples by name; writer never forgets merged; lower value from any source changes nothing; marks order by generation with epoch tiebreak; age clock stable and rebuildable including after merges; beyond-cap exactness; schema enforced on fold and on stored records; stored newer-schema record untouched by both publishers; concurrent writer and executor publishes both survive; randomized property test (200 x 5-event every permutation, 50 x 200-event reverse/shuffle) now also asserts rebuild-from-storage equals the fold. Resident two-master merge and 9 sweeper tests pass. Clippy -D warnings, fmt, typos clean. Spec §1.1 rewritten. Refs lance-format#333. Review findings 1, 4, 5 (round 4); replaces lance-format#352. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

P1.4 (docs/design/scheduler-p1-data-model.md §1.3). The scan is the reconciliation source for demand: it covers shards whose writer never ran the server flush path (retired writers, pre-upgrade shards) and corrects any writer event that was dropped. - core: all_shard_watermarks() reads every shard's latest manifest for (shard_id, writer_epoch, max flushed generation, pending count). Unlike wal_shard_snapshots it does not enforce the read cap: the scheduler must see an over-cap shard, not be refused by it. - master: demand_publish::spawn_scan_demand runs after each successful observation, detached and bounded (15 s timeout, 4 slots, drop and count), publishing one scan-sourced event per shard through the same per-dimension join as writers and executors, so a scan can only raise a watermark and never lowers a merged mark it does not know about. Metric master_scan_demand_events_total{result}. The scan's existing enqueue and retirement behaviour is untouched. Validation: etcd test with two writer shards and no server flush path: one scan event per shard, fold gives six pending; after an executor reports merged=2 on one shard, a second scan leaves merged intact and does not rewrite the record. 15 scanner unit tests pass; clippy -D warnings, fmt, typos clean. Refs lance-format#333. Stacked on lance-format#352. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

…wer schema (#359) Part of #333. **Supersedes #352** (which I will close). Fixes review findings **1 [P1], 4, 5 [P2]** (round 4) (credit: production review). Stacked on #357. ## What #352 got wrong It ordered marks by `(epoch, value)` — treating a writer epoch change as a numbering restart. Lance keeps generation numbering on a new writer claim (`ShardManifest { writer_epoch: next, ..base }`). So "merged to 40, new writer flushes to 60" is **20 pending**; #352 said 60 in both orders. ## Fix - `Mark` orders by `(value, epoch)`; epoch is a fence/tiebreak. `pending = sealed − merged`, plain. - Stored record is the per-dimension join; `fold` applies the same join → **rebuild from etcd equals in-memory fold**, now including the age clock, which the stored record carries (`sealed_times`, bounded to 64 lowest unmerged, min per generation). Finding 4: in-memory 10 vs rebuilt 20 before. - `merged_epoch` keeps provenance when merged was reported under a different epoch. - Publishers **refuse to overwrite a stored record from a newer schema** (finding 5: `v=999` was overwritten). Both publish paths share one read-join-CAS with bounded retry. - Spec §1.1 rewritten. ## Validation (15 tests, all green against real etcd) - Named counterexamples: round-3 (both), round-4 (merged 40 / epoch-2 flush 60 ⇒ 20, in every order, and after rebuild). - Writer never forgets merged; lower value from any source is a no-op; marks order by generation with epoch tiebreak. - Age clock: stable across newer flushes, moves correctly on merge, **identical after rebuild**; beyond-cap still exact for counts. - Schema: enforced on fold; stored `v=999` untouched by both publishers (mod_revision unchanged). - Concurrent writer + executor publishes both succeed and both survive. - **Randomized**: 200 × 5-event streams (every permutation) and 50 × 200-event (reverse, shuffle) over 2 shards / 3 epochs / 3 sources — fold order-independent **and** `rebuild(events) == fold(events)`. - Resident two-master merge and 9 sweeper tests pass; clippy `-D warnings`, fmt, typos clean. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

P1.4 (docs/design/scheduler-p1-data-model.md §1.3). The scan is the reconciliation source for demand: it covers shards whose writer never ran the server flush path (retired writers, pre-upgrade shards) and corrects any writer event that was dropped. - core: all_shard_watermarks() reads every shard's latest manifest for (shard_id, writer_epoch, max flushed generation, pending count). Unlike wal_shard_snapshots it does not enforce the read cap: the scheduler must see an over-cap shard, not be refused by it. - master: demand_publish::spawn_scan_demand runs after each successful observation, detached and bounded (15 s timeout, 4 slots, drop and count), publishing one scan-sourced event per shard through the same per-dimension join as writers and executors, so a scan can only raise a watermark and never lowers a merged mark it does not know about. Metric master_scan_demand_events_total{result}. The scan's existing enqueue and retirement behaviour is untouched. Validation: etcd test with two writer shards and no server flush path: one scan event per shard, fold gives six pending; after an executor reports merged=2 on one shard, a second scan leaves merged intact and does not rewrite the record. 15 scanner unit tests pass; clippy -D warnings, fmt, typos clean. Refs lance-format#333. Stacked on lance-format#352. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

#356) Part of #333 (P1.4). Spec §1.3, source `scan`. **Stacked on #359** (rebased from the superseded #352); merge after #357 and #359. ## What - **core**: `all_shard_watermarks()` — every shard's `(shard_id, writer_epoch, max flushed generation, pending)` from its latest manifest. Does **not** enforce the read cap (the scheduler must see an over-cap shard, not be refused). - **master**: `demand_publish::spawn_scan_demand` after each successful observation. Detached and bounded (15 s, 4 slots, drop-and-count), one `scan`-sourced event per shard via `publish_demand_event`'s per-dimension join, so a scan can only raise and never lowers a `merged` it does not know. Metric `master_scan_demand_events_total{result}`. Existing enqueue/retire behaviour untouched (review point 5 / P1 rule). ## Validation - `scan_publishes_every_shard_and_never_lowers_a_watermark` (etcd): two writer shards, no server flush path → one event per shard, fold = 6 pending; executor reports `merged=2` on a shard; a second scan leaves `merged` intact and does not bump `mod_revision`. - 15 scanner unit tests; clippy `-D warnings`, fmt, typos clean. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

beinan and others added 2 commits October 8, 2026 20:57

beinan closed this Oct 8, 2026

Navigation Menu

Repository navigation

fix(merge): make demand folding a per-dimension join - #352

fix(merge): make demand folding a per-dimension join#352
beinan wants to merge 2 commits into
lance-format:mainfrom
beinan:fix/demand-fold-join-semantics

beinan commented Oct 8, 2026

Uh oh!

beinan commented Oct 8, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

Conversation

beinan commented Oct 8, 2026

Problems (all reproduced)

Fix

Validation

Uh oh!

beinan commented Oct 8, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant