feat(table): checkpoint_lsm, flush_lsm, compact_lsm, get_lsm_stats (#3736)

Converge a table's LSM write path into its base table, and inspect it.

`checkpoint_lsm` is `flush` then `compact`, repeated until the fresh
tier is empty — and the loop runs **client-side**. Putting it on the
server would mean a background task, which means a single-flight intent,
an intent that leaks on panic, a bounded-iteration policy, an "is it
done" observable, and a story for every way a client can vanish
mid-operation. None of that exists in this shape: each request does a
bounded unit of work and reports what is left, so completion is *carried
in the responses* rather than inferred from a shared counter that cannot
distinguish "converged" from "hasn't started yet".

Best-effort by construction. Nothing is frozen, so `converged` means L0
was empty as of the last pass. It is idempotent, abandonable at any
point with zero consequence, and safe to run on a cadence — an
already-converged table costs one round trip and zero compaction passes,
because `flush` reports `generations_remaining` and the loop is never
entered.

## The failure taxonomy is the load-bearing part

Five distinct conditions used to arrive at a client as one 503.
`Error::LsmRoute` carries a classification read from the response body's
namespace error code **at the point of receipt** — before any generic
helper folds the body into a string and keeps only the status.

| condition | wire | client action |
|---|---|---|
| contention (latch held / pool saturated) | 429, code 21 | retry with
backoff |
| owning node draining | 503, code 19 `InvalidTableState` | **stop** |
| fenced / no slot / transport | 503, code 17 | retry with backoff |
| registry entry vanished | 404 | re-issue from `flush` (capped) |
| table being dropped / not WAL-backed | 409 / 400 | stop |

Draining is terminal because the drain gate is a one-way latch —
retrying spins until the deadline to report a failure that was knowable
on the first response. Transport retry is disabled on these routes for
the same reason: it treats every 503 alike and would burn its budget
before the classifier ever saw the body.

`get_lsm_stats` returns `Option<LsmStats>`, matching
`get_lsm_write_spec` — `None` only when the table has no LSM write path,
since a struct of zeros would read as measurements.

Python bindings mirror all four, preserving per-bucket detail rather
than flattening to a table-level summary.

## Testing

Six new unit tests against the mocked endpoint, plus the taxonomy
round-trip:
- flush into an empty L0 issues **zero** compact calls (asserts the call
count — `generations_consumed: 0` is also true of a loop that ran a
pointless pass)
- the loop drives compact until the server reports zero remaining
- **contention is not draining**: a 429 retries and converges; asserts
the retry count
- a draining node stops after **exactly one** request, no retries
- stats round-trips fully populated; `include_generation_rows` off by
default
- every `(status, code)` pair classifies correctly, including
unparseable 503 bodies falling back to *retryable* rather than terminal

`cargo test -p lancedb --features remote --lib`: 723 passed.

## Notes for review

- Depends on the sibling lance change returning `SealedGeneration` from
`force_seal_active` only at the *server* level — no lance API is used
here.
- The branch is based on `codex/update-lance-10-0-0-beta-5`, so it
carries one extra commit (`chore: update lance dependency to
v10.0.0-beta.5`) that is not part of this change.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: lancedb automation <robot@lancedb.com>
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

python/python/lancedb/_lancedb.pyi

@@ -355,6 +355,10 @@ class Table:
     async def set_lsm_write_spec(self, spec: LsmWriteSpec) -> None: ...
     async def unset_lsm_write_spec(self) -> None: ...
     async def get_lsm_write_spec(self) -> Optional[LsmWriteSpec]: ...
+    async def checkpoint_lsm(self) -> None: ...
+    async def flush_lsm(self) -> None: ...
+    async def compact_lsm(self) -> None: ...
+    async def get_lsm_stats(self, include_generation_rows: bool) -> Optional[dict]: ...
     async def close_lsm_writers(self) -> None: ...
     @property
     def tags(self) -> Tags: ...

python/python/lancedb/table.py

@@ -3976,6 +3976,28 @@ class LanceTable(Table):
         [`AsyncTable.get_lsm_write_spec`][lancedb.AsyncTable.get_lsm_write_spec]."""
         return LOOP.run(self._table.get_lsm_write_spec())
 
+    def checkpoint_lsm(self) -> None:
+        """Synchronous version of
+        [`AsyncTable.checkpoint_lsm`][lancedb.AsyncTable.checkpoint_lsm]."""
+        return LOOP.run(self._table.checkpoint_lsm())
+
+    def flush_lsm(self) -> None:
+        """Synchronous version of
+        [`AsyncTable.flush_lsm`][lancedb.AsyncTable.flush_lsm]."""
+        return LOOP.run(self._table.flush_lsm())
+
+    def compact_lsm(self) -> None:
+        """Synchronous version of
+        [`AsyncTable.compact_lsm`][lancedb.AsyncTable.compact_lsm]."""
+        return LOOP.run(self._table.compact_lsm())
+
+    def get_lsm_stats(self, *, include_generation_rows: bool = False) -> Optional[dict]:
+        """Synchronous version of
+        [`AsyncTable.get_lsm_stats`][lancedb.AsyncTable.get_lsm_stats]."""
+        return LOOP.run(
+            self._table.get_lsm_stats(include_generation_rows=include_generation_rows)
+        )
+
     def close_lsm_writers(self) -> None:
         """Close cached MemWAL shard writers. See
         [`AsyncTable.close_lsm_writers`][lancedb.AsyncTable.close_lsm_writers]."""
@@ -4686,6 +4708,67 @@ class AsyncTable:
         """
         return await self._inner.get_lsm_write_spec()
 
+    async def checkpoint_lsm(self) -> None:
+        """Converge this table's LSM write path into its base table.
+
+        One flush, sealing every memtable into L0, then compaction triggers
+        until every generation that existed at that moment has reached base.
+        The loop runs client-side, reading progress from ``get_lsm_stats``.
+
+        Best-effort: generations created *while* it runs are deliberately not
+        waited on, which is what lets it terminate on a table taking writes.
+        Idempotent and safe on a cadence.
+
+        There is no deadline, and the caller owns that. It returns when the
+        target generations are gone, raises on a terminal server fault, and
+        otherwise waits however long the server takes. A slow table and a
+        stuck one are the same picture from the client: the compactor pool is
+        shared across every table on the node, so a checkpoint queued behind
+        unrelated work looks exactly like one that is merging. Wrap this in
+        ``asyncio.wait_for`` for a wall-clock bound; abandoning it partway
+        costs nothing.
+        """
+        return await self._inner.checkpoint_lsm()
+
+    async def flush_lsm(self) -> None:
+        """Seal every bucket's active memtable into L0.
+
+        Does not touch the base table — moving L0 into base is
+        `compact_lsm`. On a node that has not claimed this table, this claims
+        it and replays its WAL log first.
+        """
+        return await self._inner.flush_lsm()
+
+    async def compact_lsm(self) -> None:
+        """Trigger a background L0 to base compaction pass per bucket.
+
+        Returns once the passes are dispatched, not once they finish: watch
+        ``get_lsm_stats`` for progress, or use ``checkpoint_lsm`` to loop
+        until the current L0 has reached base.
+        """
+        return await self._inner.compact_lsm()
+
+    async def get_lsm_stats(
+        self, *, include_generation_rows: bool = False
+    ) -> Optional[dict]:
+        """Read live per-bucket LSM state.
+
+        Answers "how far behind is my fresh tier", "which bucket is hot", and
+        "why is my fresh-tier vector search brute-force". Mutates no table
+        state, though on a node that has not claimed this table it claims it,
+        exactly as a read would.
+
+        Returns ``None`` only when the LSM write path is not enabled.
+
+        Parameters
+        ----------
+        include_generation_rows
+            Report a row count per L0 generation. Off by default: each count
+            opens an uncached Lance dataset, and ``checkpoint_lsm`` polls this
+            needing only generation numbers.
+        """
+        return await self._inner.get_lsm_stats(include_generation_rows)
+
     async def close_lsm_writers(self) -> None:
         """Drain and close any cached MemWAL shard writers for this table.