From 6a7b1e575df4996ca8ab4e7ef467188c57c485dd Mon Sep 17 00:00:00 2001 From: Beinan Date: Sat, 25 Jul 2026 07:29:53 +0000 Subject: [PATCH] fix(rollout): surface unflushed rows in observe(), which undercounts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `RolloutObservation::row_count` sums the base table and flushed MemWAL generations. Rows sitting in an unsealed memtable are in neither, so since the seal moved off the append path the count is permanently short by roughly one flush interval's worth of writes. Before that change every append was sealed before returning and the count was exact. That undercount is inherent to asynchronous flush and not a bug on its own — the problem is that it was invisible, and it feeds the control-plane stats scanner, capacity decisions, and the row counts shown in the UI. Add `unflushed_rows`, read from the resident writer's memtable stats, so the gap is observable: `row_count + unflushed_rows` is every row this instance has durably accepted, and a persistently non-zero value means the flush sweeper is not keeping up. Best-effort and non-failing — no resident writer, a WAL-only writer, or a fenced writer all report 0, so an observability read never fails on writer state. Deliberately not plumbed into the master's stats table: it reads this process's in-memory writer and cannot see another instance's memtable, so aggregating it across instances would be misleading. Also documents that `add`'s return value now carries no information about the append at all — the base version does not advance, and since the seal moved it no longer signals visibility either. Changing the signature would break `AddRolloutsResponse.version`, so that is left for a version bump. Closes #187 Co-Authored-By: Claude --- .../lance-context-core/src/rollout_store.rs | 103 +++++++++++++++++- 1 file changed, 100 insertions(+), 3 deletions(-) diff --git a/crates/lance-context-core/src/rollout_store.rs b/crates/lance-context-core/src/rollout_store.rs index 4931921..c164310 100644 --- a/crates/lance-context-core/src/rollout_store.rs +++ b/crates/lance-context-core/src/rollout_store.rs @@ -117,6 +117,13 @@ const ROLLOUT_ID_INDEX_NAME: &str = "id_idx"; #[derive(Debug, Clone, PartialEq, Eq)] pub struct RolloutObservation { /// Logical row count across the base table and every flushed MemWAL shard. + /// + /// **Excludes rows that are durable but not yet flushed.** An `add` returns + /// once the WAL entry is persisted; the row only reaches a flushed + /// generation when the memtable is sealed. In steady state this count + /// therefore lags recent writes by up to the flush interval. Use + /// [`Self::unflushed_rows`] to see the gap, or `row_count + + /// unflushed_rows` for every row this instance has durably accepted. pub row_count: i64, /// Number of fragments in the base table. pub fragment_count: i64, @@ -126,6 +133,18 @@ pub struct RolloutObservation { pub last_updated: i64, /// Flushed MemWAL generations pending merge across all shards. pub pending_wal_generations: i64, + /// Rows durably accepted by *this instance's* resident writer but not yet + /// sealed into a flushed generation, and so not yet counted by + /// [`Self::row_count`] or visible to any reader. + /// + /// Instance-local by nature: it reads this process's in-memory writer, so + /// it cannot see another instance's unflushed memtable. `0` when no writer + /// is resident (nothing buffered) — which is also the steady state right + /// after a flush. + /// + /// A persistently non-zero value means the flush sweeper is not keeping up + /// (or is disabled); that is the signal that reads are lagging writes. + pub unflushed_rows: i64, } /// Exact-match filters for rollout record browsing. @@ -550,9 +569,15 @@ impl RolloutStore { /// [`RolloutObservation::row_count`] does not count rows that are durable /// but not yet flushed. /// - /// The returned value is the base dataset version, which MemWAL appends do - /// **not** advance; it is retained for API compatibility, not as a per-append - /// snapshot handle (see the module docs on reproducibility). + /// # The return value carries no information about this append + /// + /// It is the base dataset version, which MemWAL appends do **not** advance, + /// so it is a constant unrelated to the rows just written — the same value + /// before and after. It does not identify a snapshot containing them, and + /// (since the seal moved off this path) it does not indicate they are + /// visible either. Do not treat it as a write handle or use it to poll for + /// visibility; call [`Self::flush`] instead. Retained only for API + /// compatibility — see the module docs on reproducibility. pub async fn add(&self, records: &[RolloutRecord]) -> LanceResult { if records.is_empty() { return Ok(self.dataset.manifest.version); @@ -1120,9 +1145,32 @@ impl RolloutStore { version, last_updated, pending_wal_generations, + unflushed_rows: self.unflushed_rows().await, }) } + /// Rows buffered in this instance's resident writer that have not yet been + /// sealed into a flushed generation. + /// + /// Best-effort and non-failing: no resident writer (nothing buffered), a + /// WAL-only writer with no memtable, or a fenced writer all report `0` + /// rather than erroring, so an observability read never fails because of + /// writer state. See [`RolloutObservation::unflushed_rows`]. + async fn unflushed_rows(&self) -> i64 { + let writer = { + let guard = self.write_writer.lock().await; + guard.as_ref().cloned() + }; + let Some(writer) = writer else { + return 0; + }; + writer + .memtable_stats() + .await + .map(|stats| stats.row_count as i64) + .unwrap_or(0) + } + /// Current compaction statistics for the base table. /// /// `is_compacting` is always `false`: compaction here runs synchronously @@ -3214,6 +3262,55 @@ mod tests { }); } + #[test] + fn observe_reports_unflushed_rows_excluded_from_row_count() { + // `row_count` counts base ∪ flushed generations, so durable-but-unsealed + // rows are invisible to it. That undercount is inherent to the async + // flush design; `unflushed_rows` is what makes it observable, and + // `row_count + unflushed_rows` is every row durably accepted. + let dir = TempDir::new().unwrap(); + let uri = dir.path().to_string_lossy().to_string(); + + let runtime = tokio::runtime::Runtime::new().unwrap(); + runtime.block_on(async { + let store = RolloutStore::open_with_options( + &uri, + RolloutStoreOptions { + storage_options: None, + session: None, + shard_id: Some("observe-0".to_string()), + merge_after_generations: None, + }, + ) + .await + .unwrap(); + + // No writer resident yet: nothing buffered. + assert_eq!(store.observe().await.unwrap().unflushed_rows, 0); + + store + .add(&[assistant_record("o-0"), assistant_record("o-1")]) + .await + .unwrap(); + + let obs = store.observe().await.unwrap(); + assert_eq!(obs.row_count, 0, "row_count must not count unsealed rows"); + assert_eq!( + obs.unflushed_rows, 2, + "the durable-but-invisible rows must be observable" + ); + + store.flush().await.unwrap(); + + let obs = store.observe().await.unwrap(); + assert_eq!(obs.row_count, 2, "flushed rows join row_count"); + assert_eq!( + obs.unflushed_rows, 0, + "nothing remains buffered after a flush" + ); + }); + } + #[test] fn distinct_shards_share_one_dataset() { // Two instances writing distinct shards of the same dataset both