From 106f9ddb3134d65369e172482879437cdd0ce2e3 Mon Sep 17 00:00:00 2001 From: Daniel Rammer Date: Fri, 31 Jul 2026 14:37:32 -0500 Subject: [PATCH] refactor(table): checkpoint on generation watermarks, not report bodies MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `checkpoint_lsm` looped flush → compact and read `generations_remaining` out of each compact response. Two problems. That count is stale the moment a concurrent write lands, so on a table taking writes "L0 is empty" never becomes true and the loop only ever exited on its deadline. And carrying it back at all meant eight public types describing work already visible in `get_lsm_stats`. It now fixes a target watermark — the newest generation per bucket, read after the opening flush — then triggers compaction and polls until no generation at or below that target remains. Generations created *while* it runs are deliberately not waited on, which is what makes it terminate under sustained writes and is why the deadline could go. * Deleted: `CheckpointOptions`, `CheckpointReport`, `CheckpointStopReason`, `CompactOptions`, `CompactReport`, `CompactBucketReport`, `FlushReport`, `FlushBucketReport`. `checkpoint_lsm`, `flush_lsm` and `compact_lsm` all return `()`; observability is `get_lsm_stats` before and after. * Progress, not time, is the bound: `MAX_IDLE_POLLS` consecutive polls with neither a drop in outstanding generations nor a running pass is an error. A saturated pool or a fenced writer fails loudly instead of spinning; a slow table waits, since a running pass resets the counter. Wrap the call in `tokio::time::timeout` for a wall-clock bound. * The metric counts generations, not buckets. One pass drains a bounded prefix, so a per-bucket boolean reads as "no progress" for every pass but the last — a bucket needing more passes than the bound would fail a checkpoint that was working. * `BucketStats` gains `compacting`, and the loop never triggers against a bucket already compacting: it would only collect 429s from the latch doing the work. * `get_lsm_stats` takes `include_generation_rows` again, off by default. Each count opens an uncached Lance dataset and this loop polls the route. * `send_lsm_route` splits from its JSON decode — flush and compact answer 202 with no body, and decoding an empty string is a parse error. Co-Authored-By: Claude Opus 5 (1M context) --- python/python/lancedb/_lancedb.pyi | 13 +- python/python/lancedb/table.py | 105 +++----- python/src/table.rs | 122 ++------- rust/lancedb/Cargo.toml | 2 +- rust/lancedb/src/remote/table.rs | 381 +++++++++++++++++---------- rust/lancedb/src/table.rs | 239 +++++++++-------- rust/lancedb/src/table/checkpoint.rs | 367 +++++++++++++------------- 7 files changed, 638 insertions(+), 591 deletions(-) diff --git a/python/python/lancedb/_lancedb.pyi b/python/python/lancedb/_lancedb.pyi index c3e39cabc..c0848ec4c 100644 --- a/python/python/lancedb/_lancedb.pyi +++ b/python/python/lancedb/_lancedb.pyi @@ -284,15 +284,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, - deadline_secs: Optional[float], - max_generations_per_bucket: Optional[int], - include_stats: bool, - ) -> dict: ... - async def flush_lsm(self) -> dict: ... - async def compact_lsm(self, max_generations_per_bucket: Optional[int]) -> dict: ... - async def get_lsm_stats(self) -> Optional[dict]: ... + 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 62cad4c5d..3fa003412 100644 --- a/python/python/lancedb/table.py +++ b/python/python/lancedb/table.py @@ -3872,41 +3872,29 @@ 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, - *, - deadline_secs: Optional[float] = 300.0, - max_generations_per_bucket: Optional[int] = None, - include_stats: bool = False, - ) -> dict: + def checkpoint_lsm(self) -> None: """Synchronous version of [`AsyncTable.checkpoint_lsm`][lancedb.AsyncTable.checkpoint_lsm].""" - return LOOP.run( - self._table.checkpoint_lsm( - deadline_secs=deadline_secs, - max_generations_per_bucket=max_generations_per_bucket, - include_stats=include_stats, - ) - ) + return LOOP.run(self._table.checkpoint_lsm()) - def flush_lsm(self) -> dict: + 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, *, max_generations_per_bucket: Optional[int] = None) -> dict: + def compact_lsm(self) -> None: """Synchronous version of [`AsyncTable.compact_lsm`][lancedb.AsyncTable.compact_lsm].""" - return LOOP.run( - self._table.compact_lsm( - max_generations_per_bucket=max_generations_per_bucket - ) - ) + return LOOP.run(self._table.compact_lsm()) - def get_lsm_stats(self) -> Optional[dict]: + 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()) + 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 @@ -4618,43 +4606,25 @@ class AsyncTable: """ return await self._inner.get_lsm_write_spec() - async def checkpoint_lsm( - self, - *, - deadline_secs: Optional[float] = 300.0, - max_generations_per_bucket: Optional[int] = None, - include_stats: bool = False, - ) -> dict: + 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 bounded compaction - passes until L0 is empty. The loop runs client-side, so each request - does a bounded unit of work and reports what is left. + 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: nothing is frozen, so with writes flowing there may be - new rows by the time it returns — ``converged`` means L0 was empty as - of the last pass. Idempotent and safe on a cadence; an - already-converged table costs one round trip and zero compaction - passes. + Best-effort: nothing is frozen. 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. - Parameters - ---------- - deadline_secs - Stop and report ``converged=False`` after this much wall clock. - ``None`` runs until the table converges — safe only when you know - it is not under write load. - max_generations_per_bucket - Bound on generations merged **per bucket** per request. An - N-bucket table admits ``N * max_generations_per_bucket`` merges - per call. - include_stats - Attach a ``get_lsm_stats`` block to the report. + There is no deadline. It returns when the target generations are + gone, or raises if the table stops making progress with no + compaction running. Wrap it in ``asyncio.wait_for`` for a wall-clock + bound. """ - return await self._inner.checkpoint_lsm( - deadline_secs, max_generations_per_bucket, include_stats - ) + return await self._inner.checkpoint_lsm() - async def flush_lsm(self) -> dict: + 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 @@ -4663,18 +4633,18 @@ class AsyncTable: """ return await self._inner.flush_lsm() - async def compact_lsm( - self, *, max_generations_per_bucket: Optional[int] = None - ) -> dict: - """Run one bounded L0 to base compaction pass per bucket. + async def compact_lsm(self) -> None: + """Trigger a background L0 to base compaction pass per bucket. - One pass, not convergence: that is what bounds each request's cost - and gives a caller driving its own cadence a progress signal per - round trip. Use `checkpoint_lsm` to loop to convergence. + 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(max_generations_per_bucket) + return await self._inner.compact_lsm() - async def get_lsm_stats(self) -> Optional[dict]: + 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 @@ -4684,8 +4654,15 @@ class AsyncTable: Returns ``None`` — and only — when the LSM write path is not enabled for the table. + + 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 + method needing only generation numbers. """ - return await self._inner.get_lsm_stats() + 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 73697df76..8cc13ccc7 100644 --- a/python/src/table.rs +++ b/python/src/table.rs @@ -67,6 +67,7 @@ fn lsm_stats_to_py(py: Python<'_>, stats: &lancedb::table::LsmStats) -> PyResult generations.append(ge)?; } e.set_item("generations", generations)?; + e.set_item("compacting", b.compacting)?; e.set_item( "memtables", @@ -1329,122 +1330,45 @@ impl Table { /// Converge the table's LSM write path into its base table. /// - /// Returns a dict mirroring `CheckpointReport`. Best-effort: with - /// writes flowing, new rows may land after the last pass. - #[pyo3(signature = (deadline_secs=None, max_generations_per_bucket=None, include_stats=false))] - pub fn checkpoint_lsm( - self_: PyRef<'_, Self>, - deadline_secs: Option, - max_generations_per_bucket: Option, - include_stats: bool, - ) -> PyResult> { + /// 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(); - let opts = lancedb::table::CheckpointOptions { - deadline: deadline_secs.map(std::time::Duration::from_secs_f64), - max_generations_per_bucket, - include_stats, - }; future_into_py(self_.py(), async move { - let report = inner.checkpoint_lsm(opts).await.infer_error()?; - Python::attach(|py| { - let d = PyDict::new(py); - d.set_item("converged", report.converged)?; - d.set_item( - "stop_reason", - report.stop_reason.map(|r| match r { - lancedb::table::CheckpointStopReason::DeadlineExceeded => { - "deadline_exceeded" - } - lancedb::table::CheckpointStopReason::NodeDraining => "node_draining", - lancedb::table::CheckpointStopReason::RepeatedlyUnclaimed => { - "repeatedly_unclaimed" - } - }), - )?; - d.set_item("rows_sealed", report.rows_sealed)?; - d.set_item("generations_compacted", report.generations_compacted)?; - d.set_item("rows_merged", report.rows_merged)?; - d.set_item("generations_remaining", report.generations_remaining)?; - d.set_item("compact_calls", report.compact_calls)?; - d.set_item( - "stats", - report.stats.map(|s| lsm_stats_to_py(py, &s)).transpose()?, - )?; - Ok(d.unbind()) - }) + inner.checkpoint_lsm().await.infer_error() }) } - /// Seal every bucket's active memtable into L0. Returns a dict - /// mirroring `FlushReport`. + /// 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 { - let report = inner.flush_lsm().await.infer_error()?; - Python::attach(|py| { - let d = PyDict::new(py); - let buckets = PyList::empty(py); - // Per-bucket detail is preserved rather than flattened: a - // table is N buckets on one node, and the one hot bucket is - // usually why someone is looking. - for b in &report.buckets { - let e = PyDict::new(py); - e.set_item("shard_id", &b.shard_id)?; - e.set_item("sealed_generation", b.sealed_generation)?; - e.set_item("rows_sealed", b.rows_sealed)?; - e.set_item("generations_remaining", b.generations_remaining)?; - buckets.append(e)?; - } - d.set_item("buckets", buckets)?; - d.set_item("rows_sealed", report.rows_sealed)?; - d.set_item("generations_remaining", report.generations_remaining)?; - Ok(d.unbind()) - }) - }) + future_into_py( + self_.py(), + async move { inner.flush_lsm().await.infer_error() }, + ) } - /// Run one bounded L0 → base pass per bucket. Returns a dict mirroring - /// `CompactReport`. - #[pyo3(signature = (max_generations_per_bucket=None))] - pub fn compact_lsm( - self_: PyRef<'_, Self>, - max_generations_per_bucket: Option, - ) -> PyResult> { + /// 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(); - let opts = lancedb::table::CompactOptions { - max_generations_per_bucket, - }; future_into_py(self_.py(), async move { - let report = inner.compact_lsm(opts).await.infer_error()?; - Python::attach(|py| { - let d = PyDict::new(py); - let buckets = PyList::empty(py); - for b in &report.buckets { - let e = PyDict::new(py); - e.set_item("shard_id", &b.shard_id)?; - e.set_item("generations_consumed", b.generations_consumed)?; - e.set_item("rows_merged", b.rows_merged)?; - e.set_item("generations_remaining", b.generations_remaining)?; - buckets.append(e)?; - } - d.set_item("buckets", buckets)?; - d.set_item("generations_consumed", report.generations_consumed)?; - d.set_item("rows_merged", report.rows_merged)?; - d.set_item("generations_remaining", report.generations_remaining)?; - d.set_item( - "highest_wal_entry_position", - report.highest_wal_entry_position, - )?; - Ok(d.unbind()) - }) + inner.compact_lsm().await.infer_error() }) } /// Live LSM state, or `None` when the LSM write path is not enabled. - pub fn get_lsm_stats(self_: PyRef<'_, Self>) -> PyResult> { + #[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().await.infer_error()?; + 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()) }) } diff --git a/rust/lancedb/Cargo.toml b/rust/lancedb/Cargo.toml index 7b016556f..5e8afd764 100644 --- a/rust/lancedb/Cargo.toml +++ b/rust/lancedb/Cargo.toml @@ -98,7 +98,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 e6485716d..7923db783 100644 --- a/rust/lancedb/src/remote/table.rs +++ b/rust/lancedb/src/remote/table.rs @@ -21,6 +21,7 @@ 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; @@ -31,7 +32,6 @@ use crate::table::merge::MergeFilter; use crate::table::query::create_multi_vector_plan; use crate::table::{AlterColumnsResult, FieldMetadataUpdate, UpdateFieldMetadataResult}; use crate::table::{AnyQuery, Filter, Predicate, PreprocessingOutput, TableStatistics}; -use crate::table::{CompactOptions, CompactReport, FlushReport, LsmStats}; use crate::utils::background_cache::BackgroundCache; use crate::utils::{ resolve_arrow_field_path, supported_btree_data_type, supported_vector_data_type, @@ -919,11 +919,11 @@ impl RemoteTable { /// 503. Classifying here — before any generic helper touches the /// response — is what preserves a distinction threaded through three /// crates and two error enums. - async fn send_lsm_route( + async fn send_lsm_route( &self, request: RequestBuilder, route: &str, - ) -> Result { + ) -> Result<(String, String)> { // `with_retry = false` on purpose. The transport retry layer treats // every 503 alike, so it would burn its whole budget re-asking a // node that already said, terminally, that it is draining — and it @@ -940,6 +940,18 @@ impl RemoteTable { message: format!("{route}: {status}: {body}"), }); } + Ok((body, request_id)) + } + + /// [`Self::send_lsm_route`] plus a JSON decode. Separate because + /// `flush_lsm` and `compact_lsm` answer `202` with no body at all — + /// decoding an empty string is a parse error, not an empty struct. + async fn send_lsm_route_json( + &self, + request: RequestBuilder, + route: &str, + ) -> Result { + let (body, request_id) = self.send_lsm_route(request, route).await?; serde_json::from_str(&body).map_err(|e| Error::Http { source: format!("Failed to parse {route} response: {e}").into(), request_id, @@ -2401,27 +2413,31 @@ impl BaseTable for RemoteTable { }) } - async fn flush_lsm(&self) -> Result { + async fn flush_lsm(&self) -> Result<()> { let request = self .client .post(&format!("/v1/table/{}/flush_lsm/", self.identifier)); - self.send_lsm_route(request, "flush_lsm").await + self.send_lsm_route(request, "flush_lsm").await?; + Ok(()) } - async fn compact_lsm(&self, opts: CompactOptions) -> Result { + async fn compact_lsm(&self) -> Result<()> { let request = self .client - .post(&format!("/v1/table/{}/compact_lsm/", self.identifier)) - .json(&serde_json::json!({ - "max_generations_per_bucket": opts.max_generations_per_bucket, - })); - self.send_lsm_route(request, "compact_lsm").await + .post(&format!("/v1/table/{}/compact_lsm/", self.identifier)); + self.send_lsm_route(request, "compact_lsm").await?; + Ok(()) } - async fn get_lsm_stats(&self) -> Result> { + 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)); - let parsed: GetLsmStatsResponse = self.send_lsm_route(request, "get_lsm_stats").await?; + let request = self + .post_read(&format!("/v1/table/{}/get_lsm_stats/", self.identifier)) + .json(&serde_json::json!({ + "include_generation_rows": include_generation_rows, + })); + let parsed: GetLsmStatsResponse = + self.send_lsm_route_json(request, "get_lsm_stats").await?; // `null` — and only — when the table has no LSM write path. Ok(parsed.lsm_stats) } @@ -5830,142 +5846,168 @@ mod tests { assert!(table.get_lsm_write_spec().await.unwrap().is_none()); } - /// A flush that lands in an empty L0 converges on the flush response + /// 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 that lands in an empty L0 finishes on the opening stats read /// alone — **zero** `compact` calls issued. Asserting the call count is - /// the point: `generations_consumed: 0` is also true of a loop that ran - /// a pointless pass. - #[tokio::test] + /// 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 calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); - let seen = calls.clone(); + 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"); } - assert_eq!(path, "/v1/table/my_table/flush_lsm/"); - let response = serde_json::json!({ - "buckets": [{ - "shard_id": "b0", - "sealed_generation": 7, - "rows_sealed": 12, - "generations_remaining": 0, - }], - "rows_sealed": 12, - "generations_remaining": 0, - }); - http::Response::builder() - .status(200) - .body(response.to_string()) - .unwrap() + if path.contains("flush_lsm") { + return accepted(); + } + assert_eq!(path, "/v1/table/my_table/get_lsm_stats/"); + ok_json(stats_body(&[], false)) }); - let report = table - .checkpoint_lsm(crate::table::CheckpointOptions::default()) - .await - .unwrap(); - assert!(report.converged); - assert_eq!(report.rows_sealed, 12); - assert_eq!(report.compact_calls, 0); - assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 0); + table.checkpoint_lsm().await.unwrap(); + assert_eq!(compacts.load(std::sync::atomic::Ordering::SeqCst), 0); } - /// The loop drives `compact` until the *server* reports zero remaining, - /// accumulating what each pass merged. - #[tokio::test] - async fn test_checkpoint_loops_compact_until_converged() { - let remaining = Arc::new(std::sync::atomic::AtomicUsize::new(3)); + /// 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(); - let response = if path.contains("flush_lsm") { - serde_json::json!({ - "buckets": [{ "shard_id": "b0", "sealed_generation": 4, - "rows_sealed": 30, "generations_remaining": 3 }], - "rows_sealed": 30, - "generations_remaining": 3, - }) - } else { - let left = remaining.fetch_sub(1, std::sync::atomic::Ordering::SeqCst) - 1; - serde_json::json!({ - "buckets": [{ "shard_id": "b0", "generations_consumed": 1, - "rows_merged": 10, "generations_remaining": left }], - "generations_consumed": 1, - "rows_merged": 10, - "generations_remaining": left, - "highest_wal_entry_position": 991823, - }) - }; - http::Response::builder() - .status(200) - .body(response.to_string()) - .unwrap() + 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)) }); - let report = table - .checkpoint_lsm(crate::table::CheckpointOptions::default()) - .await - .unwrap(); - assert!(report.converged); - assert_eq!(report.stop_reason, None); - assert_eq!(report.compact_calls, 3); - assert_eq!(report.generations_compacted, 3); - assert_eq!(report.rows_merged, 30); - assert_eq!(report.generations_remaining, 0); + 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. This + /// is what makes the loop terminate on a table taking writes, where + /// "L0 is empty" never becomes true — so the assertion is that it + /// returns at all, with the target set drained and newer L0 present. + #[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 not draining.** A 429 must be retried, not read as /// the one terminal condition. This is the regression that would pass /// silently otherwise: every layer says "unavailable" either way, and /// the checkpoint just stops early blaming a healthy node. - #[tokio::test] + #[tokio::test(start_paused = true)] async fn test_checkpoint_retries_contention_and_does_not_report_draining() { let compacts = Arc::new(std::sync::atomic::AtomicUsize::new(0)); let seen = compacts.clone(); let table = Table::new_with_handler("my_table", move |request| { - if request.url().path().contains("flush_lsm") { - let response = serde_json::json!({ - "buckets": [{ "shard_id": "b0", "sealed_generation": 1, - "rows_sealed": 5, "generations_remaining": 1 }], - "rows_sealed": 5, - "generations_remaining": 1, - }); - return http::Response::builder() - .status(200) - .body(response.to_string()) - .unwrap(); + let path = request.url().path().to_string(); + if path.contains("flush_lsm") { + return accepted(); } - // First two compacts: latch contention. Then it clears. - 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(); + 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 response = serde_json::json!({ - "buckets": [{ "shard_id": "b0", "generations_consumed": 1, - "rows_merged": 5, "generations_remaining": 0 }], - "generations_consumed": 1, - "rows_merged": 5, - "generations_remaining": 0, - "highest_wal_entry_position": 1, - }); - http::Response::builder() - .status(200) - .body(response.to_string()) - .unwrap() + 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)) }); - let report = table - .checkpoint_lsm(crate::table::CheckpointOptions::default()) + table + .checkpoint_lsm() .await - .unwrap(); - assert!(report.converged, "contention must not abort the checkpoint"); - assert_ne!( - report.stop_reason, - Some(crate::table::CheckpointStopReason::NodeDraining), - "a busy compactor is not a draining node" - ); + .expect("contention must not abort the checkpoint"); assert_eq!( compacts.load(std::sync::atomic::Ordering::SeqCst), 3, @@ -5974,9 +6016,9 @@ mod tests { } /// A draining node stops the loop immediately — **without retrying**. - /// Asserting the call count matters: a regression to retry-until- - /// deadline passes any test that only checks the outcome. - #[tokio::test] + /// Asserting the call count matters: a regression to retry-until-the- + /// idle-bound passes any test that only checks that it errored. + #[tokio::test(start_paused = true)] async fn test_checkpoint_stops_without_retrying_on_draining() { let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); let seen = calls.clone(); @@ -5988,14 +6030,16 @@ mod tests { .unwrap() }); - let report = table - .checkpoint_lsm(crate::table::CheckpointOptions::default()) - .await - .unwrap(); - assert!(!report.converged); - assert_eq!( - report.stop_reason, - Some(crate::table::CheckpointStopReason::NodeDraining) + let err = table.checkpoint_lsm().await.unwrap_err(); + assert!( + matches!( + err, + Error::LsmRoute { + fault: crate::table::LsmFault::Draining, + .. + } + ), + "draining must surface as itself, not as a generic failure: {err:?}" ); assert_eq!( calls.load(std::sync::atomic::Ordering::SeqCst), @@ -6004,12 +6048,79 @@ mod tests { ); } + /// A table that never drains its targets, with no pass running, must + /// fail rather than spin — this is the bound that replaced the + /// deadline. A saturated compactor pool looks exactly like this. + #[tokio::test(start_paused = true)] + async fn test_checkpoint_gives_up_when_no_progress_and_nothing_running() { + 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(); + } + // Triggers land, nothing ever compacts. + ok_json(stats_body(&[1, 2], false)) + }); + + let err = table.checkpoint_lsm().await.unwrap_err(); + let message = err.to_string(); + assert!( + message.contains("stopped making progress"), + "the error must name the stall, not a generic failure: {message}" + ); + } + + /// A pass *is* running on every outstanding bucket: that counts as + /// progress, so the loop waits instead of tripping the idle bound and + /// instead of piling on triggers the latch would only reject. + #[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(); + } + // Busy for well past MAX_IDLE_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. + /// 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": [{ @@ -6021,6 +6132,7 @@ mod tests { "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"] } @@ -6035,15 +6147,16 @@ mod tests { }); let stats = table - .get_lsm_stats() + .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, 30); + 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(); @@ -6058,7 +6171,7 @@ mod tests { .body(serde_json::json!({ "lsm_stats": null }).to_string()) .unwrap() }); - assert!(table.get_lsm_stats().await.unwrap().is_none()); + assert!(table.get_lsm_stats(false).await.unwrap().is_none()); } #[tokio::test] diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs index 43dd95156..5e3343462 100644 --- a/rust/lancedb/src/table.rs +++ b/rust/lancedb/src/table.rs @@ -83,12 +83,11 @@ pub use branch_merge::{ BranchDiff, ColumnChange, ColumnSummary, IndexSummary, MergeBlocker, MergeBlockerCode, MergeBranchResult, MergeBranchStatus, MergePreview, RowCountSummary, }; -pub use checkpoint::{ - BucketStats, CheckpointOptions, CheckpointReport, CheckpointStopReason, CompactBucketReport, - CompactOptions, CompactReport, FlushBucketReport, FlushReport, GenerationStats, LsmFault, - LsmStats, MemtableStats, +pub use checkpoint::{BucketStats, GenerationStats, LsmFault, LsmStats, MemtableStats}; +use checkpoint::{ + CheckpointControl, CheckpointOutcome, MAX_IDLE_POLLS, MAX_REISSUES, POLL_INTERVAL, backoff, + checkpoint_fault, }; -use checkpoint::{CheckpointControl, checkpoint_fault}; pub use chrono::Duration; pub use delete::DeleteResult; use futures::future::join_all; @@ -637,15 +636,15 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync { /// Seal every bucket's active memtable into L0. /// /// The default implementation returns `NotSupported`. - async fn flush_lsm(&self) -> Result { + async fn flush_lsm(&self) -> Result<()> { Err(Error::NotSupported { message: "flush_lsm is not supported on this table type".into(), }) } - /// Run one bounded L0 → base compaction pass per bucket. + /// Trigger a background L0 → base compaction pass per bucket. /// /// The default implementation returns `NotSupported`. - async fn compact_lsm(&self, _opts: CompactOptions) -> Result { + async fn compact_lsm(&self) -> Result<()> { Err(Error::NotSupported { message: "compact_lsm is not supported on this table type".into(), }) @@ -654,7 +653,7 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync { /// enabled for this table. /// /// The default implementation returns `NotSupported`. - async fn get_lsm_stats(&self) -> Result> { + 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(), }) @@ -1654,115 +1653,145 @@ impl Table { /// Converge this table's LSM write path into its base table. /// - /// One `flush` (seal every memtable into L0), then `compact` until L0 - /// is empty. The loop runs here, in the client: each request does a - /// bounded unit of work and reports what is left, so there is no held - /// socket, no server-side task, and nothing to reconcile if you drop - /// this future partway through. + /// 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 here, in the client: `compact_lsm` + /// dispatches a pass and returns, and progress is read from generation + /// numbers via `get_lsm_stats`, so there is no held socket, no + /// server-side task, and nothing to reconcile if you drop this future + /// partway through. /// - /// **Best-effort.** Nothing is frozen, so with writes flowing there may - /// be new rows by the time it returns — `converged` means L0 was empty - /// as of the last pass. Idempotent and safe to run on a cadence: an - /// already-converged table costs one round trip and zero compaction - /// passes, because the opening flush reports an empty L0 and the loop - /// is never entered. + /// **Best-effort.** Nothing is frozen. Generations created *after* the + /// opening flush are deliberately not waited on — that is what lets + /// this terminate on a table taking writes, where "L0 is empty" never + /// becomes true. Idempotent and safe to run on a cadence: an + /// already-converged table costs two round trips and triggers nothing. /// - /// Two independent bounds end the loop: `generations_remaining == 0` - /// and [`CheckpointOptions::deadline`]. A table under sustained write - /// load exits on the deadline with `converged: false` — a *result*, not - /// an error. + /// **No deadline.** It returns when the target generations are gone, or + /// errors after [`MAX_IDLE_POLLS`] polls with neither progress nor a + /// running pass — a fenced bucket, a saturated compactor pool, a + /// failing merge. Wrap it in `tokio::time::timeout` if you need a + /// wall-clock bound. /// /// # Example /// /// ```no_run - /// # use lancedb::table::{Table, CheckpointOptions}; + /// # use lancedb::Table; /// # async fn example(table: &Table) -> Result<(), Box> { - /// let report = table.checkpoint_lsm(CheckpointOptions::default()).await?; - /// if !report.converged { - /// println!("stopped early: {:?}", report.stop_reason); - /// } + /// 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, opts: CheckpointOptions) -> Result { - let started = std::time::Instant::now(); - let expired = |elapsed: std::time::Duration| { - opts.deadline.is_some_and(|deadline| elapsed >= deadline) - }; - - // Cap re-issues from `flush` after a 404. A node in a crash loop - // otherwise turns flush → compact → 404 → flush into a spin that - // burns the whole deadline making no progress and reports nothing - // useful about why. - const MAX_REISSUES: usize = 3; - - let mut report = CheckpointReport { - converged: false, - stop_reason: None, - rows_sealed: 0, - generations_compacted: 0, - rows_merged: 0, - generations_remaining: 0, - compact_calls: 0, - stats: None, - }; - let compact_opts = CompactOptions { - max_generations_per_bucket: opts.max_generations_per_bucket, - }; - - 'reissue: for attempt in 0..=MAX_REISSUES { - let flushed = match self.flush_lsm().await { - Ok(f) => f, - Err(e) => match checkpoint_fault(e, &mut report, attempt, MAX_REISSUES).await? { - // A retryable flush is just re-issued; the seal is - // idempotent (sealing an empty active memtable is a - // no-op, so repeats do not churn empty generations). + pub async fn checkpoint_lsm(&self) -> Result<()> { + for attempt in 0..=MAX_REISSUES { + // The seal is what turns everything written before this call + // into a generation, so the watermark has to be read after it. + // It is also idempotent: sealing an empty active memtable is a + // no-op, so a re-issue does not churn empty generations. + match self.flush_lsm().await { + Ok(()) => {} + Err(e) => match checkpoint_fault(&e, attempt)? { CheckpointControl::Retry | CheckpointControl::ReissueFromFlush => { - continue 'reissue; + backoff(attempt).await; + continue; } - CheckpointControl::Stop => break 'reissue, }, + } + + let Some(stats) = self.get_lsm_stats(false).await? else { + // Not WAL-backed: nothing to converge, and `flush_lsm` + // would have errored first on every path but a race. + return Ok(()); }; - // Only the first flush's seal counts toward `rows_sealed`; a - // re-issue after a 404 re-seals rows a previous attempt already - // reported. - if attempt == 0 { - report.rows_sealed = flushed.rows_sealed; + let targets: HashMap = stats + .buckets + .iter() + .filter_map(|b| Some((b.shard_id.clone(), b.newest_generation()?))) + .collect(); + if targets.is_empty() { + return Ok(()); } - report.generations_remaining = flushed.generations_remaining; - report.converged = flushed.generations_remaining == 0; - while !report.converged { - if expired(started.elapsed()) { - report.stop_reason = Some(CheckpointStopReason::DeadlineExceeded); - break 'reissue; - } - let compacted = match self.compact_lsm(compact_opts.clone()).await { - Ok(c) => c, - // A retryable fault re-enters this loop; a 404 restarts - // from `flush`, which is the call that re-claims and - // replays. - Err(e) => { - match checkpoint_fault(e, &mut report, attempt, MAX_REISSUES).await? { - CheckpointControl::Retry => continue, - CheckpointControl::ReissueFromFlush => continue 'reissue, - CheckpointControl::Stop => break 'reissue, - } - } + match self.drain_to_targets(&targets, attempt).await? { + CheckpointOutcome::Done => return Ok(()), + CheckpointOutcome::ReissueFromFlush => 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. Split out so the flush re-issue path above stays readable. + async fn drain_to_targets( + &self, + targets: &HashMap, + attempt: usize, + ) -> Result { + let mut idle_polls = 0; + let mut last_outstanding = usize::MAX; + loop { + let Some(stats) = self.get_lsm_stats(false).await? else { + return Ok(CheckpointOutcome::Done); + }; + let mut outstanding = 0; + let mut buckets_left = 0; + let mut all_compacting = true; + for b in &stats.buckets { + let Some(target) = targets.get(&b.shard_id) else { + continue; }; - report.compact_calls += 1; - report.generations_compacted += compacted.generations_consumed; - report.rows_merged += compacted.rows_merged; - report.generations_remaining = compacted.generations_remaining; - report.converged = compacted.generations_remaining == 0; + let n = b.outstanding_generations(*target); + if n > 0 { + outstanding += n; + buckets_left += 1; + all_compacting &= b.compacting; + } + } + if outstanding == 0 { + return Ok(CheckpointOutcome::Done); } - break; - } - if opts.include_stats { - report.stats = self.get_lsm_stats().await?; + // Progress, not time, is the bound. A pass already running on + // every outstanding bucket counts as progress: piling on would + // only collect 429s, and the latch it would contend for is the + // one doing the work. + if outstanding < last_outstanding || all_compacting { + idle_polls = 0; + } else { + idle_polls += 1; + if idle_polls >= MAX_IDLE_POLLS { + return Err(Error::Runtime { + message: format!( + "checkpoint_lsm: {outstanding} generation(s) across {buckets_left} \ + bucket(s) stopped making progress with no compaction running; the \ + compactor pool may be saturated or the writer fenced" + ), + }); + } + } + last_outstanding = outstanding; + + if !all_compacting { + match self.compact_lsm().await { + Ok(()) => {} + Err(e) => match checkpoint_fault(&e, attempt)? { + // Every bucket busy (429) is the state the poll + // above already handles; fall through and re-read. + CheckpointControl::Retry => {} + CheckpointControl::ReissueFromFlush => { + return Ok(CheckpointOutcome::ReissueFromFlush); + } + }, + } + } + tokio::time::sleep(POLL_INTERVAL).await; } - Ok(report) } /// Seal every bucket's active memtable into L0 without touching the @@ -1773,7 +1802,7 @@ impl Table { /// (after a restart, say) this claims it and replays its WAL log first — /// reporting "nothing to flush" without replaying would be lying about /// durable data. - pub async fn flush_lsm(&self) -> Result { + pub async fn flush_lsm(&self) -> Result<()> { self.inner.flush_lsm().await } @@ -1783,8 +1812,8 @@ impl Table { /// One pass, not convergence: that is what makes each request's cost /// bounded and gives a caller driving its own cadence a progress signal /// per round trip. - pub async fn compact_lsm(&self, opts: CompactOptions) -> Result { - self.inner.compact_lsm(opts).await + pub async fn compact_lsm(&self) -> Result<()> { + self.inner.compact_lsm().await } /// Read live per-bucket LSM state. @@ -1794,6 +1823,10 @@ impl Table { /// 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 method needing only generation numbers. + /// /// `Ok(None)` — and only — when the LSM write path is not enabled for /// the table, matching [`Table::get_lsm_write_spec`]. Stats is /// fresh-tier only, so with the WAL off there is no manifest and no @@ -1801,8 +1834,8 @@ impl Table { /// /// 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) -> Result> { - self.inner.get_lsm_stats().await + 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. diff --git a/rust/lancedb/src/table/checkpoint.rs b/rust/lancedb/src/table/checkpoint.rs index 1d392864e..bb75178ff 100644 --- a/rust/lancedb/src/table/checkpoint.rs +++ b/rust/lancedb/src/table/checkpoint.rs @@ -3,24 +3,26 @@ //! Converging a table's LSM write path into its base table. //! -//! `checkpoint_lsm` is `flush` then `compact`, repeated until the fresh -//! tier is empty — and the loop runs **here, in the client**, not on the -//! server. Putting it server-side 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. +//! `checkpoint_lsm` seals once, then triggers compaction and watches +//! generation numbers until the L0 that existed at the start is gone. The +//! loop runs **here, in the client**, not on the server: putting it +//! server-side would mean a background task, a single-flight intent, an +//! intent that leaks on panic, an "is it done" observable, and a story for +//! every way a client can vanish mid-operation. //! //! Three properties follow: //! -//! * **No held socket.** Each request does a bounded unit of work and -//! returns. Nothing sits on a connection for minutes waiting on a merge. -//! * **Termination is structural.** Each `compact` drains at most one -//! prefix per bucket and reports what is left. A caller that wants to -//! stop, stops. -//! * **Completion is carried in the responses, not inferred.** A poll-a -//! -shared-counter design cannot distinguish "converged" from "hasn't -//! started yet"; being *told* `generations_remaining == 0` can. +//! * **No held socket.** `compact_lsm` dispatches a pass and returns; the +//! merge runs on the server's compactor pool. Nothing sits on a +//! connection for minutes. +//! * **The predicate is durable state, not a response body.** Completion +//! is read from generation numbers in the shard manifest via +//! `get_lsm_stats`. A count carried back in a compact response is stale +//! the moment a concurrent write lands; a generation number is not. +//! * **Termination is reachable under write load.** The target set is +//! fixed at the start, so generations created *during* the checkpoint +//! are ignored. A predicate of "L0 is empty" chases a moving target on a +//! table taking writes and only ever exits on a timeout. //! //! **Best-effort by construction.** Nothing is frozen, so a checkpoint //! converges the fresh tier *as of some instant*; with writes flowing @@ -34,143 +36,16 @@ use std::time::Duration; use serde::Deserialize; -/// Options for [`crate::Table::checkpoint_lsm`]. -#[derive(Debug, Clone)] -pub struct CheckpointOptions { - /// Stop and report `converged: false` once this much wall-clock has - /// elapsed. `None` runs until the table converges or a terminal error - /// arrives — safe only on a table you know is not under write load. - pub deadline: Option, - /// Per-request, per-bucket bound on generations merged. `None` uses the - /// server's `compact_prefix_max`. Note this is **per bucket**: an - /// N-bucket table admits `N × max_generations_per_bucket` merges per - /// request. [`LsmStats::buckets`] has one entry per bucket. - pub max_generations_per_bucket: Option, - /// Collect a [`LsmStats`] block into the report when the loop ends. - pub include_stats: bool, -} - -impl Default for CheckpointOptions { - fn default() -> Self { - Self { - deadline: Some(Duration::from_secs(300)), - max_generations_per_bucket: None, - include_stats: false, - } - } -} - -/// Why a checkpoint stopped short of convergence. Absent ⇒ it converged. -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum CheckpointStopReason { - /// The caller's deadline elapsed with generations still in L0. A - /// *result*, not an error: a table under sustained write load may - /// legitimately never reach zero. Callers on a cadence ignore it; - /// callers about to disable the WAL check it. - DeadlineExceeded, - /// The owning node is draining. **Terminal** — the drain gate is a - /// one-way latch that never releases until the process restarts, so - /// retrying against that node spins until the deadline and then reports - /// failure when the truthful answer was available on the first - /// response. It resolves only when the ring changes. - /// - /// Stopping is also the *correct* answer: a node drain is a checkpoint - /// of every claimed table on the pod, driven by something that will - /// finish it. - NodeDraining, - /// The registry entry vanished repeatedly under pod restarts. A pod in - /// a crash loop would otherwise turn flush → compact → 404 → flush into - /// a spin that burns the whole deadline making no progress. - RepeatedlyUnclaimed, -} - -/// Outcome of a [`crate::Table::checkpoint_lsm`] run. -#[derive(Debug, Clone)] -pub struct CheckpointReport { - /// L0 was empty across every bucket as of the last pass. - /// - /// Precisely: the loop seals once at the top, so rows written *during* - /// the loop sit in the active memtable and are not sealed again. - /// Writes are never blocked to make this look better — a caller who - /// wants a quiescent table quiesces it. - pub converged: bool, - /// Set when `converged` is false. - pub stop_reason: Option, - /// Rows sealed out of the memtables by the single opening flush. - pub rows_sealed: u64, - /// L0 generations merged into base across every pass. - pub generations_compacted: usize, - /// Rows merged into base across every pass. - pub rows_merged: u64, - /// Generations still in L0 as of the last response. - pub generations_remaining: usize, - /// Number of `compact` round trips issued. Zero when the opening flush - /// landed in an empty L0 — the common case on a cadence. - pub compact_calls: usize, - /// Present when [`CheckpointOptions::include_stats`] was set. - pub stats: Option, -} - -/// Per-bucket result of one flush. -#[derive(Debug, Clone, Deserialize)] -pub struct FlushBucketReport { - pub shard_id: String, - /// `None` when the bucket's active memtable was already empty. - #[serde(default)] - pub sealed_generation: Option, - pub rows_sealed: u64, - pub generations_remaining: usize, -} - -/// Result of [`crate::Table::flush_lsm`]: everything written before the -/// seal is now in L0. -#[derive(Debug, Clone, Deserialize)] -pub struct FlushReport { - pub buckets: Vec, - pub rows_sealed: u64, - /// Summed across buckets, so `== 0` means every bucket is empty. This - /// is what lets an already-converged table cost one round trip and zero - /// compaction passes. - pub generations_remaining: usize, -} - -/// Options for [`crate::Table::compact_lsm`]. -#[derive(Debug, Clone, Default)] -pub struct CompactOptions { - /// See [`CheckpointOptions::max_generations_per_bucket`]. - pub max_generations_per_bucket: Option, -} - -/// Per-bucket result of one compaction pass. -#[derive(Debug, Clone, Deserialize)] -pub struct CompactBucketReport { - pub shard_id: String, - pub generations_consumed: usize, - pub rows_merged: u64, - pub generations_remaining: usize, -} - -/// Result of [`crate::Table::compact_lsm`]: one bounded L0 → base pass per -/// bucket. -/// -/// The per-bucket list is authoritative; the table-level fields are a -/// convenience. Consumed/merged/remaining are **sums**; -/// `highest_wal_entry_position` is a **max**, since it is a watermark. -#[derive(Debug, Clone, Deserialize)] -pub struct CompactReport { - pub buckets: Vec, - pub generations_consumed: usize, - pub rows_merged: u64, - pub generations_remaining: usize, - pub highest_wal_entry_position: u64, -} - /// One flushed L0 generation. #[derive(Debug, Clone, Deserialize)] pub struct GenerationStats { pub generation: u64, pub bytes: u64, - pub rows: 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. @@ -203,12 +78,39 @@ pub struct BucketStats { 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*: the server's periodic trigger + /// competes for the same latch, so a caller that dispatched a compact + /// cannot read this as "mine is progressing", only as "do not pile on". + 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: it is the checkpoint's progress metric, + /// and one pass drains a bounded prefix rather than the whole target + /// set. A per-bucket boolean would read as "no progress" for every + /// pass but the last, so a bucket needing more passes than the idle + /// bound would fail a checkpoint that was working correctly. + /// 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. /// /// Every field is measured. There is no "WAL is off" shape here — that @@ -282,6 +184,37 @@ pub(crate) fn classify_lsm_fault(status: u16, body: &str) -> LsmFault { } } +/// How long the loop waits between `get_lsm_stats` polls. A pass over a +/// `prefix_max` prefix takes seconds, so a shorter interval only adds +/// round trips. +pub(crate) const POLL_INTERVAL: Duration = Duration::from_secs(1); + +/// Consecutive polls tolerated with no progress *and* no pass running +/// before the checkpoint gives up. This is the bound in place of a +/// deadline: a slow table waits (a running pass resets the counter) while +/// a stuck one — a fenced bucket, a saturated compactor pool, a failing +/// merge — fails loudly instead of spinning. +pub(crate) const MAX_IDLE_POLLS: usize = 10; + +/// Cap on re-issues from `flush` after a 404. A node in a crash loop would +/// otherwise turn flush → compact → 404 → flush into a spin. +pub(crate) const MAX_REISSUES: usize = 3; + +/// Base backoff between retries, doubled per consecutive retry up to +/// [`RETRY_BACKOFF_MAX`]. Latch contention clears in about the time one +/// compaction pass takes, so starting small is right; a saturated pool +/// wants the ceiling. +pub(crate) const RETRY_BACKOFF_BASE: Duration = Duration::from_millis(100); +pub(crate) const RETRY_BACKOFF_MAX: Duration = Duration::from_secs(5); + +/// Sleep before re-issuing a retryable request. +pub(crate) 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; +} + /// What the checkpoint loop should do next after a failed request. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum CheckpointControl { @@ -289,29 +222,29 @@ pub(crate) enum CheckpointControl { Retry, /// Restart from `flush`, which is what re-claims and replays. ReissueFromFlush, - /// Stop; `report.stop_reason` has been set. - Stop, } -/// Base backoff between retries, doubled per consecutive retry up to -/// [`RETRY_BACKOFF_MAX`]. Latch contention clears in about the time one -/// compaction pass takes, so starting small is right; a saturated pool -/// wants the ceiling. -const RETRY_BACKOFF_BASE: Duration = Duration::from_millis(100); -const RETRY_BACKOFF_MAX: Duration = Duration::from_secs(5); +/// Whether the drain loop finished or needs the table re-claimed first. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum CheckpointOutcome { + Done, + ReissueFromFlush, +} -/// Decide what a failed LSM request means for the loop, recording the stop -/// reason on `report` when it is terminal. Returns `Err` only for genuinely -/// fatal faults (dropping table, not WAL-backed, anything unrecognized) — -/// those are errors the caller should see, not outcomes. -pub(crate) async fn checkpoint_fault( - e: crate::Error, - report: &mut CheckpointReport, +/// Decide what a failed LSM request means for the loop. Returns `Err` for +/// genuinely terminal faults — a draining node, a dropping table, a table +/// with no LSM write path, anything unrecognized — which are errors the +/// caller should see rather than outcomes to keep looping on. +/// +/// Draining is terminal by design: the drain gate is a one-way latch that +/// never releases until the process restarts, so retrying against that +/// node spins forever when the truthful answer was in the first response. +pub(crate) fn checkpoint_fault( + e: &crate::Error, attempt: usize, - max_reissues: usize, ) -> crate::Result { #[cfg(feature = "remote")] - let fault = match &e { + let fault = match e { crate::Error::LsmRoute { fault, .. } => Some(*fault), _ => None, }; @@ -319,25 +252,33 @@ pub(crate) async fn checkpoint_fault( let fault: Option = None; match fault { - Some(LsmFault::Retry) => { - let backoff = RETRY_BACKOFF_BASE - .saturating_mul(1u32 << attempt.min(8) as u32) - .min(RETRY_BACKOFF_MAX); - tokio::time::sleep(backoff).await; - Ok(CheckpointControl::Retry) - } - Some(LsmFault::ReissueFromFlush) if attempt < max_reissues => { + Some(LsmFault::Retry) => Ok(CheckpointControl::Retry), + Some(LsmFault::ReissueFromFlush) if attempt < MAX_REISSUES => { Ok(CheckpointControl::ReissueFromFlush) } - Some(LsmFault::ReissueFromFlush) => { - report.stop_reason = Some(CheckpointStopReason::RepeatedlyUnclaimed); - Ok(CheckpointControl::Stop) - } - Some(LsmFault::Draining) => { - report.stop_reason = Some(CheckpointStopReason::NodeDraining); - Ok(CheckpointControl::Stop) - } - Some(LsmFault::Fatal) | None => Err(e), + _ => Err(clone_lsm_error(e)), + } +} + +/// `crate::Error` is not `Clone`, and the loop borrows the error to +/// classify it before deciding whether to surface it. Rebuild the variant +/// that matters and fall back to a flattened message for the rest. +fn clone_lsm_error(e: &crate::Error) -> crate::Error { + #[cfg(feature = "remote")] + if let crate::Error::LsmRoute { + fault, + status, + message, + } = e + { + return crate::Error::LsmRoute { + fault: *fault, + status: *status, + message: message.clone(), + }; + } + crate::Error::Runtime { + message: e.to_string(), } } @@ -378,4 +319,68 @@ mod tests { assert_eq!(classify_lsm_fault(503, "gateway timeout"), LsmFault::Retry); assert_eq!(classify_lsm_fault(503, r#"{"code":"19"}"#), LsmFault::Retry); } + + 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 + /// the whole reason 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 progress metric counts generations, not buckets. A pass drains + /// a bounded prefix, so one bucket going 3 → 2 → 1 → 0 must read as + /// three steps of progress, not three idle polls. + #[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()); + } }