From 79ba076429d60dbdc245278ea3959d97bd5b5aa3 Mon Sep 17 00:00:00 2001 From: Dan Rammer Date: Fri, 7 Aug 2026 13:44:49 -0500 Subject: [PATCH] feat(table): checkpoint_lsm, flush_lsm, compact_lsm, get_lsm_stats (#3736) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Converge a table's LSM write path into its base table, and inspect it. `checkpoint_lsm` is `flush` then `compact`, repeated until the fresh tier is empty — and the loop runs **client-side**. Putting it on the server would mean a background task, which means a single-flight intent, an intent that leaks on panic, a bounded-iteration policy, an "is it done" observable, and a story for every way a client can vanish mid-operation. None of that exists in this shape: each request does a bounded unit of work and reports what is left, so completion is *carried in the responses* rather than inferred from a shared counter that cannot distinguish "converged" from "hasn't started yet". Best-effort by construction. Nothing is frozen, so `converged` means L0 was empty as of the last pass. It is idempotent, abandonable at any point with zero consequence, and safe to run on a cadence — an already-converged table costs one round trip and zero compaction passes, because `flush` reports `generations_remaining` and the loop is never entered. ## The failure taxonomy is the load-bearing part Five distinct conditions used to arrive at a client as one 503. `Error::LsmRoute` carries a classification read from the response body's namespace error code **at the point of receipt** — before any generic helper folds the body into a string and keeps only the status. | condition | wire | client action | |---|---|---| | contention (latch held / pool saturated) | 429, code 21 | retry with backoff | | owning node draining | 503, code 19 `InvalidTableState` | **stop** | | fenced / no slot / transport | 503, code 17 | retry with backoff | | registry entry vanished | 404 | re-issue from `flush` (capped) | | table being dropped / not WAL-backed | 409 / 400 | stop | Draining is terminal because the drain gate is a one-way latch — retrying spins until the deadline to report a failure that was knowable on the first response. Transport retry is disabled on these routes for the same reason: it treats every 503 alike and would burn its budget before the classifier ever saw the body. `get_lsm_stats` returns `Option`, matching `get_lsm_write_spec` — `None` only when the table has no LSM write path, since a struct of zeros would read as measurements. Python bindings mirror all four, preserving per-bucket detail rather than flattening to a table-level summary. ## Testing Six new unit tests against the mocked endpoint, plus the taxonomy round-trip: - flush into an empty L0 issues **zero** compact calls (asserts the call count — `generations_consumed: 0` is also true of a loop that ran a pointless pass) - the loop drives compact until the server reports zero remaining - **contention is not draining**: a 429 retries and converges; asserts the retry count - a draining node stops after **exactly one** request, no retries - stats round-trips fully populated; `include_generation_rows` off by default - every `(status, code)` pair classifies correctly, including unparseable 503 bodies falling back to *retryable* rather than terminal `cargo test -p lancedb --features remote --lib`: 723 passed. ## Notes for review - Depends on the sibling lance change returning `SealedGeneration` from `force_seal_active` only at the *server* level — no lance API is used here. - The branch is based on `codex/update-lance-10-0-0-beta-5`, so it carries one extra commit (`chore: update lance dependency to v10.0.0-beta.5`) that is not part of this change. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: lancedb automation Co-authored-by: Claude Opus 5 (1M context) --- python/python/lancedb/_lancedb.pyi | 4 + python/python/lancedb/table.py | 83 ++++ python/src/table.rs | 108 +++++- rust/lancedb/Cargo.toml | 2 +- rust/lancedb/src/remote/table.rs | 541 +++++++++++++++++++++++++++ rust/lancedb/src/table.rs | 107 ++++++ rust/lancedb/src/table/checkpoint.rs | 315 ++++++++++++++++ rust/lancedb/src/table/lsm_stats.rs | 162 ++++++++ 8 files changed, 1320 insertions(+), 2 deletions(-) create mode 100644 rust/lancedb/src/table/checkpoint.rs create mode 100644 rust/lancedb/src/table/lsm_stats.rs diff --git a/python/python/lancedb/_lancedb.pyi b/python/python/lancedb/_lancedb.pyi index 47e727f99..fad2744d3 100644 --- a/python/python/lancedb/_lancedb.pyi +++ b/python/python/lancedb/_lancedb.pyi @@ -355,6 +355,10 @@ class Table: async def set_lsm_write_spec(self, spec: LsmWriteSpec) -> None: ... async def unset_lsm_write_spec(self) -> None: ... async def get_lsm_write_spec(self) -> Optional[LsmWriteSpec]: ... + async def checkpoint_lsm(self) -> None: ... + async def flush_lsm(self) -> None: ... + async def compact_lsm(self) -> None: ... + async def get_lsm_stats(self, include_generation_rows: bool) -> Optional[dict]: ... async def close_lsm_writers(self) -> None: ... @property def tags(self) -> Tags: ... diff --git a/python/python/lancedb/table.py b/python/python/lancedb/table.py index 59e2650eb..0828f04dc 100644 --- a/python/python/lancedb/table.py +++ b/python/python/lancedb/table.py @@ -3976,6 +3976,28 @@ class LanceTable(Table): [`AsyncTable.get_lsm_write_spec`][lancedb.AsyncTable.get_lsm_write_spec].""" return LOOP.run(self._table.get_lsm_write_spec()) + def checkpoint_lsm(self) -> None: + """Synchronous version of + [`AsyncTable.checkpoint_lsm`][lancedb.AsyncTable.checkpoint_lsm].""" + return LOOP.run(self._table.checkpoint_lsm()) + + def flush_lsm(self) -> None: + """Synchronous version of + [`AsyncTable.flush_lsm`][lancedb.AsyncTable.flush_lsm].""" + return LOOP.run(self._table.flush_lsm()) + + def compact_lsm(self) -> None: + """Synchronous version of + [`AsyncTable.compact_lsm`][lancedb.AsyncTable.compact_lsm].""" + return LOOP.run(self._table.compact_lsm()) + + def get_lsm_stats(self, *, include_generation_rows: bool = False) -> Optional[dict]: + """Synchronous version of + [`AsyncTable.get_lsm_stats`][lancedb.AsyncTable.get_lsm_stats].""" + return LOOP.run( + self._table.get_lsm_stats(include_generation_rows=include_generation_rows) + ) + def close_lsm_writers(self) -> None: """Close cached MemWAL shard writers. See [`AsyncTable.close_lsm_writers`][lancedb.AsyncTable.close_lsm_writers].""" @@ -4686,6 +4708,67 @@ class AsyncTable: """ return await self._inner.get_lsm_write_spec() + async def checkpoint_lsm(self) -> None: + """Converge this table's LSM write path into its base table. + + One flush, sealing every memtable into L0, then compaction triggers + until every generation that existed at that moment has reached base. + The loop runs client-side, reading progress from ``get_lsm_stats``. + + Best-effort: generations created *while* it runs are deliberately not + waited on, which is what lets it terminate on a table taking writes. + Idempotent and safe on a cadence. + + There is no deadline, and the caller owns that. It returns when the + target generations are gone, raises on a terminal server fault, and + otherwise waits however long the server takes. A slow table and a + stuck one are the same picture from the client: the compactor pool is + shared across every table on the node, so a checkpoint queued behind + unrelated work looks exactly like one that is merging. Wrap this in + ``asyncio.wait_for`` for a wall-clock bound; abandoning it partway + costs nothing. + """ + return await self._inner.checkpoint_lsm() + + async def flush_lsm(self) -> None: + """Seal every bucket's active memtable into L0. + + Does not touch the base table — moving L0 into base is + `compact_lsm`. On a node that has not claimed this table, this claims + it and replays its WAL log first. + """ + return await self._inner.flush_lsm() + + async def compact_lsm(self) -> None: + """Trigger a background L0 to base compaction pass per bucket. + + Returns once the passes are dispatched, not once they finish: watch + ``get_lsm_stats`` for progress, or use ``checkpoint_lsm`` to loop + until the current L0 has reached base. + """ + return await self._inner.compact_lsm() + + async def get_lsm_stats( + self, *, include_generation_rows: bool = False + ) -> Optional[dict]: + """Read live per-bucket LSM state. + + Answers "how far behind is my fresh tier", "which bucket is hot", and + "why is my fresh-tier vector search brute-force". Mutates no table + state, though on a node that has not claimed this table it claims it, + exactly as a read would. + + Returns ``None`` only when the LSM write path is not enabled. + + Parameters + ---------- + include_generation_rows + Report a row count per L0 generation. Off by default: each count + opens an uncached Lance dataset, and ``checkpoint_lsm`` polls this + needing only generation numbers. + """ + return await self._inner.get_lsm_stats(include_generation_rows) + async def close_lsm_writers(self) -> None: """Drain and close any cached MemWAL shard writers for this table. diff --git a/python/src/table.rs b/python/src/table.rs index 5b5d6596a..119388708 100644 --- a/python/src/table.rs +++ b/python/src/table.rs @@ -28,11 +28,72 @@ use pyo3::{ Bound, FromPyObject, Py, PyAny, PyRef, PyResult, Python, exceptions::{PyRuntimeError, PyValueError}, pyclass, pyfunction, pymethods, - types::{IntoPyDict, PyAnyMethods, PyBytes, PyDict, PyDictMethods}, + types::{IntoPyDict, PyAnyMethods, PyBytes, PyDict, PyDictMethods, PyList, PyListMethods}, }; mod scannable; +/// Convert `LsmStats` to a Python dict, preserving the per-bucket list. +/// +/// Deliberately not flattened to a table-level summary: a table is N +/// buckets on one node, and the per-bucket detail is the reason the +/// endpoint exists — flattening hides the single hot bucket someone opened +/// it to find. +fn lsm_stats_to_py(py: Python<'_>, stats: &lancedb::table::LsmStats) -> PyResult> { + let out = PyDict::new(py); + let buckets = PyList::empty(py); + for b in &stats.buckets { + let e = PyDict::new(py); + e.set_item("shard_id", &b.shard_id)?; + e.set_item("status", &b.status)?; + e.set_item("writer_epoch", b.writer_epoch)?; + e.set_item("manifest_version", b.manifest_version)?; + e.set_item("current_generation", b.current_generation)?; + e.set_item( + "replay_after_wal_entry_position", + b.replay_after_wal_entry_position, + )?; + e.set_item( + "wal_entry_position_last_seen", + b.wal_entry_position_last_seen, + )?; + + let generations = PyList::empty(py); + for g in &b.generations { + let ge = PyDict::new(py); + ge.set_item("generation", g.generation)?; + ge.set_item("bytes", g.bytes)?; + ge.set_item("rows", g.rows)?; + generations.append(ge)?; + } + e.set_item("generations", generations)?; + e.set_item("compacting", b.compacting)?; + + e.set_item( + "memtables", + b.memtables + .as_ref() + .map(|ms| { + let l = PyList::empty(py); + for m in ms { + let d = PyDict::new(py); + d.set_item("generation", m.generation)?; + d.set_item("rows", m.rows)?; + d.set_item("bytes", m.bytes)?; + d.set_item("batches", m.batches)?; + d.set_item("indexes", m.indexes.clone())?; + l.append(d)?; + } + PyResult::Ok(l.unbind()) + }) + .transpose()?, + )?; + buckets.append(e)?; + } + out.set_item("buckets", buckets)?; + Ok(out.unbind()) +} + #[derive(FromPyObject)] enum PredicateArg { Expr(PyExpr), @@ -1339,6 +1400,51 @@ impl Table { }) } + /// Converge the table's LSM write path into its base table. + /// + /// Best-effort: with writes flowing, new rows may land after the last + /// pass. Errors if the table stops making progress. + pub fn checkpoint_lsm(self_: PyRef<'_, Self>) -> PyResult> { + let inner = self_.inner_ref()?.clone(); + future_into_py(self_.py(), async move { + inner.checkpoint_lsm().await.infer_error() + }) + } + + /// Seal every bucket's active memtable into L0. + pub fn flush_lsm(self_: PyRef<'_, Self>) -> PyResult> { + let inner = self_.inner_ref()?.clone(); + future_into_py( + self_.py(), + async move { inner.flush_lsm().await.infer_error() }, + ) + } + + /// Trigger a background L0 → base pass per bucket. Returns once the + /// passes are dispatched, not once they finish — watch `get_lsm_stats`. + pub fn compact_lsm(self_: PyRef<'_, Self>) -> PyResult> { + let inner = self_.inner_ref()?.clone(); + future_into_py(self_.py(), async move { + inner.compact_lsm().await.infer_error() + }) + } + + /// Live LSM state, or `None` when the LSM write path is not enabled. + #[pyo3(signature = (include_generation_rows=false))] + pub fn get_lsm_stats( + self_: PyRef<'_, Self>, + include_generation_rows: bool, + ) -> PyResult> { + let inner = self_.inner_ref()?.clone(); + future_into_py(self_.py(), async move { + let stats = inner + .get_lsm_stats(include_generation_rows) + .await + .infer_error()?; + Python::attach(|py| stats.map(|s| lsm_stats_to_py(py, &s)).transpose()) + }) + } + pub fn close_lsm_writers(self_: PyRef<'_, Self>) -> PyResult> { let inner = self_.inner_ref()?.clone(); future_into_py(self_.py(), async move { diff --git a/rust/lancedb/Cargo.toml b/rust/lancedb/Cargo.toml index 96ea9ec95..c2137d0b5 100644 --- a/rust/lancedb/Cargo.toml +++ b/rust/lancedb/Cargo.toml @@ -100,7 +100,7 @@ anyhow = "1" lance-testing = { workspace = true } tempfile = "3.5.0" random_word = { version = "0.4.3", features = ["en"] } -tokio = { version = "1.23", features = ["io-util", "macros", "net", "rt-multi-thread", "sync"] } +tokio = { version = "1.23", features = ["io-util", "macros", "net", "rt-multi-thread", "sync", "test-util"] } uuid = { version = "1.7.0", features = ["v4"] } walkdir = "2" aws-sdk-dynamodb = { version = "1.55.0" } diff --git a/rust/lancedb/src/remote/table.rs b/rust/lancedb/src/remote/table.rs index 388bed0f7..f3e872cbe 100644 --- a/rust/lancedb/src/remote/table.rs +++ b/rust/lancedb/src/remote/table.rs @@ -23,11 +23,13 @@ use crate::table::AddResult; use crate::table::BranchDiff; use crate::table::DeleteResult; use crate::table::DropColumnsResult; +use crate::table::LsmStats; use crate::table::LsmWriteSpec; use crate::table::MergeBranchResult; use crate::table::MergeResult; use crate::table::Tags; use crate::table::UpdateResult; +use crate::table::lsm_stats::GetLsmStatsResponse; use crate::table::merge::MergeFilter; use crate::table::query::create_multi_vector_plan; use crate::table::write_progress::FinishOnDrop; @@ -991,6 +993,18 @@ impl RemoteTable { } } + /// Send an LSM operator request with the transport retry layer **off**. + /// + /// Retry policy on these routes belongs to the checkpoint loop, which + /// reads the status and can tell contention from a lost claim. Leaving the + /// transport layer on would re-ask on its own schedule first, and surface + /// an `Error::Retry` whose status the loop would then have to unwrap. + async fn send_lsm_route(&self, request: RequestBuilder) -> Result<(String, reqwest::Response)> { + let (request_id, response) = self.send(request, false).await?; + let response = self.check_table_response(&request_id, response).await?; + Ok((request_id, response)) + } + /// Build a POST request and attach the read-freshness headers /// (`x-lancedb-min-version`, `x-lancedb-min-timestamp`). fn post_read(&self, uri: &str) -> RequestBuilder { @@ -2468,6 +2482,40 @@ impl BaseTable for RemoteTable { }) } + async fn flush_lsm(&self) -> Result<()> { + let request = self + .client + .post(&format!("/v1/table/{}/flush_lsm/", self.identifier)); + self.send_lsm_route(request).await?; + Ok(()) + } + + async fn compact_lsm(&self) -> Result<()> { + let request = self + .client + .post(&format!("/v1/table/{}/compact_lsm/", self.identifier)); + self.send_lsm_route(request).await?; + Ok(()) + } + + async fn get_lsm_stats(&self, include_generation_rows: bool) -> Result> { + // Read-semantics POST, like `get_lsm_write_spec`. + let request = self + .post_read(&format!("/v1/table/{}/get_lsm_stats/", self.identifier)) + .json(&serde_json::json!({ + "include_generation_rows": include_generation_rows, + })); + let (request_id, response) = self.send_lsm_route(request).await?; + let body = response.text().await.err_to_http(request_id.clone())?; + let parsed: GetLsmStatsResponse = serde_json::from_str(&body).map_err(|e| Error::Http { + source: format!("Failed to parse get_lsm_stats response: {e}").into(), + request_id, + status_code: None, + })?; + // `null` — and only — when the table has no LSM write path. + Ok(parsed.lsm_stats) + } + async fn set_lsm_write_spec(&self, spec: LsmWriteSpec) -> Result<()> { self.check_mutable().await?; @@ -6682,6 +6730,499 @@ mod tests { assert!(table.get_lsm_write_spec().await.unwrap().is_none()); } + /// Build a `get_lsm_stats` body for one bucket holding `generations`. + fn stats_body(generations: &[u64], compacting: bool) -> String { + serde_json::json!({ + "lsm_stats": { + "buckets": [{ + "shard_id": "b0", + "status": "Active", + "writer_epoch": 1, + "manifest_version": 1, + "current_generation": generations.iter().max().copied().unwrap_or(0) + 1, + "replay_after_wal_entry_position": 0, + "wal_entry_position_last_seen": 0, + "generations": generations.iter() + .map(|g| serde_json::json!({ "generation": g, "bytes": 1 })) + .collect::>(), + "compacting": compacting, + "memtables": [], + }], + } + }) + .to_string() + } + + /// `flush_lsm` / `compact_lsm` answer 202 with no body at all. + fn accepted() -> http::Response { + http::Response::builder() + .status(202) + .body(String::new()) + .unwrap() + } + + fn ok_json(body: String) -> http::Response { + http::Response::builder().status(200).body(body).unwrap() + } + + /// A flush landing in an empty L0 finishes on the opening stats read + /// alone. Asserting zero compacts is the point: "it returned Ok" is also + /// true of a loop that ran a pointless pass. + #[tokio::test(start_paused = true)] + async fn test_checkpoint_short_circuits_on_empty_l0() { + let compacts = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let seen = compacts.clone(); + let table = Table::new_with_handler("my_table", move |request| { + let path = request.url().path().to_string(); + if path.contains("compact_lsm") { + seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + panic!("an already-converged table must issue no compact calls"); + } + if path.contains("flush_lsm") { + return accepted(); + } + assert_eq!(path, "/v1/table/my_table/get_lsm_stats/"); + ok_json(stats_body(&[], false)) + }); + + table.checkpoint_lsm().await.unwrap(); + assert_eq!(compacts.load(std::sync::atomic::Ordering::SeqCst), 0); + } + + /// The loop triggers compaction until every generation that existed at + /// the start is gone, one bounded prefix per pass. + #[tokio::test(start_paused = true)] + async fn test_checkpoint_triggers_until_targets_are_drained() { + let compacts = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let seen = compacts.clone(); + let table = Table::new_with_handler("my_table", move |request| { + let path = request.url().path().to_string(); + if path.contains("flush_lsm") { + return accepted(); + } + if path.contains("compact_lsm") { + seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + return accepted(); + } + // Each pass drains the oldest generation. + let drained = seen.load(std::sync::atomic::Ordering::SeqCst); + let left: Vec = [1u64, 2, 3].into_iter().skip(drained).collect(); + ok_json(stats_body(&left, false)) + }); + + table.checkpoint_lsm().await.unwrap(); + assert_eq!( + compacts.load(std::sync::atomic::Ordering::SeqCst), + 3, + "one trigger per generation prefix, then stop" + ); + } + + /// Generations created *during* the checkpoint are not waited on, which + /// is what lets the loop terminate on a table taking writes where "L0 is + /// empty" never becomes true. + #[tokio::test(start_paused = true)] + async fn test_checkpoint_ignores_generations_created_while_it_runs() { + let compacts = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let seen = compacts.clone(); + let table = Table::new_with_handler("my_table", move |request| { + let path = request.url().path().to_string(); + if path.contains("flush_lsm") { + return accepted(); + } + if path.contains("compact_lsm") { + seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + return accepted(); + } + // Target is 5. One pass drains it; a writer keeps adding above. + let n = seen.load(std::sync::atomic::Ordering::SeqCst); + let body = if n == 0 { + stats_body(&[5], false) + } else { + stats_body(&[6, 7], false) + }; + ok_json(body) + }); + + table.checkpoint_lsm().await.unwrap(); + assert_eq!( + compacts.load(std::sync::atomic::Ordering::SeqCst), + 1, + "the loop must not chase generations written after it started" + ); + } + + /// Contention is a 429 and must be retried. The server keeps it off 503 + /// precisely so the client can act on the status alone — reading it as + /// terminal stops the checkpoint early on a healthy node. + #[tokio::test(start_paused = true)] + async fn test_checkpoint_retries_contention() { + let compacts = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let seen = compacts.clone(); + let table = Table::new_with_handler("my_table", move |request| { + let path = request.url().path().to_string(); + if path.contains("flush_lsm") { + return accepted(); + } + if path.contains("compact_lsm") { + // First two triggers: every bucket already latched. + if seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst) < 2 { + return http::Response::builder() + .status(429) + .body(r#"{"code":21,"error":"Too many concurrent writes"}"#.to_string()) + .unwrap(); + } + return accepted(); + } + let accepted_triggers = seen + .load(std::sync::atomic::Ordering::SeqCst) + .saturating_sub(2); + let left: Vec = if accepted_triggers == 0 { + vec![1] + } else { + vec![] + }; + ok_json(stats_body(&left, false)) + }); + + table + .checkpoint_lsm() + .await + .expect("contention must not abort the checkpoint"); + assert_eq!( + compacts.load(std::sync::atomic::Ordering::SeqCst), + 3, + "assert the retry count, not just the outcome" + ); + } + + /// A transient fault on the poll must not abort the checkpoint. This route + /// meets the most contention — it runs every `POLL_INTERVAL` for the + /// checkpoint's whole life, with the transport retry layer disabled — yet + /// was the one call reached with a bare `?`. + #[tokio::test(start_paused = true)] + async fn test_checkpoint_retries_a_contended_stats_poll() { + let polls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let seen = polls.clone(); + let table = Table::new_with_handler("my_table", move |request| { + let path = request.url().path().to_string(); + if path.contains("flush_lsm") || path.contains("compact_lsm") { + return accepted(); + } + // The opening read lands; the next two polls are latched out. + let n = seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + if (1..3).contains(&n) { + return http::Response::builder() + .status(429) + .body(r#"{"code":21,"error":"Too many concurrent writes"}"#.to_string()) + .unwrap(); + } + ok_json(stats_body(if n < 4 { &[1] } else { &[] }, false)) + }); + + table + .checkpoint_lsm() + .await + .expect("a contended poll must be retried, not surfaced"); + assert_eq!( + polls.load(std::sync::atomic::Ordering::SeqCst), + 5, + "the two rejected polls must be re-issued, not skipped" + ); + } + + /// Contention and a lost claim draw on separate budgets: five straight + /// 429s on `flush`, more than `MAX_REISSUES`, must still converge. On one + /// shared counter this spent the re-issue cap and then reported a lost + /// claim nothing had ever reported. + #[tokio::test(start_paused = true)] + async fn test_contention_does_not_exhaust_the_reissue_budget() { + let flushes = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let seen = flushes.clone(); + let table = Table::new_with_handler("my_table", move |request| { + let path = request.url().path().to_string(); + if path.contains("flush_lsm") { + if seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst) < 5 { + return http::Response::builder() + .status(429) + .body(r#"{"code":21,"error":"Too many concurrent writes"}"#.to_string()) + .unwrap(); + } + return accepted(); + } + if path.contains("compact_lsm") { + return accepted(); + } + ok_json(stats_body(&[], false)) + }); + + table + .checkpoint_lsm() + .await + .expect("contention must not be reported as a lost claim"); + assert_eq!( + flushes.load(std::sync::atomic::Ordering::SeqCst), + 6, + "five retries against one seal, then it lands" + ); + } + + /// An exhausted retry budget surfaces the fault that consumed it, not a + /// message the loop invented: "429, nine times" points an operator at a + /// saturated pool, a generic runtime error points them nowhere. + #[tokio::test(start_paused = true)] + async fn test_exhausted_retries_surface_the_underlying_fault() { + let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let seen = calls.clone(); + let table = Table::new_with_handler("my_table", move |_request| { + seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + http::Response::builder() + .status(429) + .body(r#"{"code":21,"error":"Too many concurrent writes"}"#.to_string()) + .unwrap() + }); + + let err = table.checkpoint_lsm().await.unwrap_err(); + assert!( + matches!(&err, Error::Http { status_code: Some(s), .. } if s.as_u16() == 429), + "the fault that spent the budget must be the one reported: {err:?}" + ); + assert_eq!( + calls.load(std::sync::atomic::Ordering::SeqCst), + 9, + "one call plus MAX_RETRIES — the re-issue budget is not spent on top" + ); + } + + /// A draining node is terminal, but the client does not know that from the + /// status: draining and a proxy blip are both 503, and telling them apart + /// takes parsing the body for a namespace code. So it spends the retry + /// budget and then reports what the server said — the drain gate never + /// releases, so the answer does not change, and the operator still reads + /// "WAL node draining" in the error. + #[tokio::test(start_paused = true)] + async fn test_draining_surfaces_after_the_retry_budget() { + let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let seen = calls.clone(); + let table = Table::new_with_handler("my_table", move |_request| { + seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + http::Response::builder() + .status(503) + .body(r#"{"code":19,"error":"WAL node draining"}"#.to_string()) + .unwrap() + }); + + let err = table.checkpoint_lsm().await.unwrap_err(); + let message = err.to_string(); + assert!( + matches!(&err, Error::Http { status_code: Some(s), .. } if s.as_u16() == 503), + "the 503 must surface as itself: {err:?}" + ); + assert!( + message.contains("WAL node draining"), + "the server's own diagnosis must survive to the caller: {message}" + ); + assert_eq!( + calls.load(std::sync::atomic::Ordering::SeqCst), + 9, + "one call plus MAX_RETRIES, then it reports rather than spinning" + ); + } + + /// A long stall with nothing compacting must keep waiting, not fail. The + /// client cannot judge this: a checkpoint queued behind unrelated tables + /// on the pod-wide compactor pool reports exactly these numbers — flat + /// generations, an idle latch — as one whose merges are failing. The + /// deadline is the caller's. + #[tokio::test(start_paused = true)] + async fn test_checkpoint_waits_out_a_long_stall_rather_than_failing() { + let polls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let seen = polls.clone(); + let table = Table::new_with_handler("my_table", move |request| { + let path = request.url().path().to_string(); + if path.contains("flush_lsm") || path.contains("compact_lsm") { + return accepted(); + } + // Flat for far longer than any bound this loop ever had, with + // `compacting: false` throughout — then it drains. + let n = seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + ok_json(stats_body(if n < 40 { &[1, 2] } else { &[] }, false)) + }); + + table + .checkpoint_lsm() + .await + .expect("a stall is the server being slow, not the client's call to make"); + assert!( + polls.load(std::sync::atomic::Ordering::SeqCst) > 40, + "the loop must have kept polling well past the old ten-poll bound" + ); + } + + /// A pass already owns the latch on every outstanding bucket, so the loop + /// waits rather than piling on triggers it would only refuse. This is the + /// sole thing `compacting` is read for. + #[tokio::test(start_paused = true)] + async fn test_checkpoint_waits_while_a_pass_is_running() { + let polls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let compacts = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let seen_polls = polls.clone(); + let seen_compacts = compacts.clone(); + let table = Table::new_with_handler("my_table", move |request| { + let path = request.url().path().to_string(); + if path.contains("flush_lsm") { + return accepted(); + } + if path.contains("compact_lsm") { + seen_compacts.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + return accepted(); + } + // Latched for many polls, then done. + let n = seen_polls.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + ok_json(if n > 15 { + stats_body(&[], false) + } else { + stats_body(&[1], true) + }) + }); + + table + .checkpoint_lsm() + .await + .expect("a running pass is progress, not a stall"); + assert_eq!( + compacts.load(std::sync::atomic::Ordering::SeqCst), + 0, + "never trigger against a bucket already compacting" + ); + } + + /// WAL off ⇒ `None`; WAL on ⇒ a fully populated `Some` with no field + /// defaulting to a zero it did not measure. `include_generation_rows` + /// rides in the body and is off unless asked for. + #[tokio::test] + async fn test_get_lsm_stats_round_trip() { + let table = Table::new_with_handler("my_table", |request| { + assert_eq!(request.url().path(), "/v1/table/my_table/get_lsm_stats/"); + let body = request.body().unwrap().as_bytes().unwrap(); + let body: serde_json::Value = serde_json::from_slice(body).unwrap(); + assert_eq!( + body["include_generation_rows"], true, + "the flag must reach the server, not be silently dropped" + ); + let response = serde_json::json!({ + "lsm_stats": { + "buckets": [{ + "shard_id": "b0", + "status": "Active", + "writer_epoch": 3, + "manifest_version": 11, + "current_generation": 9, + "replay_after_wal_entry_position": 100, + "wal_entry_position_last_seen": 140, + "generations": [{ "generation": 8, "bytes": 4096, "rows": 30 }], + "compacting": false, + "memtables": [ + { "generation": 9, "rows": 12, "bytes": 900, "batches": 2, + "indexes": ["vec_idx"] } + ], + }], + } + }); + http::Response::builder() + .status(200) + .body(response.to_string()) + .unwrap() + }); + + let stats = table + .get_lsm_stats(true) + .await + .unwrap() + .expect("a WAL-backed table reports Some"); + let bucket = &stats.buckets[0]; + assert_eq!(bucket.replay_after_wal_entry_position, 100); + assert_eq!(bucket.wal_entry_position_last_seen, 140); + assert!(!bucket.compacting); + assert_eq!(bucket.generations[0].generation, 8); + assert_eq!(bucket.generations[0].rows, Some(30)); + // The line that answers "why is my fresh-tier vector search + // brute-force" — an absent index name is the whole explanation. + let memtables = bucket.memtables.as_ref().unwrap(); + assert_eq!(memtables[0].indexes, vec!["vec_idx".to_string()]); + } + + /// A 404 arrives as `TableNotFound`, not as a lost claim the loop + /// re-issues from flush until its cap. The two are distinguished by + /// status: 404 is "no such table", 421 is "this node holds no claim". + /// They shared 404 once, and the loop chased a name that never existed. + #[tokio::test(start_paused = true)] + async fn test_missing_table_is_not_read_as_a_lost_claim() { + let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let seen = calls.clone(); + let table = Table::new_with_handler("my_table", move |_request| { + seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + http::Response::builder() + .status(404) + .body(r#"{"code":4,"error":"Not found: Table not found: my_table"}"#.to_string()) + .unwrap() + }); + + let err = table.checkpoint_lsm().await.unwrap_err(); + assert!( + matches!(err, Error::TableNotFound { .. }), + "a missing table must say so: {err:?}" + ); + assert_eq!( + calls.load(std::sync::atomic::Ordering::SeqCst), + 1, + "no point re-claiming a table that does not exist" + ); + } + + /// A lost claim — 421, not 404 — does re-issue from flush, the call that + /// re-claims and replays. + #[tokio::test(start_paused = true)] + async fn test_registry_miss_reissues_from_flush() { + let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let seen = calls.clone(); + let table = Table::new_with_handler("my_table", move |request| { + let path = request.url().path().to_string(); + let n = seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + if path.contains("flush_lsm") { + // First flush lands; the claim is then lost, and the + // re-issued flush succeeds. + return accepted(); + } + if path.contains("compact_lsm") { + if n < 4 { + return http::Response::builder() + .status(421) + .body(r#"{"code":19,"error":"table not claimed"}"#.to_string()) + .unwrap(); + } + return accepted(); + } + ok_json(stats_body(if n < 6 { &[1] } else { &[] }, false)) + }); + + table + .checkpoint_lsm() + .await + .expect("a lost claim must be recovered by re-flushing, not surfaced"); + } + + #[tokio::test] + async fn test_get_lsm_stats_absent_when_wal_off() { + let table = Table::new_with_handler("my_table", |_request| { + http::Response::builder() + .status(200) + .body(serde_json::json!({ "lsm_stats": null }).to_string()) + .unwrap() + }); + assert!(table.get_lsm_stats(false).await.unwrap().is_none()); + } + #[tokio::test] async fn test_wait_for_index() { let table = _make_table_with_indices(0); diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs index 0d8a8e8b9..74ab17921 100644 --- a/rust/lancedb/src/table.rs +++ b/rust/lancedb/src/table.rs @@ -68,10 +68,12 @@ use self::merge::MergeInsertBuilder; pub mod add_columns; mod add_data; pub mod branch_merge; +pub mod checkpoint; mod create_index; pub mod datafusion; pub(crate) mod dataset; pub mod delete; +pub mod lsm_stats; pub mod merge; pub mod optimize; mod primary_key; @@ -95,6 +97,7 @@ pub use lance::dataset::refs::{BranchContents, Ref, TagContents, Tags as LanceTa pub use lance::dataset::scanner::DatasetRecordBatchStream; use lance::dataset::statistics::DatasetStatisticsExt; pub use lance_index::optimize::OptimizeOptions; +pub use lsm_stats::{BucketStats, GenerationStats, LsmStats, MemtableStats}; pub use optimize::{CompactionOptions, OptimizeAction, OptimizeStats}; pub use schema_evolution::{ AddColumnsResult, AlterColumnsResult, DropColumnsResult, FieldMetadataUpdate, @@ -685,6 +688,31 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync { message: "get_lsm_write_spec is not supported on this table type".into(), }) } + /// Seal every bucket's active memtable into L0. + /// + /// The default implementation returns `NotSupported`. + async fn flush_lsm(&self) -> Result<()> { + Err(Error::NotSupported { + message: "flush_lsm is not supported on this table type".into(), + }) + } + /// Trigger a background L0 → base compaction pass per bucket. + /// + /// The default implementation returns `NotSupported`. + async fn compact_lsm(&self) -> Result<()> { + Err(Error::NotSupported { + message: "compact_lsm is not supported on this table type".into(), + }) + } + /// Read live LSM state, or `None` when the LSM write path is not + /// enabled for this table. + /// + /// The default implementation returns `NotSupported`. + async fn get_lsm_stats(&self, _include_generation_rows: bool) -> Result> { + Err(Error::NotSupported { + message: "get_lsm_stats is not supported on this table type".into(), + }) + } /// Drain and close any cached MemWAL shard writers for this table. /// /// The default implementation is a no-op; table types that maintain @@ -1726,6 +1754,85 @@ impl Table { self.inner.get_lsm_write_spec().await } + /// Converge this table's LSM write path into its base table. + /// + /// One `flush` to seal every memtable into L0, then compaction triggers + /// until every generation that existed at that moment has reached base. + /// The loop runs client-side, reading progress from `get_lsm_stats`, so + /// there is no held socket and nothing to reconcile if you drop this + /// future partway through. + /// + /// **Best-effort.** Generations created *after* the opening flush are + /// deliberately not waited on — that is what lets this terminate on a + /// table taking writes. Idempotent and safe on a cadence: an + /// already-converged table costs two round trips and triggers nothing. + /// + /// **No deadline, and the caller owns that.** It returns when the target + /// generations are gone, propagates a terminal server fault, and + /// otherwise waits however long the server takes. A slow table and a + /// stuck one are the same picture from here: the compactor pool is shared + /// across every table on the node, so a checkpoint queued behind + /// unrelated work is indistinguishable from one that is merging. Wrap + /// this in `tokio::time::timeout` for a wall-clock bound; abandoning it + /// partway costs nothing. + /// + /// # Example + /// + /// ```no_run + /// # use lancedb::Table; + /// # async fn example(table: &Table) -> Result<(), Box> { + /// let before = table.get_lsm_stats(false).await?; + /// table.checkpoint_lsm().await?; + /// let after = table.get_lsm_stats(false).await?; + /// # Ok(()) + /// # } + /// ``` + pub async fn checkpoint_lsm(&self) -> Result<()> { + checkpoint::checkpoint_lsm(self).await + } + + /// Seal every bucket's active memtable into L0 without touching the + /// base table. + /// + /// Independently useful: flushing makes memtable rows readable from L0 at + /// a lower per-query cost. On a node that has not claimed this table it + /// claims it and replays the WAL log first — reporting "nothing to flush" + /// without replaying would lie about durable data. + pub async fn flush_lsm(&self) -> Result<()> { + self.inner.flush_lsm().await + } + + /// Run one bounded L0 → base compaction pass per bucket, reporting what + /// it merged and what is left. + /// + /// One pass, not convergence: that bounds each request's cost and gives a + /// caller driving its own cadence a progress signal per round trip. + pub async fn compact_lsm(&self) -> Result<()> { + self.inner.compact_lsm().await + } + + /// Read live per-bucket LSM state. + /// + /// Answers "how far behind is my fresh tier", "which bucket is hot", and + /// "why is my fresh-tier vector search brute-force". Mutates no table + /// state, though on a node that has not claimed this table it claims it, + /// exactly as a read would. + /// + /// `include_generation_rows` reports a row count per L0 generation. Off by + /// default: each count opens an uncached Lance dataset, and + /// `checkpoint_lsm` polls this needing only generation numbers. + /// + /// `Ok(None)` only when the LSM write path is not enabled, matching + /// [`Table::get_lsm_write_spec`]. Stats is fresh-tier only, so with the + /// WAL off there is no manifest to report and a struct of zeros would + /// read as measurements. + /// + /// Do not build a checkpoint's termination on this: the completion + /// predicate lives in the `flush` and `compact` responses. + pub async fn get_lsm_stats(&self, include_generation_rows: bool) -> Result> { + self.inner.get_lsm_stats(include_generation_rows).await + } + /// Drain and close any cached MemWAL shard writers held for this table. /// /// When an [`LsmWriteSpec`] is installed, `merge_insert` opens MemWAL shard diff --git a/rust/lancedb/src/table/checkpoint.rs b/rust/lancedb/src/table/checkpoint.rs new file mode 100644 index 000000000..bb76604ed --- /dev/null +++ b/rust/lancedb/src/table/checkpoint.rs @@ -0,0 +1,315 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The LanceDB Authors + +//! Converging a table's LSM write path into its base table. +//! +//! `checkpoint_lsm` seals once, then triggers compaction and watches +//! generation numbers until the L0 that existed at the start is gone. +//! +//! The loop runs in the client, not the server: `compact_lsm` dispatches a +//! pass and returns, so nothing holds a socket and a client can vanish +//! mid-operation with nothing to reconcile. Completion is read from +//! generation numbers in the shard manifest — durable state, unlike a count +//! in a compact response, which a concurrent write invalidates. +//! +//! The target set is fixed at the start, so generations created *during* the +//! checkpoint are ignored. That is what lets it terminate under write load, +//! and what makes it best-effort: it converges the fresh tier as of some +//! instant. Idempotent, abandonable at any point, safe on a cadence. +//! +//! No liveness bound — the caller owns the deadline. The compactor pool is +//! shared pod-wide, so a checkpoint queued behind unrelated tables looks +//! exactly like one that is merging. + +use std::collections::HashMap; +use std::future::Future; +use std::time::Duration; + +use crate::{Error, Result, Table}; + +/// The HTTP status a failed request carried, if it carried one. +/// +/// `None` for anything with no retry story: a `TableNotFound` that +/// `check_table_response` already translated, or a connection failure that +/// never reached the server. Both are terminal. +fn status_of(e: &Error) -> Option { + #[cfg(feature = "remote")] + { + match e { + Error::Http { + status_code: Some(status), + .. + } => Some(status.as_u16()), + _ => None, + } + } + #[cfg(not(feature = "remote"))] + { + let _ = e; + None + } +} + +/// 429 (latch held, pool saturated, or the pod replaying its WAL) and 503 (a +/// draining node, or a proxy between here and it). +/// +/// The status is the whole signal: the server deliberately keeps contention +/// off 503, so a latch collision is a 429. A draining node *is* terminal, but +/// it is also a 503 that stays a 503, so retrying spends one budget and then +/// reports the server's own message — cheaper than parsing the body for the +/// namespace code it would take to tell the two apart. +fn is_retryable(e: &Error) -> bool { + matches!(status_of(e), Some(429 | 503)) +} + +/// 421: the owning node holds no claim. Only `flush` re-claims and replays, +/// so this cannot be retried in place — the caller has to start over. +fn is_lost_claim(e: &Error) -> bool { + status_of(e) == Some(421) +} + +/// Interval between `get_lsm_stats` polls. One interval is roughly one +/// compaction pass, the granularity at which the answer can change. +/// +/// Fixed rather than configurable, matching `wait_for_index`. It costs +/// nothing on an already-converged table and at most one interval of tail +/// latency after the final pass lands. +const POLL_INTERVAL: Duration = Duration::from_secs(5); + +/// Cap on re-issues from `flush` after a 421, so a crash-looping node cannot +/// turn flush → compact → 421 → flush into a spin. +/// +/// Deliberately not shared with [`MAX_RETRIES`]: a claim that keeps +/// evaporating is a broken node, while contention is routine and wants a real +/// budget. One shared counter let a merely contended table exhaust this cap +/// and then blame a claim it never lost. +const MAX_REISSUES: usize = 3; + +/// Retryable faults tolerated on a *single* request, reset on every success — +/// scattered contention across a long checkpoint must not accumulate toward a +/// cap. Roughly 16s of retrying against the backoff below. +const MAX_RETRIES: usize = 8; + +/// Backoff between retries, doubling up to [`RETRY_BACKOFF_MAX`]. Latch +/// contention clears in about the time one pass takes, so start small; a +/// saturated pool wants the ceiling. +const RETRY_BACKOFF_BASE: Duration = Duration::from_millis(100); +const RETRY_BACKOFF_MAX: Duration = Duration::from_secs(5); + +/// Sleep before re-issuing a retryable request. +async fn backoff(attempt: usize) { + let delay = RETRY_BACKOFF_BASE + .saturating_mul(1u32 << attempt.min(8) as u32) + .min(RETRY_BACKOFF_MAX); + tokio::time::sleep(delay).await; +} + +/// Whether the drain loop finished or needs the table re-claimed first. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum CheckpointOutcome { + Done, + ReissueFromFlush, +} + +/// What one LSM request produced: its value, or word that the owning node +/// holds no claim and only `flush` can get it back. +enum Attempt { + Ok(T), + ReissueFromFlush, +} + +/// Issue one LSM request, retrying in place while the fault is retryable. +/// +/// The two recoverable faults have separate budgets: contention clears on its +/// own and retries here against [`MAX_RETRIES`], while a 421 needs `flush` to +/// re-claim, which only the caller can drive. +/// +/// An exhausted budget propagates the last error *as itself* rather than a +/// synthesized one — "429 after nine tries" beats "checkpoint failed", and a +/// draining node arrives carrying the server's own message. +async fn issue(mut call: F) -> Result> +where + F: FnMut() -> Fut, + Fut: Future>, +{ + let mut retries = 0; + loop { + let e = match call().await { + Ok(value) => return Ok(Attempt::Ok(value)), + Err(e) => e, + }; + if is_lost_claim(&e) { + return Ok(Attempt::ReissueFromFlush); + } + if !is_retryable(&e) || retries >= MAX_RETRIES { + return Err(e); + } + backoff(retries).await; + retries += 1; + } +} + +/// Drive [`Table::checkpoint_lsm`]: seal once, fix the target watermark +/// from the resulting L0, then trigger and poll until it drains. +pub(crate) async fn checkpoint_lsm(table: &Table) -> Result<()> { + for reissue in 0..=MAX_REISSUES { + // The seal turns everything written before this call into a + // generation, so the watermark has to be read after it. Idempotent: + // sealing an empty memtable is a no-op, so a re-issue does not churn + // empty generations. + match issue(|| table.flush_lsm()).await? { + Attempt::Ok(()) => {} + Attempt::ReissueFromFlush => { + backoff(reissue).await; + continue; + } + } + + let stats = match issue(|| table.get_lsm_stats(false)).await? { + Attempt::Ok(stats) => stats, + Attempt::ReissueFromFlush => { + backoff(reissue).await; + continue; + } + }; + let Some(stats) = stats else { + // Not WAL-backed; `flush_lsm` would have errored first but for a race. + return Ok(()); + }; + let targets: HashMap = stats + .buckets + .iter() + .filter_map(|b| Some((b.shard_id.clone(), b.newest_generation()?))) + .collect(); + if targets.is_empty() { + return Ok(()); + } + + match drain_to_targets(table, &targets).await? { + CheckpointOutcome::Done => return Ok(()), + CheckpointOutcome::ReissueFromFlush => { + backoff(reissue).await; + continue; + } + } + } + Err(Error::Runtime { + message: "checkpoint_lsm: the owning node kept losing its claim; \ + re-issued from flush the maximum number of times" + .into(), + }) +} + +/// Trigger and poll until no bucket holds a generation at or below its +/// target. +/// +/// No liveness bound, deliberately. The pod-wide compactor pool (a semaphore +/// of 2 by default, shared across every table on the node) is taken *inside* +/// the pass, after the bucket latch, so a checkpoint queued behind unrelated +/// tables is indistinguishable from one that is merging. An idle-poll counter +/// here could only ever have fired on a table that would have finished. +async fn drain_to_targets( + table: &Table, + targets: &HashMap, +) -> Result { + loop { + let stats = match issue(|| table.get_lsm_stats(false)).await? { + Attempt::Ok(stats) => stats, + Attempt::ReissueFromFlush => return Ok(CheckpointOutcome::ReissueFromFlush), + }; + let Some(stats) = stats else { + return Ok(CheckpointOutcome::Done); + }; + // `compacting` is the bucket's compaction latch, held from dispatch + // until the pass ends — including while it waits on the pod-wide + // permit. So it answers one question only: do not pile on. Buckets + // with nothing outstanding are skipped, not counted as idle. + let mut outstanding = 0; + let mut all_compacting = true; + for b in &stats.buckets { + let Some(target) = targets.get(&b.shard_id) else { + continue; + }; + let n = b.outstanding_generations(*target); + if n > 0 { + outstanding += n; + all_compacting &= b.compacting; + } + } + if outstanding == 0 { + return Ok(CheckpointOutcome::Done); + } + + if !all_compacting { + match table.compact_lsm().await { + Ok(()) => {} + Err(e) if is_lost_claim(&e) => return Ok(CheckpointOutcome::ReissueFromFlush), + Err(e) if !is_retryable(&e) => return Err(e), + // A 429 here means the server could latch no bucket at all, + // which the poll above already handles. Not retried in place: + // the latch it would contend for is the one doing the work, so + // fall through and re-read — `POLL_INTERVAL` is the backoff. + Err(_) => {} + } + } + tokio::time::sleep(POLL_INTERVAL).await; + } +} + +#[cfg(all(test, feature = "remote"))] +mod tests { + use super::*; + + fn http(status: u16) -> Error { + Error::Http { + source: "server said no".into(), + request_id: "rid".into(), + status_code: reqwest::StatusCode::from_u16(status).ok(), + } + } + + /// Every status the loop acts on. The two predicates are checked together + /// because their overlap is what would be wrong: a status must never be + /// both, and 421 in particular must not read as retryable — retrying it in + /// place re-issues the call that just said the node holds no claim. + #[test] + fn taxonomy_round_trips() { + for status in [429, 503] { + assert!(is_retryable(&http(status)), "{status} must retry"); + assert!( + !is_lost_claim(&http(status)), + "{status} is not a lost claim" + ); + } + assert!(is_lost_claim(&http(421)), "a lost claim must re-claim"); + assert!( + !is_retryable(&http(421)), + "retrying a lost claim in place only asks the same node again" + ); + for status in [400, 404, 409, 500] { + assert!(!is_retryable(&http(status)), "{status} is terminal"); + assert!(!is_lost_claim(&http(status)), "{status} is terminal"); + } + } + + /// An error carrying no status has no retry story and must be terminal — + /// a connection that never reached the server, or a `TableNotFound` that + /// `check_table_response` translated before the loop saw it. + #[test] + fn errors_without_a_status_are_terminal() { + let no_status = Error::Http { + source: "connection reset".into(), + request_id: "rid".into(), + status_code: None, + }; + assert!(!is_retryable(&no_status)); + assert!(!is_lost_claim(&no_status)); + + let translated = Error::TableNotFound { + name: "t".into(), + source: "gone".into(), + }; + assert!(!is_retryable(&translated)); + assert!(!is_lost_claim(&translated)); + } +} diff --git a/rust/lancedb/src/table/lsm_stats.rs b/rust/lancedb/src/table/lsm_stats.rs new file mode 100644 index 000000000..953aea90f --- /dev/null +++ b/rust/lancedb/src/table/lsm_stats.rs @@ -0,0 +1,162 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The LanceDB Authors + +//! Live per-bucket LSM state — the shape [`crate::Table::get_lsm_stats`] +//! returns and [`super::checkpoint`] polls. +//! +//! Nothing here is derived: sums and differences (total L0 bytes, WAL lag) +//! are the caller's to compute. There is no "WAL is off" shape — that case is +//! `None`, because a struct of zeros would read as measurements. + +use serde::Deserialize; + +/// One flushed L0 generation. +#[derive(Debug, Clone, Deserialize)] +pub struct GenerationStats { + pub generation: u64, + pub bytes: u64, + /// Present only when `include_generation_rows` was requested. Off by + /// default because each count opens an uncached Lance dataset, and the + /// checkpoint loop polls this route needing only generation numbers. + #[serde(default)] + pub rows: Option, +} + +/// One in-memory memtable. +#[derive(Debug, Clone, Deserialize)] +pub struct MemtableStats { + pub generation: u64, + pub rows: u64, + pub bytes: u64, + pub batches: u64, + /// Names of the indexes this memtable carries. An absent name is the whole + /// answer to "why is my fresh-tier search on that column brute-force". + pub indexes: Vec, +} + +/// Live state of one bucket. A table is N buckets on one node; flattening to +/// a single number hides the one hot bucket that is usually why someone +/// opened this endpoint. +#[derive(Debug, Clone, Deserialize)] +pub struct BucketStats { + pub shard_id: String, + /// `Active` | `Sealed` (drop-table 2PC in flight). + pub status: String, + pub writer_epoch: u64, + pub manifest_version: u64, + pub current_generation: u64, + pub replay_after_wal_entry_position: u64, + pub wal_entry_position_last_seen: u64, + pub generations: Vec, + /// Whether a pass owns this bucket's compaction latch right now. Says *a* + /// driver is running, not *whose*, and the latch is held from dispatch — + /// including while the pass queues for a pod-wide compactor permit. Read + /// it as "do not pile on", never as "mine is progressing". + pub compacting: bool, + /// Oldest first, active last. Absent for a `Sealed` bucket, whose + /// in-memory state is torn down. + #[serde(default)] + pub memtables: Option>, +} + +impl BucketStats { + /// The newest flushed generation, or `None` when L0 is empty. + pub(crate) fn newest_generation(&self) -> Option { + self.generations.iter().map(|g| g.generation).max() + } + + /// How many generations at or below `target` are still in L0. + /// + /// A count, not a boolean: one pass drains a bounded prefix rather than + /// the whole target set, so a boolean would read as "no progress" for + /// every pass but the last. Compaction drains oldest-first, so this + /// decreases monotonically. + pub(crate) fn outstanding_generations(&self, target: u64) -> usize { + self.generations + .iter() + .filter(|g| g.generation <= target) + .count() + } +} + +/// Live LSM state, one entry per bucket. +#[derive(Debug, Clone, Deserialize)] +pub struct LsmStats { + pub buckets: Vec, +} + +/// Server-side JSON envelope for `get_lsm_stats`. `lsm_stats` is null when +/// the table has no LSM write path. +#[derive(Debug, Deserialize)] +pub(crate) struct GetLsmStatsResponse { + #[serde(default)] + pub lsm_stats: Option, +} + +#[cfg(test)] +mod tests { + use super::*; + + fn bucket(shard: &str, generations: &[u64], compacting: bool) -> BucketStats { + BucketStats { + shard_id: shard.into(), + status: "Active".into(), + writer_epoch: 1, + manifest_version: 1, + current_generation: generations.iter().max().copied().unwrap_or(0) + 1, + replay_after_wal_entry_position: 0, + wal_entry_position_last_seen: 0, + generations: generations + .iter() + .map(|g| GenerationStats { + generation: *g, + bytes: 1, + rows: None, + }) + .collect(), + compacting, + memtables: None, + } + } + + /// The target watermark is the newest generation at the start, and a + /// generation created after it must not hold the loop open — that is why + /// the predicate terminates under write load. + #[test] + fn newer_generations_do_not_extend_the_target() { + let start = bucket("b0", &[7, 8], false); + let target = start.newest_generation().expect("L0 is non-empty"); + assert_eq!(target, 8); + + // Compaction drained 7 and 8; 9 and 10 arrived while it ran. + let later = bucket("b0", &[9, 10], false); + assert_eq!( + later.outstanding_generations(target), + 0, + "generations above the target are somebody else's problem" + ); + + // Still holding 8 means still outstanding. + assert_eq!( + bucket("b0", &[8, 9], false).outstanding_generations(target), + 1 + ); + } + + /// The metric counts generations, not buckets: a pass drains a bounded + /// prefix, so one bucket going 3 → 2 → 1 → 0 is three steps. + #[test] + fn progress_is_measured_in_generations() { + let target = 3; + let counts: Vec = [&[1u64, 2, 3][..], &[2, 3][..], &[3][..], &[][..]] + .iter() + .map(|gens| bucket("b0", gens, false).outstanding_generations(target)) + .collect(); + assert_eq!(counts, vec![3, 2, 1, 0]); + } + + #[test] + fn empty_l0_has_no_target() { + assert!(bucket("b0", &[], false).newest_generation().is_none()); + } +}