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()); + } }