test(python): skip pandas regression without pandas

test(python): cover float16 IVF search with open reader
fix: get table size from metadata, not files (#3790 )
2026-08-28 08:58:41 +00:00 · 2026-08-08 12:31:17 +00:00 · 2026-08-08 12:12:25 +00:00 · 2026-08-07 17:41:41 -04:00 · 2026-08-07 21:16:07 +00:00 · 2026-08-07 14:05:45 -07:00
53 changed files with 2805 additions and 200 deletions
@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "0.37.1-beta.0"
+current_version = "0.37.1-beta.1"
 parse = """(?x)
    (?P<major>0|[1-9]\\d*)\\.
    (?P<minor>0|[1-9]\\d*)\\.
@@ -0,0 +1,222 @@
+name: Check doc links
+
+# Checking external links is inherently noisy: third-party sites rate-limit
+# automated clients, reject non-browser user agents, and go down temporarily.
+# Blocking pull requests on that trades a lot of false failures for very little
+# signal, so this runs on a schedule and reports findings in a single tracking
+# issue instead of failing anyone's build.
+on:
+  schedule:
+    - cron: "0 7 * * *"
+  workflow_dispatch:
+
+# The report lives in one repository-global issue, so runs must not overlap: a
+# lookup racing a create produces duplicate issues, and a healthy run closing
+# the issue while a failing run only rewrites its body would leave a broken
+# report closed. The group is deliberately ref-independent so that a manual
+# dispatch serializes against the scheduled run.
+concurrency:
+  group: docs-link-check
+  cancel-in-progress: false
+
+permissions: {}
+
+env:
+  REPORT_TITLE: "Docs link checker report"
+
+jobs:
+  scan:
+    name: Scan links
+    runs-on: ubuntu-24.04
+    # lychee-action is pinned by SHA, but its wrapper downloads the lychee
+    # release tarball at run time without verifying a digest, and hands the
+    # resulting binary a GitHub token. Release assets remain replaceable, so
+    # that binary is confined to a job whose token can only read public
+    # content; everything that writes runs in the report job below.
+    permissions:
+      contents: read
+    outputs:
+      exit_code: ${{ steps.lychee.outputs.exit_code }}
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v6
+        with:
+          # workflow_dispatch can run from any ref, but the report is
+          # repository-global. Always measure the default branch so a manual
+          # run from a topic branch cannot close a report that main warrants,
+          # or overwrite it with branch-only findings.
+          ref: ${{ github.event.repository.default_branch }}
+          persist-credentials: false
+
+      - name: Check links
+        id: lychee
+        uses: lycheeverse/lychee-action@e7477775783ea5526144ba13e8db5eec57747ce8 # v2.9.0
+        with:
+          # Restricted to http(s) on purpose. Much of docs/src is generated
+          # API reference (the js/ tree comes from `npm run docs` in nodejs)
+          # and the hand-written pages use mkdocstrings cross-references and
+          # nav-relative paths that only resolve in the site mkdocs builds,
+          # not in this checkout, so relative links would be reported as
+          # broken on every run.
+          args: >-
+            --scheme https
+            --scheme http
+            --no-progress
+            --max-retries 3
+            --timeout 20
+            'docs/src/**/*.md'
+          format: json
+          output: ./lychee/out.json
+          jobSummary: false
+          # The report, not a red build, is the signal for broken links. The
+          # validation step below still fails the run if the check itself
+          # breaks.
+          fail: false
+
+      - name: Validate report
+        # lychee does not reserve exit code 2 for broken links: its CLI
+        # parser also exits 2 on an invalid option, before any link was
+        # checked or any report written. Only a parseable report whose
+        # counts agree with the exit code counts as a link verdict; anything
+        # else fails here, and the report job below is skipped entirely, so
+        # the tracking issue is never touched. Exit 2 covers timeouts as
+        # well as errors, and a timed-out host is exactly the transient
+        # unavailability this report exists to surface, so both count as
+        # findings. Requiring total > 0 also catches a glob that silently
+        # stopped matching any file.
+        if: steps.lychee.outputs.exit_code == 0 || steps.lychee.outputs.exit_code == 2
+        env:
+          EXIT_CODE: ${{ steps.lychee.outputs.exit_code }}
+        run: |
+          jq -e --argjson code "$EXIT_CODE" '
+            (.total > 0) and
+            (if $code == 0
+             then .errors == 0 and .timeouts == 0
+               and (.error_map | length == 0) and (.timeout_map | length == 0)
+             else (.errors + .timeouts) > 0
+               and ((.error_map | length) + (.timeout_map | length)) > 0
+             end)
+          ' ./lychee/out.json
+
+      - name: Upload report
+        if: steps.lychee.outputs.exit_code == 2
+        uses: actions/upload-artifact@v7
+        with:
+          name: link-report
+          path: ./lychee/out.json
+          retention-days: 7
+
+  report:
+    name: Update report issue
+    needs: scan
+    runs-on: ubuntu-24.04
+    # Deliberately no checkout: this job needs the report artifact and the
+    # issues API, not the repository contents.
+    permissions:
+      issues: write
+    env:
+      EXIT_CODE: ${{ needs.scan.outputs.exit_code }}
+      GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+    steps:
+      - name: Classify checker result
+        # lychee exits 0 when every link resolves and 2 when links fail,
+        # both already cross-checked against the report by the scan job's
+        # validation step. Anything else (1 runtime, 3 bad config) means the
+        # check never produced a link verdict, which must surface as a failed
+        # run rather than be published as "broken documentation links".
+        run: |
+          case "$EXIT_CODE" in
+            0|2)
+              echo "lychee exit code $EXIT_CODE"
+              ;;
+            *)
+              echo "::error::lychee exited with '$EXIT_CODE': the link check did not complete. Leaving the report issue untouched."
+              exit 1
+              ;;
+          esac
+
+      - name: Find existing report issue
+        id: report
+        # Matched on title alone, and through search rather than a listing:
+        # the issue action applies labels in a separate call after creating the
+        # issue, so a label filter misses a half-created report, and this
+        # repository has far more open issues than one listing page holds.
+        # Closed issues are included because a healthy run closes the report:
+        # an open-only lookup would forget that identity and the next failing
+        # run would open a duplicate. The oldest match stays the canonical
+        # report and is reopened below when links break again.
+        run: |
+          match=$(gh issue list --repo "$GITHUB_REPOSITORY" --state all \
+            --search "in:title \"$REPORT_TITLE\" author:app/github-actions" \
+            --limit 50 --json number,title,state \
+            --jq "[.[] | select(.title == \"$REPORT_TITLE\")] | sort_by(.number) | first // empty")
+          echo "number=$(jq -r '.number // empty' <<<"$match")" >> "$GITHUB_OUTPUT"
+          echo "state=$(jq -r '.state // empty' <<<"$match")" >> "$GITHUB_OUTPUT"
+
+      - name: Download report
+        if: env.EXIT_CODE == 2
+        uses: actions/download-artifact@v8
+        with:
+          name: link-report
+          path: ./lychee
+
+      - name: Compose report
+        if: env.EXIT_CODE == 2
+        run: |
+          run_url="$GITHUB_SERVER_URL/$GITHUB_REPOSITORY/actions/runs/$GITHUB_RUN_ID"
+          {
+            echo "Broken documentation links found by [\`$GITHUB_WORKFLOW\`]($run_url)."
+            echo
+            echo "This issue is rewritten by every scheduled run and closed automatically once all links resolve."
+            echo
+            echo "Entries can be false positives: some sites rate-limit or block automated clients while working fine in a browser. Confirm before editing the docs, and add persistent offenders to \`--exclude\` in \`.github/workflows/docs-link-check.yml\`."
+            echo
+            # Timeouts are reported alongside errors: entries land in
+            # timeout_map with a status text instead of an HTTP code.
+            jq -r '
+              "\(.errors) of \(.total) links failed, \(.timeouts) timed out.",
+              "",
+              ([(.error_map | to_entries[]), (.timeout_map | to_entries[])]
+               | group_by(.key)[] |
+                "### Errors in \(.[0].key)",
+                "",
+                (map(.value[])[] | "* [\(.status.code // .status.text // "ERR")] <\(.url)> — \(.status.details // .status.text // "unknown error")"),
+                "")
+            ' ./lychee/out.json
+          } > ./lychee/issue.md
+
+      - name: Reopen report issue
+        # A healthy run closes the report, and the issue action below only
+        # rewrites the body of whatever number it is given. Without an
+        # explicit reopen, the 2 -> 0 -> 2 sequence would keep rewriting a
+        # closed issue while links are broken. A CLOSED state implies the
+        # lookup found a canonical issue, so no separate emptiness check.
+        if: env.EXIT_CODE == 2 && steps.report.outputs.state == 'CLOSED'
+        env:
+          ISSUE_NUMBER: ${{ steps.report.outputs.number }}
+        run: |
+          run_url="$GITHUB_SERVER_URL/$GITHUB_REPOSITORY/actions/runs/$GITHUB_RUN_ID"
+          gh issue reopen "$ISSUE_NUMBER" --repo "$GITHUB_REPOSITORY" \
+            --comment "Broken documentation links found again in [the latest run]($run_url)."
+
+      - name: Report broken links
+        if: env.EXIT_CODE == 2
+        uses: peter-evans/create-issue-from-file@fca9117c27cdc29c6c4db3b86c48e4115a786710 # v6.0.0
+        with:
+          # Empty on the first failing run, which creates the issue; afterwards
+          # the same issue is updated in place.
+          issue-number: ${{ steps.report.outputs.number }}
+          title: ${{ env.REPORT_TITLE }}
+          content-filepath: ./lychee/issue.md
+          labels: documentation
+
+      - name: Close report issue once links are healthy
+        # An OPEN state implies the lookup found a canonical issue; a report
+        # that is already closed needs nothing.
+        if: env.EXIT_CODE == 0 && steps.report.outputs.state == 'OPEN'
+        env:
+          ISSUE_NUMBER: ${{ steps.report.outputs.number }}
+        run: |
+          run_url="$GITHUB_SERVER_URL/$GITHUB_REPOSITORY/actions/runs/$GITHUB_RUN_ID"
+          gh issue close "$ISSUE_NUMBER" --repo "$GITHUB_REPOSITORY" \
+            --comment "All documentation links resolved in [the latest run]($run_url)."
@@ -3455,8 +3455,8 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"

 [[package]]
 name = "fsst"
-version = "11.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow-array",
 "rand 0.9.5",
@@ -4815,8 +4815,8 @@ checksum = "e037a2e1d8d5fdbd49b16a4ea09d5d6401c1f29eca5ff29d03d3824dba16256a"

 [[package]]
 name = "lance"
-version = "11.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arc-swap",
 "arrow",
@@ -4890,8 +4890,8 @@ dependencies = [

 [[package]]
 name = "lance-arrow"
-version = "11.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4913,7 +4913,7 @@ dependencies = [
 [[package]]
 name = "lance-arrow-scalar"
 version = "58.0.0"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4927,7 +4927,7 @@ dependencies = [
 [[package]]
 name = "lance-arrow-stats"
 version = "58.0.0"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -4936,8 +4936,8 @@ dependencies = [

 [[package]]
 name = "lance-bitpacking"
-version = "11.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrayref",
 "crunchy",
@@ -4947,8 +4947,8 @@ dependencies = [

 [[package]]
 name = "lance-core"
-version = "11.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4988,8 +4988,8 @@ dependencies = [

 [[package]]
 name = "lance-datafusion"
-version = "11.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5019,8 +5019,8 @@ dependencies = [

 [[package]]
 name = "lance-datagen"
-version = "11.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5037,8 +5037,8 @@ dependencies = [

 [[package]]
 name = "lance-derive"
-version = "11.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "proc-macro2",
 "quote",
@@ -5047,8 +5047,8 @@ dependencies = [

 [[package]]
 name = "lance-encoding"
-version = "11.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow-arith",
 "arrow-array",
@@ -5082,8 +5082,8 @@ dependencies = [

 [[package]]
 name = "lance-file"
-version = "11.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow-arith",
 "arrow-array",
@@ -5114,8 +5114,8 @@ dependencies = [

 [[package]]
 name = "lance-index"
-version = "11.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arc-swap",
 "arrow",
@@ -5182,8 +5182,8 @@ dependencies = [

 [[package]]
 name = "lance-index-core"
-version = "11.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -5205,8 +5205,8 @@ dependencies = [

 [[package]]
 name = "lance-io"
-version = "11.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5242,8 +5242,8 @@ dependencies = [

 [[package]]
 name = "lance-linalg"
-version = "11.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -5259,8 +5259,8 @@ dependencies = [

 [[package]]
 name = "lance-namespace"
-version = "11.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow",
 "async-trait",
@@ -5272,8 +5272,8 @@ dependencies = [

 [[package]]
 name = "lance-namespace-impls"
-version = "11.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow",
 "arrow-ipc",
@@ -5303,7 +5303,6 @@ dependencies = [
 "serde",
 "serde_json",
 "sha2 0.10.9",
- "time",
 "tokio",
 "tower",
 "tower-http 0.5.2",
@@ -5327,8 +5326,8 @@ dependencies = [

 [[package]]
 name = "lance-select"
-version = "11.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -5343,8 +5342,8 @@ dependencies = [

 [[package]]
 name = "lance-table"
-version = "11.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5354,6 +5353,7 @@ dependencies = [
 "async-trait",
 "aws-credential-types",
 "aws-sdk-dynamodb",
+ "blake3",
 "byteorder",
 "bytes",
 "chrono",
@@ -5383,8 +5383,8 @@ dependencies = [

 [[package]]
 name = "lance-testing"
-version = "11.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -5397,8 +5397,8 @@ dependencies = [

 [[package]]
 name = "lance-tokenizer"
-version = "11.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "frostem",
 "icu_segmenter",
@@ -5411,7 +5411,7 @@ dependencies = [

 [[package]]
 name = "lancedb"
-version = "0.37.1-beta.0"
+version = "0.37.1-beta.1"
 dependencies = [
 "ahash",
 "anyhow",
@@ -5480,6 +5480,7 @@ dependencies = [
 "random_word",
 "regex",
 "reqwest 0.12.28",
+ "roaring",
 "rstest",
 "semver",
 "serde",
@@ -5499,7 +5500,7 @@ dependencies = [

 [[package]]
 name = "lancedb-nodejs"
-version = "0.37.1-beta.0"
+version = "0.37.1-beta.1"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -5524,7 +5525,7 @@ dependencies = [

 [[package]]
 name = "lancedb-python"
-version = "0.37.1-beta.0"
+version = "0.37.1-beta.1"
 dependencies = [
 "arrow",
 "async-trait",
@@ -13,20 +13,20 @@ categories = ["database-implementations"]
 rust-version = "1.91.0"

 [workspace.dependencies]
-lance = { "version" = "=11.0.0-beta.2", default-features = false, "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-core = { "version" = "=11.0.0-beta.2", "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-datagen = { "version" = "=11.0.0-beta.2", "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-file = { "version" = "=11.0.0-beta.2", "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-io = { "version" = "=11.0.0-beta.2", default-features = false, "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-index = { "version" = "=11.0.0-beta.2", "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-linalg = { "version" = "=11.0.0-beta.2", "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace = { "version" = "=11.0.0-beta.2", "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace-impls = { "version" = "=11.0.0-beta.2", default-features = false, "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-table = { "version" = "=11.0.0-beta.2", "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-testing = { "version" = "=11.0.0-beta.2", "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-datafusion = { "version" = "=11.0.0-beta.2", "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-encoding = { "version" = "=11.0.0-beta.2", "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-arrow = { "version" = "=11.0.0-beta.2", "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
+lance = { "version" = "=11.0.0-beta.3", default-features = false, "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
+lance-core = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
+lance-datagen = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
+lance-file = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
+lance-io = { "version" = "=11.0.0-beta.3", default-features = false, "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
+lance-index = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
+lance-linalg = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace-impls = { "version" = "=11.0.0-beta.3", default-features = false, "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
+lance-table = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
+lance-testing = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
+lance-datafusion = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
+lance-encoding = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
+lance-arrow = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
 ahash = "0.8"
 # Note that this one does not include pyarrow
 arrow = { version = "58.0.0", optional = false }
@@ -52,7 +52,7 @@ env_logger = "0.11"
 half = { "version" = "2.7.1", default-features = false, features = [
    "num-traits",
 ] }
-futures = "0"
+futures = "0.3"
 log = "0.4"
 metrics = "0.24"
 metrics-util = "0.19"
@@ -14,7 +14,7 @@ Add the following dependency to your `pom.xml`:
 <dependency>
    <groupId>com.lancedb</groupId>
    <artifactId>lancedb-core</artifactId>
-    <version>0.37.1-beta.0</version>
+    <version>0.37.1-beta.1</version>
 </dependency>
 ```

@@ -431,9 +431,10 @@ Read the [LsmWriteSpec](../interfaces/LsmWriteSpec.md) currently installed on th

 Resolves to `undefined` when the MemWAL LSM write path is not enabled (no
 spec has been set, or it was removed with [Table#unsetLsmWriteSpec](Table.md#unsetlsmwritespec)).
-The returned spec — including its `maintainedIndexes` and
-`writerConfigDefaults` — mirrors what was passed to
-[Table#setLsmWriteSpec](Table.md#setlsmwritespec).
+The returned spec mirrors what was passed to
+[Table#setLsmWriteSpec](Table.md#setlsmwritespec), except that `maintainedIndexes` always
+reports the concrete list resolved when the spec was set — `undefined`
+never round-trips.

 #### Returns

@@ -806,6 +807,11 @@ All variants require the table to have an unenforced primary key
 ([Table#setUnenforcedPrimaryKey](Table.md#setunenforcedprimarykey)); bucket sharding additionally
 requires it to be the single column being bucketed.

+Omitting `maintainedIndexes` maintains every index on the table, resolved
+here, failing if one cannot be maintained — name them to install anyway.
+Naming them pins an exact set, and a still-building index is rejected
+rather than quietly omitted.
+
 #### Parameters

 * **spec**: [`LsmWriteSpec`](../interfaces/LsmWriteSpec.md)
@@ -34,7 +34,9 @@ Bucket and identity variants: the sharding column.
 optional maintainedIndexes: string[];
 ```

-Names of indexes the MemWAL should keep up to date during writes.
+Indexes the MemWAL keeps up to date. Omit to maintain every supported
+index, resolved on install — a snapshot, so indexes created later are not
+maintained. Pass `[]` for none.

 ***

@@ -44,4 +44,7 @@ The number of rows in the table
 totalBytes: number;
 ```

-The total number of bytes in the table
+The total size, in bytes, of the table's data files, index files, and
+overlay files
+
+Read from the manifest, so this excludes deletion files and manifests.
@@ -31,7 +31,7 @@ is also an [asynchronous API client](#connections-asynchronous).
 ## Namespaces (Synchronous)

 A namespace-backed connection resolves tables through a
-[Lance namespace](https://lancedb.github.io/lance-namespace/) service instead of
+[Lance namespace](https://lance-format.github.io/lance-namespace/) service instead of
 listing a storage directory.

 ::: lancedb.connect_namespace
@@ -8,7 +8,7 @@
    <parent>
      <groupId>com.lancedb</groupId>
      <artifactId>lancedb-parent</artifactId>
-      <version>0.37.1-beta.0</version>
+      <version>0.37.1-beta.1</version>
      <relativePath>../pom.xml</relativePath>
    </parent>

@@ -6,7 +6,7 @@

    <groupId>com.lancedb</groupId>
    <artifactId>lancedb-parent</artifactId>
-    <version>0.37.1-beta.0</version>
+    <version>0.37.1-beta.1</version>
    <packaging>pom</packaging>
    <name>${project.artifactId}</name>
    <description>LanceDB Java SDK Parent POM</description>
@@ -28,7 +28,7 @@
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <arrow.version>15.0.0</arrow.version>
-        <lance-core.version>11.0.0-beta.2</lance-core.version>
+        <lance-core.version>11.0.0-beta.3</lance-core.version>
        <spotless.skip>false</spotless.skip>
        <spotless.version>2.30.0</spotless.version>
        <spotless.java.googlejavaformat.version>1.7</spotless.java.googlejavaformat.version>
@@ -1,7 +1,7 @@
 [package]
 name = "lancedb-nodejs"
 edition.workspace = true
-version = "0.37.1-beta.0"
+version = "0.37.1-beta.1"
 publish = false
 license.workspace = true
 description.workspace = true
@@ -11,8 +11,11 @@ import {
  Float16,
  Float32,
  Float64,
+  Int32,
  Schema,
  Utf8,
+  fromDataToBuffer,
+  tableFromIPC,
 } from "../lancedb/arrow";
 import { EmbeddingFunction, LanceSchema } from "../lancedb/embedding";
 import { getRegistry, register } from "../lancedb/embedding/registry";
@@ -184,6 +187,63 @@ describe("embedding functions", () => {
    const vector0 = JSON.parse(JSON.stringify(arr[0].vector));
    expect(vector0).toEqual([1, 2, 3]);
  });
+
+  it("should append generated vectors to a non-nullable schema", async () => {
+    @register("non_nullable_schema_test")
+    class MockEmbeddingFunction extends EmbeddingFunction<string> {
+      ndims() {
+        return 3;
+      }
+      embeddingDataType(): Float {
+        return new Float64();
+      }
+      async computeSourceEmbeddings(data: string[]) {
+        return data.map(() => [1, 2, 3]);
+      }
+    }
+
+    const schema = new Schema([
+      new Field("id", new Int32()),
+      new Field("text", new Utf8()),
+      new Field("type", new Utf8()),
+      new Field(
+        "vector",
+        new FixedSizeList(3, new Field("item", new Float64())),
+      ),
+    ]);
+    const func = new MockEmbeddingFunction();
+    const db = await connect(tmpDir.name);
+    const table = await db.createEmptyTable("test_non_nullable", schema, {
+      embeddingFunction: {
+        function: func,
+        sourceColumn: "text",
+      },
+    });
+
+    const data = [
+      { id: 1, text: "Carrot", type: "vegetable" },
+      { id: 2, text: "Apple", type: "fruit" },
+    ];
+    const buffer = await fromDataToBuffer(
+      data,
+      undefined,
+      await table.schema(),
+    );
+    const generatedTable = tableFromIPC(buffer);
+    const vectorField = generatedTable.schema.fields.find(
+      (field) => field.name === "vector",
+    );
+    expect(vectorField?.nullable).toBe(false);
+
+    await table.add(data);
+
+    const rows = await table.query().toArray();
+    expect(rows).toHaveLength(2);
+    for (const row of rows) {
+      expect([...row.vector]).toEqual([1, 2, 3]);
+    }
+  });
+
  it("should error when appending to a table with an unregistered embedding function", async () => {
    @register("mock")
    class MockEmbeddingFunction extends EmbeddingFunction<string> {
@@ -0,0 +1,14 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+import packageJson = require("../package.json");
+
+describe("package metadata", () => {
+  it("requires Node.js type declarations compatible with the runtime", () => {
+    expect(packageJson.engines.node).toBe(">= 18");
+    expect(packageJson.peerDependencies["@types/node"]).toBe(">=18");
+    expect(packageJson.peerDependenciesMeta["@types/node"]).toEqual({
+      optional: true,
+    });
+  });
+});
@@ -110,6 +110,81 @@ describe("Query outputSchema", () => {
  });
 });

+describe("Search pagination", () => {
+  let tmpDir: tmp.DirResult;
+  let table: Table;
+
+  beforeEach(async () => {
+    tmpDir = tmp.dirSync({ unsafeCleanup: true });
+    const db = await connect(tmpDir.name);
+    const schema = new Schema([
+      new Field("id", new Int64(), false),
+      new Field("text", new Utf8(), false),
+      new Field(
+        "vector",
+        new FixedSizeList(2, new Field("item", new Float32())),
+        false,
+      ),
+    ]);
+    const data = makeArrowTable(
+      [
+        { id: 1n, text: "common", vector: [0, 0] },
+        { id: 2n, text: "common common", vector: [1, 1] },
+        { id: 3n, text: "common common common", vector: [2, 2] },
+        { id: 4n, text: "common common common common", vector: [3, 3] },
+      ],
+      { schema },
+    );
+    table = await db.createTable("test", data);
+  });
+
+  afterEach(() => {
+    tmpDir.removeCallback();
+  });
+
+  it("applies offset after the vector search limit", async () => {
+    const allResults = await table
+      .vectorSearch([0, 0])
+      .select(["id"])
+      .limit(4)
+      .toArray();
+    const secondPage = await table
+      .vectorSearch([0, 0])
+      .select(["id"])
+      .limit(2)
+      .offset(2)
+      .toArray();
+
+    expect(allResults).toHaveLength(4);
+    expect(secondPage).toHaveLength(2);
+    expect(secondPage.map((row) => row.id)).toEqual(
+      allResults.slice(2, 4).map((row) => row.id),
+    );
+  });
+
+  it("applies offset after the full-text search limit", async () => {
+    await table.createIndex("text", { config: Index.fts() });
+
+    const allResults = await table
+      .search("common", "fts")
+      .select(["id"])
+      .limit(4)
+      .toArray();
+    const secondPage = await table
+      .search("common", "fts")
+      .select(["id"])
+      .limit(2)
+      .offset(2)
+      .toArray();
+
+    expect(allResults).toHaveLength(4);
+    expect(secondPage).toHaveLength(2);
+    expect(secondPage.map((row) => row.id)).toEqual(
+      allResults.slice(2, 4).map((row) => row.id),
+    );
+  });
+});
+
 describe("Query orderBy", () => {
  let tmpDir: tmp.DirResult;
  let table: Table;
@@ -277,8 +277,16 @@ describe.each([arrow15, arrow16, arrow17, arrow18])(
        },
        numIndices: 0,
        numRows: 3,
-        totalBytes: 44,
+        // Full on-disk size of the two data files, footers and metadata included.
+        totalBytes: 684,
      });
+
+      // Index files count toward totalBytes too (only deletion files and
+      // manifests are excluded).
+      await table.createIndex("id", { config: Index.btree() });
+      const statsWithIndex = await table.stats();
+      expect(statsWithIndex.numIndices).toBe(1);
+      expect(statsWithIndex.totalBytes).toBeGreaterThan(684);
    });

    it("should overwrite data if asked", async () => {
@@ -197,7 +197,11 @@ export interface LsmWriteSpec {
  column?: string;
  /** Bucket variant: the number of buckets, in `[1, 1024]`. */
  numBuckets?: number;
-  /** Names of indexes the MemWAL should keep up to date during writes. */
+  /**
+   * Indexes the MemWAL keeps up to date. Omit to maintain every supported
+   * index, resolved on install — a snapshot, so indexes created later are not
+   * maintained. Pass `[]` for none.
+   */
  maintainedIndexes?: string[];
  /** Default `ShardWriter` configuration recorded in the MemWAL index. */
  writerConfigDefaults?: Record<string, string>;
@@ -595,6 +599,11 @@ export abstract class Table {
   * All variants require the table to have an unenforced primary key
   * ({@link Table#setUnenforcedPrimaryKey}); bucket sharding additionally
   * requires it to be the single column being bucketed.
+   *
+   * Omitting `maintainedIndexes` maintains every index on the table, resolved
+   * here, failing if one cannot be maintained — name them to install anyway.
+   * Naming them pins an exact set, and a still-building index is rejected
+   * rather than quietly omitted.
   * @param {LsmWriteSpec} spec The sharding spec to install.
   * @returns {Promise<void>}
   * @example
@@ -622,9 +631,10 @@ export abstract class Table {
   *
   * Resolves to `undefined` when the MemWAL LSM write path is not enabled (no
   * spec has been set, or it was removed with {@link Table#unsetLsmWriteSpec}).
-   * The returned spec — including its `maintainedIndexes` and
-   * `writerConfigDefaults` — mirrors what was passed to
-   * {@link Table#setLsmWriteSpec}.
+   * The returned spec mirrors what was passed to
+   * {@link Table#setLsmWriteSpec}, except that `maintainedIndexes` always
+   * reports the concrete list resolved when the spec was set — `undefined`
+   * never round-trips.
   * @returns {Promise<LsmWriteSpec | undefined>}
   */
  abstract getLsmWriteSpec(): Promise<LsmWriteSpec | undefined>;
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-darwin-arm64",
-	"version": "0.37.1-beta.0",
+	"version": "0.37.1-beta.1",
 	"os": ["darwin"],
 	"cpu": ["arm64"],
 	"main": "lancedb.darwin-arm64.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-gnu",
-	"version": "0.37.1-beta.0",
+	"version": "0.37.1-beta.1",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-gnu.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-musl",
-	"version": "0.37.1-beta.0",
+	"version": "0.37.1-beta.1",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-musl.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-gnu",
-	"version": "0.37.1-beta.0",
+	"version": "0.37.1-beta.1",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-gnu.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-musl",
-	"version": "0.37.1-beta.0",
+	"version": "0.37.1-beta.1",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-musl.node",
@@ -1,6 +1,6 @@
 {
  "name": "@lancedb/lancedb-win32-arm64-msvc",
-  "version": "0.37.1-beta.0",
+  "version": "0.37.1-beta.1",
  "os": [
    "win32"
  ],
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-win32-x64-msvc",
-	"version": "0.37.1-beta.0",
+	"version": "0.37.1-beta.1",
 	"os": ["win32"],
 	"cpu": ["x64"],
 	"main": "lancedb.win32-x64-msvc.node",
@@ -1,12 +1,12 @@
 {
  "name": "@lancedb/lancedb",
-  "version": "0.37.1-beta.0",
+  "version": "0.37.1-beta.1",
  "lockfileVersion": 3,
  "requires": true,
  "packages": {
    "": {
      "name": "@lancedb/lancedb",
-      "version": "0.37.1-beta.0",
+      "version": "0.37.1-beta.1",
      "cpu": [
        "x64",
        "arm64"
@@ -55,7 +55,13 @@
        "openai": "4.29.2"
      },
      "peerDependencies": {
+        "@types/node": ">=18",
        "apache-arrow": ">=15.0.0 <=18.1.0"
+      },
+      "peerDependenciesMeta": {
+        "@types/node": {
+          "optional": true
+        }
      }
    },
    "node_modules/@aws-crypto/crc32": {
@@ -11,7 +11,7 @@
    "ann"
  ],
  "private": false,
-  "version": "0.37.1-beta.0",
+  "version": "0.37.1-beta.1",
  "main": "dist/index.js",
  "exports": {
    ".": "./dist/index.js",
@@ -101,6 +101,12 @@
    "openai": "4.29.2"
  },
  "peerDependencies": {
+    "@types/node": ">=18",
    "apache-arrow": ">=15.0.0 <=18.1.0"
+  },
+  "peerDependenciesMeta": {
+    "@types/node": {
+      "optional": true
+    }
  }
 }
@@ -772,7 +772,8 @@ pub struct LsmWriteSpec {
    pub column: Option<String>,
    /// Bucket variant: the number of buckets, in `[1, 1024]`.
    pub num_buckets: Option<u32>,
-    /// Names of indexes the MemWAL should keep up to date during writes.
+    /// Indexes the MemWAL keeps up to date. Omitted resolves every
+    /// maintainable index on install; an empty array means none.
    pub maintained_indexes: Option<Vec<String>>,
    /// Default `ShardWriter` configuration recorded in the MemWAL index.
    pub writer_config_defaults: Option<HashMap<String, String>>,
@@ -782,7 +783,6 @@ impl TryFrom<LsmWriteSpec> for lancedb::table::LsmWriteSpec {
    type Error = napi::Error;

    fn try_from(value: LsmWriteSpec) -> napi::Result<Self> {
-        let maintained = value.maintained_indexes.unwrap_or_default();
        let writer_config_defaults = value.writer_config_defaults.unwrap_or_default();
        let spec = match value.spec_type.as_str() {
            "bucket" => {
@@ -809,7 +809,7 @@ impl TryFrom<LsmWriteSpec> for lancedb::table::LsmWriteSpec {
            }
        };
        Ok(spec
-            .with_maintained_indexes(maintained)
+            .with_maintained_indexes(value.maintained_indexes)
            .with_writer_config_defaults(writer_config_defaults))
    }
 }
@@ -827,7 +827,7 @@ impl From<lancedb::table::LsmWriteSpec> for LsmWriteSpec {
                spec_type: "bucket".to_string(),
                column: Some(column),
                num_buckets: Some(num_buckets),
-                maintained_indexes: Some(maintained_indexes),
+                maintained_indexes,
                writer_config_defaults: Some(writer_config_defaults),
            },
            Native::Identity {
@@ -838,7 +838,7 @@ impl From<lancedb::table::LsmWriteSpec> for LsmWriteSpec {
                spec_type: "identity".to_string(),
                column: Some(column),
                num_buckets: None,
-                maintained_indexes: Some(maintained_indexes),
+                maintained_indexes,
                writer_config_defaults: Some(writer_config_defaults),
            },
            Native::Unsharded {
@@ -848,7 +848,7 @@ impl From<lancedb::table::LsmWriteSpec> for LsmWriteSpec {
                spec_type: "unsharded".to_string(),
                column: None,
                num_buckets: None,
-                maintained_indexes: Some(maintained_indexes),
+                maintained_indexes,
                writer_config_defaults: Some(writer_config_defaults),
            },
        }
@@ -1043,7 +1043,10 @@ impl From<lancedb::index::IndexStatistics> for IndexStatistics {

 #[napi(object)]
 pub struct TableStatistics {
-    /// The total number of bytes in the table
+    /// The total size, in bytes, of the table's data files, index files, and
+    /// overlay files
+    ///
+    /// Read from the manifest, so this excludes deletion files and manifests.
    pub total_bytes: i64,

    /// The number of rows in the table
@@ -1,6 +1,6 @@
 [package]
 name = "lancedb-python"
-version = "0.37.1-beta.0"
+version = "0.37.1-beta.1"
 publish = false
 edition.workspace = true
 description = "Python bindings for LanceDB"
@@ -60,7 +60,7 @@ tests = [
    "pytest-asyncio>=0.21",
    "duckdb>=0.9.0",
    "pytz>=2023.3",
-    "polars>=0.19, <=1.3.0",
+    "polars>=0.19, <=1.32.3",
    "pyarrow<25",
    "pyarrow-stubs>=16.0",
    "pylance==9.0.0rc1",
@@ -140,6 +140,7 @@ include = [
    "python/lancedb/remote/errors.py",
    "python/lancedb/embeddings/__init__.py",
    "python/lancedb/_lancedb.pyi",
+    "python/type_tests/connect.py",
 ]
 exclude = ["python/tests/"]
 pythonVersion = "3.13"
@@ -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: ...
@@ -649,9 +653,10 @@ class LsmWriteSpec:
    def identity(column: str) -> "LsmWriteSpec": ...
    @staticmethod
    def unsharded() -> "LsmWriteSpec": ...
-    def with_maintained_indexes(self, indexes: List[str]) -> "LsmWriteSpec":
-        """Return a copy of this spec asking the MemWAL to keep the named
-        indexes up to date as rows are appended."""
+    def with_maintained_indexes(self, indexes: Optional[List[str]]) -> "LsmWriteSpec":
+        """Set which indexes the MemWAL keeps up to date. None resolves every
+        index on the table at install, failing if one cannot be maintained;
+        a list is verbatim, empty means none."""
        ...
    def with_writer_config_defaults(self, defaults: Dict[str, str]) -> "LsmWriteSpec":
        """Return a copy of this spec recording the given default
@@ -666,7 +671,9 @@ class LsmWriteSpec:
    @property
    def num_buckets(self) -> Optional[int]: ...
    @property
-    def maintained_indexes(self) -> List[str]: ...
+    def maintained_indexes(self) -> Optional[List[str]]:
+        """Indexes the MemWAL keeps up to date, or None for every supported one."""
+        ...
    @property
    def writer_config_defaults(self) -> Dict[str, str]: ...

@@ -87,12 +87,13 @@ class JinaEmbeddings(EmbeddingFunction):
        if isinstance(image, bytes):
            image_dict = {"image": base64.b64encode(image).decode("utf-8")}
        elif isinstance(image, (str, Path)):
-            parsed = urlparse.urlparse(image)
-            # TODO handle drive letter on windows.
+            parsed = urlparse(str(image))
            PIL_Image = attempt_import_or_raise("PIL.Image", "pillow")
            if parsed.scheme == "file":
                pil_image = PIL_Image.open(parsed.path)
-            elif parsed.scheme == "":
+            elif parsed.scheme == "" or (os.name == "nt" and len(parsed.scheme) == 1):
+                # A Windows drive letter parses as a one-character scheme
+                # ("C:\\img.png" -> scheme="c"), so treat it as a local path.
                pil_image = PIL_Image.open(image if os.name == "nt" else parsed.path)
            elif parsed.scheme.startswith("http"):
                pil_image = PIL_Image.open(io.BytesIO(url_retrieve(image)))
@@ -0,0 +1 @@
+
@@ -153,6 +153,16 @@ def Vector(
    return FixedSizeList


+def _raise_bare_vector_error(*_args):
+    raise TypeError("Vector must be parameterized with a dimension, e.g. Vector(128).")
+
+
+# Pydantic v1 and v2 otherwise treat the bare Vector factory as a field validator
+# and inspect its signature, which produces misleading errors about internal types.
+setattr(Vector, "__get_validators__", _raise_bare_vector_error)
+setattr(Vector, "__get_pydantic_core_schema__", _raise_bare_vector_error)
+
+
 def MultiVector(
    dim: int, value_type: pa.DataType = pa.float32(), nullable: bool = True
 ) -> Type:
@@ -108,6 +108,11 @@ def _should_push_down_query_table(
    return namespace_client is not None and "QueryTable" in pushdown_operations


+def _polars_predicate_pushdown_barrier(frame: Any) -> Any:
+    """Return a Polars frame unchanged while blocking predicate pushdown."""
+    return frame
+
+
 _MODEL_BACKED_TOKENIZER_PREFIXES = ("jieba", "lindera")
 _MODEL_BACKED_TOKENIZER_ERRORS = (
    "unknown base tokenizer",
@@ -864,12 +869,18 @@ class Table(ABC):
        """
        raise NotImplementedError

-    def to_polars(self, **kwargs) -> "pl.DataFrame":
-        """Return the table as a polars.DataFrame.
+    def to_polars(self, **kwargs) -> "pl.LazyFrame":
+        """Return the table as a Polars LazyFrame.
+
+        Note
+        ----
+        The Polars streaming engine is not supported because it does not currently
+        implement Python PyArrow dataset scans. Use the default engine when collecting
+        this LazyFrame.

        Returns
        -------
-        polars.DataFrame
+        polars.LazyFrame
        """
        raise NotImplementedError

@@ -2569,6 +2580,9 @@ class LanceTable(Table):
        2. Currently we've disabled push-down of the filters from polars
           because polars pushdown into pyarrow uses pyarrow compute
           expressions rather than SQl strings (which LanceDB supports)
+        3. The Polars streaming engine is not supported because it does not
+           currently implement Python PyArrow dataset scans. Use the default
+           engine when collecting this LazyFrame.

        Returns
        -------
@@ -2577,8 +2591,12 @@ class LanceTable(Table):
        from lancedb.integrations.pyarrow import PyarrowDatasetAdapter

        dataset = PyarrowDatasetAdapter(self)
-        return pl.scan_pyarrow_dataset(
-            dataset, allow_pyarrow_filter=False, batch_size=batch_size
+        # Polars 1.32's non-PyArrow callback path passes batch_size twice.  Keep
+        # the compatible PyArrow path, but block predicates because this adapter
+        # cannot translate PyArrow expressions into LanceDB filters.
+        return pl.scan_pyarrow_dataset(dataset, batch_size=batch_size).map_batches(
+            _polars_predicate_pushdown_barrier,
+            predicate_pushdown=False,
        )

    # New unified API overload
@@ -3958,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]."""
@@ -4636,6 +4676,13 @@ class AsyncTable:
        via [`set_unenforced_primary_key`]; bucket sharding additionally
        requires it to be the single column being bucketed.

+        By default the MemWAL maintains every index on the table, resolved
+        here — a snapshot, so an index created afterwards needs the spec unset
+        and set again. This fails if one cannot be maintained; name the set
+        with ``with_maintained_indexes`` to install anyway. That pins an exact
+        set (a still-building index is rejected, not omitted); ``[]`` maintains
+        none.
+
        Parameters
        ----------
        spec : LsmWriteSpec
@@ -4662,12 +4709,73 @@ class AsyncTable:

        Returns ``None`` when the MemWAL LSM write path is not enabled (no
        spec has been set, or it was removed with `unset_lsm_write_spec`).
-        The returned spec — including its ``maintained_indexes`` and
-        ``writer_config_defaults`` — mirrors what was passed to
-        `set_lsm_write_spec`.
+        The returned spec mirrors what was passed to `set_lsm_write_spec`,
+        except that ``maintained_indexes`` always reports the concrete list
+        resolved when the spec was set — ``None`` never round-trips.
        """
        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.

@@ -6233,7 +6341,9 @@ class TableStatistics:
    Attributes
    ----------
    total_bytes: int
-        The total number of bytes in the table.
+        The total size, in bytes, of the table's data files, index files, and
+        overlay files. Read from the manifest, so this excludes deletion files
+        and manifests.
    num_rows: int
        The total number of rows in the table.
    num_indices: int
@@ -6,6 +6,7 @@ import inspect
 import re
 import sys
 from datetime import timedelta
+from importlib import resources
 import os
 from types import SimpleNamespace

@@ -18,6 +19,10 @@ from lance_namespace.errors import NamespaceNotEmptyError, TableNotFoundError
 from lancedb.pydantic import LanceModel, Vector


+def test_package_includes_pep_561_marker():
+    assert resources.files(lancedb).joinpath("py.typed").is_file()
+
+
 def test_basic(tmp_path):
    db = lancedb.connect(tmp_path)

@@ -631,3 +631,23 @@ def test_url_retrieve_downloads_image():
    image_bytes = url_retrieve(image_url)
    img = Image.open(io.BytesIO(image_bytes))
    assert img.size[0] > 0 and img.size[1] > 0
+
+
+def test_jina_generate_image_input_dict_local_path(tmp_path):
+    """
+    JinaEmbeddings._generate_image_input_dict must accept a local image path
+    (str or Path), not just bytes. Previously it crashed with
+    `AttributeError: 'function' object has no attribute 'urlparse'` on any
+    str/Path input because it called `urlparse.urlparse(image)` instead of
+    `urlparse(image)` (urlparse was imported as a function, not a module).
+    """
+    Image = pytest.importorskip("PIL.Image")
+    from lancedb.embeddings.jinaai import JinaEmbeddings
+
+    image_path = tmp_path / "test.png"
+    Image.new("RGB", (4, 4), color="red").save(image_path, format="PNG")
+
+    for image in (str(image_path), image_path):
+        image_dict = JinaEmbeddings._generate_image_input_dict(image)
+        assert "image" in image_dict
+        assert isinstance(image_dict["image"], str) and len(image_dict["image"]) > 0
@@ -83,7 +83,9 @@ def test_lsm_write_spec_repr():
    assert s.spec_type == "bucket"
    assert s.column == "id"
    assert s.num_buckets == 4
-    assert s.maintained_indexes == []
+    # A fresh spec defers its maintained set to install time.
+    assert s.maintained_indexes is None
+    assert s.with_maintained_indexes([]).maintained_indexes == []
    assert "bucket" in repr(s)
    assert "id" in repr(s)
    assert "4" in repr(s)
@@ -169,18 +171,23 @@ def test_get_lsm_write_spec(tmp_path):
    table.unset_lsm_write_spec()
    assert table.get_lsm_write_spec() is None

-    # Identity round-trips (column recovered from the schema).
+    # Identity round-trips (column recovered from the schema). Leaving the
+    # maintained set to be inferred picks up the index on the table, so the
+    # spec reads back naming it rather than as "infer".
    table.set_lsm_write_spec(LsmWriteSpec.identity("id"))
    spec = table.get_lsm_write_spec()
    assert spec.spec_type == "identity"
    assert spec.column == "id"
+    assert spec.maintained_indexes == [idx_name]
    table.unset_lsm_write_spec()

-    # Unsharded round-trips (no routing column).
-    table.set_lsm_write_spec(LsmWriteSpec.unsharded())
+    # Unsharded round-trips (no routing column). Opting out is distinct from
+    # the inferred default.
+    table.set_lsm_write_spec(LsmWriteSpec.unsharded().with_maintained_indexes([]))
    spec = table.get_lsm_write_spec()
    assert spec.spec_type == "unsharded"
    assert spec.column is None
+    assert spec.maintained_indexes == []


@pytest.mark.asyncio
@@ -544,7 +544,7 @@ def test_lsm_read_fts_unmaintained_index_errors(tmp_path):
    table.create_index("text", config=FTS())
    # No maintained indexes: the active memtable FTS arm cannot serve un-compacted
    # docs, so the search would silently omit them — reject instead.
-    table.set_lsm_write_spec(LsmWriteSpec.unsharded())
+    table.set_lsm_write_spec(LsmWriteSpec.unsharded().with_maintained_indexes([]))
    with pytest.raises(Exception, match="maintained"):
        table.search("fox", query_type="fts", fts_columns="text").to_arrow()

@@ -631,7 +631,7 @@ def test_lsm_read_vector_unmaintained_index_errors(tmp_path):
    )
    # Spec with NO maintained indexes: the base vector index's catch-up is untracked,
    # so the scanner rejects rather than risk dropping compacted-but-unindexed rows.
-    table.set_lsm_write_spec(LsmWriteSpec.unsharded())
+    table.set_lsm_write_spec(LsmWriteSpec.unsharded().with_maintained_indexes([]))
    with pytest.raises(Exception, match="maintained"):
        table.search([1.0] * VECTOR_DIM).to_arrow()

@@ -415,6 +415,17 @@ def test_nullable_vector():
    assert schema == pa.schema([pa.field("vec", pa.list_(pa.float32(), 16), True)])


+def test_bare_vector_raises_clear_error():
+    namespace = {
+        "__name__": "test_model_without_pyarrow",
+        "LanceModel": LanceModel,
+        "Vector": Vector,
+    }
+
+    with pytest.raises(TypeError, match=r"Vector must be parameterized.*Vector\(128\)"):
+        exec("class TestModel(LanceModel):\n    vector: Vector", namespace)
+
+
 def test_fixed_size_list_field():
    class TestModel(pydantic.BaseModel):
        vec: Vector(16)
@@ -570,6 +570,15 @@ def test_query_builder(table):
    assert all(np.array(rs[0]["vector"]) == [1, 2])


+def test_query_multiple_vectors(table):
+    results = table.search([np.array([1, 2]), np.array([4, 5])]).limit(1).to_list()
+
+    assert len(results) == 2
+    results_by_query = {result["query_index"]: result for result in results}
+    assert results_by_query[0]["id"] == 1
+    assert results_by_query[1]["id"] == 2
+
+
 def test_with_row_id(table: lancedb.table.Table):
    rs = table.search().with_row_id(True).to_arrow()
    assert "_rowid" in rs.column_names
@@ -17,7 +17,7 @@ from unittest.mock import patch

 import lancedb
 from lancedb.dependencies import _PANDAS_AVAILABLE
-from lancedb.index import BTree, FTS, HnswFlat, HnswPq, HnswSq, IvfPq
+from lancedb.index import BTree, FTS, HnswFlat, HnswPq, HnswSq, IvfFlat, IvfPq
 import numpy as np
 import polars as pl
 import pyarrow as pa
@@ -929,6 +929,7 @@ def test_polars(mem_db: DBConnection):

    # enter table to polars dataframe
    result = table.to_polars()
+    assert isinstance(result, pl.LazyFrame)
    assert np.allclose(result.collect()["vector"].to_list(), data["vector"])

    # make sure filtering isn't broken
@@ -1845,6 +1846,27 @@ def test_add_with_empty_fixed_size_list_drops_bad_rows(mem_db: DBConnection):
    assert np.allclose(data["embedding"].to_pylist()[0], np.array([0.1] * 16))


+def test_add_nullable_fixed_size_list_with_none(mem_db: DBConnection):
+    """Regression test for issue #2340."""
+    table = mem_db.create_table(
+        "test_nullable_fixed_size_list",
+        schema=pa.schema(
+            [
+                pa.field("id", pa.string()),
+                pa.field("feature", pa.list_(pa.float32(), 256)),
+                pa.field("tags", pa.list_(pa.string())),
+            ]
+        ),
+    )
+
+    table.add([{"id": "1", "feature": None, "tags": ["tag1", "tag2"]}])
+
+    result = table.to_arrow()
+    assert result.to_pylist() == [
+        {"id": "1", "feature": None, "tags": ["tag1", "tag2"]}
+    ]
+
+
 def test_add_nullable_struct_with_none(mem_db: DBConnection):
    """Regression test for issue #2654: a nullable struct column whose
    first batch contains only None values must not crash in
@@ -2196,6 +2218,45 @@ def test_merge(tmp_db: DBConnection, tmp_path):
    table.merge(other_dataset, left_on="id")


+@pytest.mark.parametrize("storage_version", ["legacy", "stable"])
+def test_search_after_merge(tmp_path, storage_version):
+    pytest.importorskip("lance")
+    pd = pytest.importorskip("pandas")
+
+    db = lancedb.connect(
+        tmp_path,
+        storage_options={"new_table_data_storage_version": storage_version},
+    )
+    rng = np.random.default_rng(42)
+    row_count = 512
+    vectors = rng.standard_normal((row_count, 8)).astype(np.float32)
+    table = db.create_table(
+        "search_after_merge",
+        data=pd.DataFrame(
+            {
+                "id": [str(i) for i in range(row_count)],
+                "vector": list(vectors),
+            }
+        ),
+    )
+    table.create_index("vector", config=IvfPq(num_partitions=1, num_sub_vectors=2))
+
+    links = pd.DataFrame(
+        {
+            "id": [str(i) for i in range(row_count // 2)],
+            "link": [f"https://example.com/{i}" for i in range(row_count // 2)],
+        }
+    )
+    table.merge(links, left_on="id")
+
+    query = table.search(vectors[-1]).refine_factor(50).limit(10)
+    assert "ANN" in query.explain_plan(verbose=True)
+
+    result = query.to_arrow()
+    links_by_id = dict(zip(result["id"].to_pylist(), result["link"].to_pylist()))
+    assert links_by_id[str(row_count - 1)] is None
+
+
 def test_delete(mem_db: DBConnection):
    table = mem_db.create_table(
        "my_table",
@@ -2738,15 +2799,40 @@ def test_create_with_embedding_function(mem_db: DBConnection):
    assert actual == expected


+def test_create_f16_table_from_arrow_data(mem_db: DBConnection):
+    dimension = 32
+    num_rows = 512
+    values = pa.array(
+        np.random.default_rng(42)
+        .standard_normal(num_rows * dimension)
+        .astype(np.float16)
+    )
+    df = pa.table(
+        {
+            "text": [f"s-{i}" for i in range(num_rows)],
+            "vector": pa.FixedSizeListArray.from_arrays(values, dimension),
+        }
+    )
+    table = mem_db.create_table("f16_tbl", data=df)
+    assert table.schema.field("vector").type == pa.list_(pa.float16(), dimension)
+    table.create_index(num_partitions=2, num_sub_vectors=2)
+
+    query = df["vector"][2].as_py()
+    expected = table.search(query).limit(2).to_arrow()
+
+    assert "s-2" in expected["text"].to_pylist()
+
+
 def test_create_f16_table(mem_db: DBConnection):
    class MyTable(LanceModel):
        text: str
        vector: Vector(32, value_type=pa.float16())

+    rng = np.random.default_rng(42)
    df = pa.table(
        {
            "text": [f"s-{i}" for i in range(512)],
-            "vector": [np.random.randn(32).astype(np.float16) for _ in range(512)],
+            "vector": [rng.standard_normal(32).astype(np.float16) for _ in range(512)],
        }
    )
    table = mem_db.create_table(
@@ -2762,6 +2848,52 @@ def test_create_f16_table(mem_db: DBConnection):
    assert "s-2" in expected["text"].to_pylist()


+@pytest.mark.parametrize(
+    "index_config",
+    [
+        IvfPq(distance_type="cosine", num_partitions=2, num_sub_vectors=2),
+        IvfFlat(distance_type="cosine", num_partitions=2),
+    ],
+    ids=["ivf-pq", "ivf-flat"],
+)
+def test_f16_index_search_with_open_batch_reader(mem_db: DBConnection, index_config):
+    """Regression test for https://github.com/lancedb/lancedb/issues/2611."""
+    pytest.importorskip("pandas")
+    dimension = 32
+    num_rows = 512
+    rng = np.random.default_rng(42)
+    text_vectors = rng.standard_normal((num_rows, dimension)).astype(np.float16)
+    image_vectors = rng.standard_normal((num_rows, dimension)).astype(np.float16)
+    data = pa.table(
+        {
+            "id": np.arange(num_rows),
+            "text_embedding": pa.FixedSizeListArray.from_arrays(
+                pa.array(text_vectors.reshape(-1)), dimension
+            ),
+            "image_embedding": pa.FixedSizeListArray.from_arrays(
+                pa.array(image_vectors.reshape(-1)), dimension
+            ),
+        }
+    )
+    table = mem_db.create_table("f16_index_with_open_reader", data=data)
+    table.create_index("image_embedding", config=index_config)
+
+    reader = table.search().select(["id", "text_embedding"]).to_batches()
+    for batch in reader:
+        for _, _row in batch.to_pandas().iterrows():
+            result = (
+                table.search(image_vectors[2], vector_column_name="image_embedding")
+                .select(["id", "_distance"])
+                .distance_type("cosine")
+                .limit(10)
+                .to_pandas()
+            )
+            assert result.iloc[0]["id"] == 2
+            return
+
+    pytest.fail("expected the outer query to return a batch")
+
+
 def test_add_with_embedding_function(mem_db: DBConnection):
    emb = EmbeddingFunctionRegistry.get_instance().get("test").create()

@@ -3627,7 +3759,8 @@ def test_stats(mem_db: DBConnection):
    stats = table.stats()
    print(f"{stats=}")
    assert stats == {
-        "total_bytes": 60,
+        # Full on-disk size of the data file, footer and metadata included.
+        "total_bytes": 633,
        "num_rows": 2,
        "num_indices": 0,
        "fragment_stats": {
@@ -3645,6 +3778,13 @@ def test_stats(mem_db: DBConnection):
        },
    }

+    # Index files count toward total_bytes too (only deletion files and
+    # manifests are excluded).
+    table.create_index("id", config=BTree())
+    stats_with_index = table.stats()
+    assert stats_with_index["num_indices"] == 1
+    assert stats_with_index["total_bytes"] > stats["total_bytes"]
+

 def test_create_table_empty_list_with_schema(mem_db: DBConnection):
    """Test creating table with empty list data and schema
@@ -0,0 +1,15 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+from typing import assert_type
+
+import lancedb
+from lancedb import AsyncConnection, DBConnection
+
+
+def check_connect_type() -> None:
+    assert_type(lancedb.connect("memory://"), DBConnection)
+
+
+async def check_connect_async_type() -> None:
+    assert_type(await lancedb.connect_async("memory://"), AsyncConnection)
@@ -28,11 +28,72 @@ use pyo3::{
    Bound, FromPyObject, Py, PyAny, PyRef, PyResult, Python,
    exceptions::{PyRuntimeError, PyValueError},
    pyclass, pyfunction, pymethods,
-    types::{IntoPyDict, PyAnyMethods, PyBytes, PyDict, PyDictMethods},
+    types::{IntoPyDict, PyAnyMethods, PyBytes, PyDict, PyDictMethods, PyList, PyListMethods},
 };

 mod scannable;

+/// Convert `LsmStats` to a Python dict, preserving the per-bucket list.
+///
+/// Deliberately not flattened to a table-level summary: a table is N
+/// buckets on one node, and the per-bucket detail is the reason the
+/// endpoint exists — flattening hides the single hot bucket someone opened
+/// it to find.
+fn lsm_stats_to_py(py: Python<'_>, stats: &lancedb::table::LsmStats) -> PyResult<Py<PyDict>> {
+    let out = PyDict::new(py);
+    let buckets = PyList::empty(py);
+    for b in &stats.buckets {
+        let e = PyDict::new(py);
+        e.set_item("shard_id", &b.shard_id)?;
+        e.set_item("status", &b.status)?;
+        e.set_item("writer_epoch", b.writer_epoch)?;
+        e.set_item("manifest_version", b.manifest_version)?;
+        e.set_item("current_generation", b.current_generation)?;
+        e.set_item(
+            "replay_after_wal_entry_position",
+            b.replay_after_wal_entry_position,
+        )?;
+        e.set_item(
+            "wal_entry_position_last_seen",
+            b.wal_entry_position_last_seen,
+        )?;
+
+        let generations = PyList::empty(py);
+        for g in &b.generations {
+            let ge = PyDict::new(py);
+            ge.set_item("generation", g.generation)?;
+            ge.set_item("bytes", g.bytes)?;
+            ge.set_item("rows", g.rows)?;
+            generations.append(ge)?;
+        }
+        e.set_item("generations", generations)?;
+        e.set_item("compacting", b.compacting)?;
+
+        e.set_item(
+            "memtables",
+            b.memtables
+                .as_ref()
+                .map(|ms| {
+                    let l = PyList::empty(py);
+                    for m in ms {
+                        let d = PyDict::new(py);
+                        d.set_item("generation", m.generation)?;
+                        d.set_item("rows", m.rows)?;
+                        d.set_item("bytes", m.bytes)?;
+                        d.set_item("batches", m.batches)?;
+                        d.set_item("indexes", m.indexes.clone())?;
+                        l.append(d)?;
+                    }
+                    PyResult::Ok(l.unbind())
+                })
+                .transpose()?,
+        )?;
+        buckets.append(e)?;
+    }
+    out.set_item("buckets", buckets)?;
+    Ok(out.unbind())
+}
+
 #[derive(FromPyObject)]
 enum PredicateArg {
    Expr(PyExpr),
@@ -185,12 +246,22 @@ impl From<lancedb::table::MergeResult> for MergeResult {
    }
 }

+/// Render for `__repr__`, so the default reads as Python's `None` rather than
+/// Rust's `Some([..])`.
+fn fmt_maintained(maintained: &Option<Vec<String>>) -> String {
+    match maintained {
+        Some(names) => format!("{:?}", names),
+        None => "None".to_string(),
+    }
+}
+
 /// Specification selecting Lance's MemWAL LSM-style write path for
 /// `merge_insert`.
 ///
 /// Constructed via the `bucket(...)`, `identity(...)`, or `unsharded()`
 /// classmethods, then optionally chain `with_maintained_indexes(...)` and
-/// `with_writer_config_defaults(...)`.
+/// `with_writer_config_defaults(...)`. A fresh spec maintains every index the
+/// MemWAL supports, resolved on install.
 #[pyclass(from_py_object)]
 #[derive(Clone, Debug)]
 pub struct LsmWriteSpec {
@@ -230,11 +301,11 @@ impl LsmWriteSpec {
        }
    }

-    /// Replace the list of indexes the MemWAL should keep up to date as
-    /// rows are appended. Each name must reference an index that
-    /// already exists on the table at the time `set_lsm_write_spec`
-    /// is called.
-    pub fn with_maintained_indexes(&self, indexes: Vec<String>) -> Self {
+    /// Set which indexes the MemWAL maintains. `None` (the default)
+    /// resolves every supported index on install; a list is verbatim,
+    /// and an empty list maintains nothing.
+    #[pyo3(signature = (indexes))]
+    pub fn with_maintained_indexes(&self, indexes: Option<Vec<String>>) -> Self {
        Self {
            inner: self.inner.clone().with_maintained_indexes(indexes),
        }
@@ -256,23 +327,29 @@ impl LsmWriteSpec {
                maintained_indexes,
                writer_config_defaults,
            } => format!(
-                "LsmWriteSpec.bucket(column={:?}, num_buckets={}, maintained_indexes={:?}, writer_config_defaults={:?})",
-                column, num_buckets, maintained_indexes, writer_config_defaults,
+                "LsmWriteSpec.bucket(column={:?}, num_buckets={}, maintained_indexes={}, writer_config_defaults={:?})",
+                column,
+                num_buckets,
+                fmt_maintained(maintained_indexes),
+                writer_config_defaults,
            ),
            lancedb::table::LsmWriteSpec::Identity {
                column,
                maintained_indexes,
                writer_config_defaults,
            } => format!(
-                "LsmWriteSpec.identity(column={:?}, maintained_indexes={:?}, writer_config_defaults={:?})",
-                column, maintained_indexes, writer_config_defaults,
+                "LsmWriteSpec.identity(column={:?}, maintained_indexes={}, writer_config_defaults={:?})",
+                column,
+                fmt_maintained(maintained_indexes),
+                writer_config_defaults,
            ),
            lancedb::table::LsmWriteSpec::Unsharded {
                maintained_indexes,
                writer_config_defaults,
            } => format!(
-                "LsmWriteSpec.unsharded(maintained_indexes={:?}, writer_config_defaults={:?})",
-                maintained_indexes, writer_config_defaults,
+                "LsmWriteSpec.unsharded(maintained_indexes={}, writer_config_defaults={:?})",
+                fmt_maintained(maintained_indexes),
+                writer_config_defaults,
            ),
        }
    }
@@ -307,10 +384,10 @@ impl LsmWriteSpec {
        }
    }

-    /// Names of indexes the MemWAL should keep up to date during writes.
+    /// Indexes the MemWAL keeps up to date, or `None` for every supported one.
    #[getter]
-    pub fn maintained_indexes(&self) -> Vec<String> {
-        self.inner.maintained_indexes().to_vec()
+    pub fn maintained_indexes(&self) -> Option<Vec<String>> {
+        self.inner.maintained_indexes().map(<[String]>::to_vec)
    }

    /// Default `ShardWriter` configuration recorded by this spec.
@@ -1339,6 +1416,51 @@ impl Table {
        })
    }

+    /// Converge the table's LSM write path into its base table.
+    ///
+    /// Best-effort: with writes flowing, new rows may land after the last
+    /// pass. Errors if the table stops making progress.
+    pub fn checkpoint_lsm(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.inner_ref()?.clone();
+        future_into_py(self_.py(), async move {
+            inner.checkpoint_lsm().await.infer_error()
+        })
+    }
+
+    /// Seal every bucket's active memtable into L0.
+    pub fn flush_lsm(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.inner_ref()?.clone();
+        future_into_py(
+            self_.py(),
+            async move { inner.flush_lsm().await.infer_error() },
+        )
+    }
+
+    /// Trigger a background L0 → base pass per bucket. Returns once the
+    /// passes are dispatched, not once they finish — watch `get_lsm_stats`.
+    pub fn compact_lsm(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.inner_ref()?.clone();
+        future_into_py(self_.py(), async move {
+            inner.compact_lsm().await.infer_error()
+        })
+    }
+
+    /// Live LSM state, or `None` when the LSM write path is not enabled.
+    #[pyo3(signature = (include_generation_rows=false))]
+    pub fn get_lsm_stats(
+        self_: PyRef<'_, Self>,
+        include_generation_rows: bool,
+    ) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.inner_ref()?.clone();
+        future_into_py(self_.py(), async move {
+            let stats = inner
+                .get_lsm_stats(include_generation_rows)
+                .await
+                .infer_error()?;
+            Python::attach(|py| stats.map(|s| lsm_stats_to_py(py, &s)).transpose())
+        })
+    }
+
    pub fn close_lsm_writers(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
        let inner = self_.inner_ref()?.clone();
        future_into_py(self_.py(), async move {
@@ -1998,7 +1998,7 @@ requires-dist = [
    { name = "pillow", marker = "extra == 'clip'", specifier = ">=12.1.1" },
    { name = "pillow", marker = "extra == 'embeddings'", specifier = ">=12.1.1" },
    { name = "pillow", marker = "extra == 'siglip'", specifier = ">=12.1.1" },
-    { name = "polars", marker = "extra == 'tests'", specifier = ">=0.19,<=1.3.0" },
+    { name = "polars", marker = "extra == 'tests'", specifier = ">=0.19,<=1.32.3" },
    { name = "pre-commit", marker = "extra == 'dev'", specifier = ">=3.5.0" },
    { name = "pyarrow", specifier = ">=16" },
    { name = "pyarrow", marker = "extra == 'tests'", specifier = "<25" },
@@ -1,6 +1,6 @@
 [package]
 name = "lancedb"
-version = "0.37.1-beta.0"
+version = "0.37.1-beta.1"
 edition.workspace = true
 description = "LanceDB: A serverless, low-latency vector database for AI applications"
 license.workspace = true
@@ -100,7 +100,8 @@ 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"] }
+roaring = "0.11.4"
+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" }
@@ -1294,9 +1294,11 @@ mod tests {
    use crate::connection::ConnectRequest;
    use crate::data::scannable::Scannable;
    use crate::database::{CreateTableMode, CreateTableRequest};
-    use crate::table::WriteOptions;
+    use crate::query::QueryRequest;
+    use crate::table::{AnyQuery, WriteOptions};
    use arrow_array::{Int32Array, RecordBatch, StringArray};
    use arrow_schema::{DataType, Field, Schema};
+    use futures::TryStreamExt;
    use std::path::PathBuf;
    use tempfile::tempdir;

@@ -1438,6 +1440,94 @@ mod tests {
        assert!(after_open.hits >= before_open.hits + 3);
    }

+    /// Regression test for https://github.com/lancedb/lancedb/issues/3197.
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn test_open_table_follows_hugging_face_symlinks() {
+        let (tempdir, db) = setup_database().await;
+        let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
+        db.create_table(CreateTableRequest {
+            name: "test".to_string(),
+            namespace_path: vec![],
+            data: Box::new(
+                RecordBatch::try_new(schema, vec![Arc::new(Int32Array::from(vec![1, 2, 3]))])
+                    .unwrap(),
+            ) as Box<dyn Scannable>,
+            mode: CreateTableMode::Create,
+            write_options: Default::default(),
+            location: None,
+            namespace_client: None,
+        })
+        .await
+        .unwrap();
+
+        let table_dir = tempdir.path().join("test.lance");
+        let versions_dir = table_dir.join("_versions");
+        let manifest_path = std::fs::read_dir(&versions_dir)
+            .unwrap()
+            .map(|entry| entry.unwrap().path())
+            .find(|path| path.extension().is_some_and(|ext| ext == "manifest"))
+            .unwrap();
+        let data_path = std::fs::read_dir(table_dir.join("data"))
+            .unwrap()
+            .map(|entry| entry.unwrap().path())
+            .find(|path| path.extension().is_some_and(|ext| ext == "lance"))
+            .unwrap();
+
+        // Hugging Face snapshots keep dataset objects in a separate blob directory and
+        // expose them through relative symlinks.
+        let blobs_dir = tempdir.path().join("blobs");
+        std::fs::create_dir(&blobs_dir).unwrap();
+        let manifest_blob = "9b603c63d0e692e05d58be25605f2f2064cc781e5ff94fe983a405059547b816";
+        let data_blob = "be64f20e5723bd0a27cfdbdb41cf7d6fad94cd572a71973b717fb8340f4310c5";
+        std::fs::rename(&manifest_path, blobs_dir.join(manifest_blob)).unwrap();
+        std::fs::rename(&data_path, blobs_dir.join(data_blob)).unwrap();
+        std::os::unix::fs::symlink(Path::new("../../blobs").join(manifest_blob), &manifest_path)
+            .unwrap();
+        std::os::unix::fs::symlink(Path::new("../../blobs").join(data_blob), &data_path).unwrap();
+        let symlink_len = std::fs::symlink_metadata(&manifest_path).unwrap().len();
+        let target_len = std::fs::metadata(&manifest_path).unwrap().len();
+        assert_ne!(symlink_len, target_len);
+
+        drop(db);
+        let db = ListingDatabase::connect_with_options(&ConnectRequest {
+            uri: tempdir.path().to_str().unwrap().to_string(),
+            #[cfg(feature = "remote")]
+            client_config: Default::default(),
+            options: Default::default(),
+            namespace_client_properties: Default::default(),
+            manifest_enabled: false,
+            read_consistency_interval: None,
+            session: None,
+        })
+        .await
+        .unwrap();
+
+        let table = db
+            .open_table(OpenTableRequest {
+                name: "test".to_string(),
+                namespace_path: vec![],
+                index_cache_size: None,
+                lance_read_params: None,
+                location: None,
+                namespace_client: None,
+                managed_versioning: None,
+            })
+            .await
+            .unwrap();
+        let batches = table
+            .query(
+                &AnyQuery::Query(QueryRequest::default()),
+                Default::default(),
+            )
+            .await
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+        assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 3);
+    }
+
    #[tokio::test]
    async fn test_clone_table_basic() {
        let (_tempdir, db) = setup_database().await;
@@ -2342,7 +2432,7 @@ mod tests {

    #[tokio::test]
    async fn test_table_uri() {
-        let (_tempdir, db) = setup_database().await;
+        let (_tempdir, mut db) = setup_database().await;

        let mut pb = PathBuf::new();
        pb.push(db.uri.clone());
@@ -2351,6 +2441,18 @@ mod tests {
        let expected = pb.to_str().unwrap();
        let uri = db.table_uri("test").ok().unwrap();
        assert_eq!(uri, expected);
+
+        // URI paths always use forward slashes, even on Windows. Using
+        // `Path::join` here used to produce `az://container/prefix\\test.lance`,
+        // which Azure treated as a different object from the table returned by
+        // `table_names` (https://github.com/lancedb/lancedb/issues/1072).
+        for base_uri in ["az://container/prefix", "az://container/prefix/"] {
+            db.uri = base_uri.to_string();
+            assert_eq!(
+                db.table_uri("test").unwrap(),
+                "az://container/prefix/test.lance"
+            );
+        }
    }

    /// Regression: connecting via a URL-style URI (which goes through
@@ -169,6 +169,12 @@ impl From<DataFusionError> for Error {

 impl From<lance::Error> for Error {
    fn from(source: lance::Error) -> Self {
+        if has_unsupported_local_filesystem_source(&source) {
+            return Self::NotSupported {
+                message: "the filesystem does not support an operation required for safe Lance commits (such as atomic rename). Object-storage mounts such as Mountpoint for Amazon S3 are not supported; use the native object-store URI (for example, s3://bucket/path) instead".to_string(),
+            };
+        }
+
        // Try to unwrap external errors that were wrapped by lance
        match source {
            lance::Error::Wrapped { error, .. } => Self::from_box_error(error),
@@ -181,6 +187,27 @@ impl From<lance::Error> for Error {
    }
 }

+fn has_unsupported_local_filesystem_source(error: &(dyn std::error::Error + 'static)) -> bool {
+    let mut current = Some(error);
+    let mut is_local_filesystem = false;
+    let mut is_unsupported = false;
+    while let Some(error) = current {
+        is_local_filesystem |= error
+            .downcast_ref::<object_store::Error>()
+            .is_some_and(|error| {
+                matches!(error, object_store::Error::Generic { store, .. } if *store == "LocalFileSystem")
+            });
+        is_unsupported |= error
+            .downcast_ref::<std::io::Error>()
+            .is_some_and(|error| error.kind() == std::io::ErrorKind::Unsupported);
+        if is_local_filesystem && is_unsupported {
+            return true;
+        }
+        current = error.source();
+    }
+    false
+}
+
 impl Error {
    fn from_box_error(mut source: Box<dyn std::error::Error + Send + Sync>) -> Self {
        source = match source.downcast::<Self>() {
@@ -270,3 +297,46 @@ impl From<candle_core::Error> for Error {
        }
    }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn unsupported_filesystem_operations_have_actionable_error() {
+        let object_store_error = object_store::Error::Generic {
+            store: "LocalFileSystem",
+            source: Box::new(std::io::Error::from(std::io::ErrorKind::Unsupported)),
+        };
+        let lance_error = lance::Error::io_source(Box::new(object_store_error));
+
+        let error = Error::from(lance_error);
+
+        assert!(matches!(
+            error,
+            Error::NotSupported { message }
+                if message.contains("Mountpoint for Amazon S3")
+                    && message.contains("s3://bucket/path")
+        ));
+    }
+
+    #[test]
+    fn other_io_errors_remain_lance_errors() {
+        let object_store_error = object_store::Error::Generic {
+            store: "LocalFileSystem",
+            source: Box::new(std::io::Error::from(std::io::ErrorKind::PermissionDenied)),
+        };
+        let lance_error = lance::Error::io_source(Box::new(object_store_error));
+
+        assert!(matches!(Error::from(lance_error), Error::Lance { .. }));
+    }
+
+    #[test]
+    fn unsupported_non_filesystem_errors_remain_lance_errors() {
+        let lance_error = lance::Error::io_source(Box::new(std::io::Error::from(
+            std::io::ErrorKind::Unsupported,
+        )));
+
+        assert!(matches!(Error::from(lance_error), Error::Lance { .. }));
+    }
+}
@@ -23,11 +23,13 @@ use crate::table::AddResult;
 use crate::table::BranchDiff;
 use crate::table::DeleteResult;
 use crate::table::DropColumnsResult;
+use crate::table::LsmStats;
 use crate::table::LsmWriteSpec;
 use crate::table::MergeBranchResult;
 use crate::table::MergeResult;
 use crate::table::Tags;
 use crate::table::UpdateResult;
+use crate::table::lsm_stats::GetLsmStatsResponse;
 use crate::table::merge::MergeFilter;
 use crate::table::query::create_multi_vector_plan;
 use crate::table::write_progress::FinishOnDrop;
@@ -991,6 +993,18 @@ impl<S: HttpSend> RemoteTable<S> {
        }
    }

+    /// Send an LSM operator request with the transport retry layer **off**.
+    ///
+    /// Retry policy on these routes belongs to the checkpoint loop, which
+    /// reads the status and can tell contention from a lost claim. Leaving the
+    /// transport layer on would re-ask on its own schedule first, and surface
+    /// an `Error::Retry` whose status the loop would then have to unwrap.
+    async fn send_lsm_route(&self, request: RequestBuilder) -> Result<(String, reqwest::Response)> {
+        let (request_id, response) = self.send(request, false).await?;
+        let response = self.check_table_response(&request_id, response).await?;
+        Ok((request_id, response))
+    }
+
    /// Build a POST request and attach the read-freshness headers
    /// (`x-lancedb-min-version`, `x-lancedb-min-timestamp`).
    fn post_read(&self, uri: &str) -> RequestBuilder {
@@ -2468,13 +2482,47 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
        })
    }

+    async fn flush_lsm(&self) -> Result<()> {
+        let request = self
+            .client
+            .post(&format!("/v1/table/{}/flush_lsm/", self.identifier));
+        self.send_lsm_route(request).await?;
+        Ok(())
+    }
+
+    async fn compact_lsm(&self) -> Result<()> {
+        let request = self
+            .client
+            .post(&format!("/v1/table/{}/compact_lsm/", self.identifier));
+        self.send_lsm_route(request).await?;
+        Ok(())
+    }
+
+    async fn get_lsm_stats(&self, include_generation_rows: bool) -> Result<Option<LsmStats>> {
+        // Read-semantics POST, like `get_lsm_write_spec`.
+        let request = self
+            .post_read(&format!("/v1/table/{}/get_lsm_stats/", self.identifier))
+            .json(&serde_json::json!({
+                "include_generation_rows": include_generation_rows,
+            }));
+        let (request_id, response) = self.send_lsm_route(request).await?;
+        let body = response.text().await.err_to_http(request_id.clone())?;
+        let parsed: GetLsmStatsResponse = serde_json::from_str(&body).map_err(|e| Error::Http {
+            source: format!("Failed to parse get_lsm_stats response: {e}").into(),
+            request_id,
+            status_code: None,
+        })?;
+        // `null` — and only — when the table has no LSM write path.
+        Ok(parsed.lsm_stats)
+    }
+
    async fn set_lsm_write_spec(&self, spec: LsmWriteSpec) -> Result<()> {
        self.check_mutable().await?;

        // Map the spec onto the server's request DTO. `sharding` is internally
-        // tagged on `mode` to mirror sophon's `Sharding` enum; `maintained_indexes`
-        // and `writer_config_defaults` are sent verbatim (an empty list means "no
-        // maintained indexes", not "default to all").
+        // tagged on `mode` to mirror sophon's `Sharding` enum. A null
+        // `maintained_indexes` asks the server to resolve every maintainable
+        // index at HEAD; a list is verbatim, an empty one meaning none.
        let sharding = match &spec {
            LsmWriteSpec::Bucket {
                column,
@@ -6551,7 +6599,7 @@ mod tests {
                .unwrap()
        });
        let spec = crate::table::LsmWriteSpec::unsharded()
-            .with_maintained_indexes(["id_idx"])
+            .with_maintained_indexes(vec!["id_idx".to_string()])
            .with_writer_config_defaults([("max_memtable_rows", "1000")]);
        table.set_lsm_write_spec(spec).await.unwrap();
    }
@@ -6570,7 +6618,8 @@ mod tests {
                body["sharding"],
                serde_json::json!({ "mode": "bucket", "column": "id", "num_buckets": 16 })
            );
-            assert_eq!(body["maintained_indexes"], serde_json::json!([]));
+            // An unpinned maintained set sends null: resolve server-side.
+            assert_eq!(body["maintained_indexes"], serde_json::Value::Null);
            http::Response::builder().status(200).body("{}").unwrap()
        });
        table
@@ -6579,6 +6628,23 @@ mod tests {
            .unwrap();
    }

+    /// `[]` (none) must stay distinguishable on the wire from null (all).
+    #[tokio::test]
+    async fn test_set_lsm_write_spec_no_maintained_indexes() {
+        let table = Table::new_with_handler("my_table", |request| {
+            let body = request.body().unwrap().as_bytes().unwrap();
+            let body: serde_json::Value = serde_json::from_slice(body).unwrap();
+            assert_eq!(body["maintained_indexes"], serde_json::json!([]));
+            http::Response::builder().status(200).body("{}").unwrap()
+        });
+        table
+            .set_lsm_write_spec(
+                crate::table::LsmWriteSpec::bucket("id", 16).with_maintained_indexes(Vec::new()),
+            )
+            .await
+            .unwrap();
+    }
+
    #[tokio::test]
    async fn test_set_lsm_write_spec_identity() {
        let table = Table::new_with_handler("my_table", |request| {
@@ -6653,7 +6719,7 @@ mod tests {
            } => {
                assert_eq!(column, "id");
                assert_eq!(num_buckets, 4);
-                assert_eq!(maintained_indexes, vec!["id_idx".to_string()]);
+                assert_eq!(maintained_indexes, Some(vec!["id_idx".to_string()]));
                assert_eq!(
                    writer_config_defaults
                        .get("durable_write")
@@ -6682,6 +6748,499 @@ mod tests {
        assert!(table.get_lsm_write_spec().await.unwrap().is_none());
    }

+    /// Build a `get_lsm_stats` body for one bucket holding `generations`.
+    fn stats_body(generations: &[u64], compacting: bool) -> String {
+        serde_json::json!({
+            "lsm_stats": {
+                "buckets": [{
+                    "shard_id": "b0",
+                    "status": "Active",
+                    "writer_epoch": 1,
+                    "manifest_version": 1,
+                    "current_generation": generations.iter().max().copied().unwrap_or(0) + 1,
+                    "replay_after_wal_entry_position": 0,
+                    "wal_entry_position_last_seen": 0,
+                    "generations": generations.iter()
+                        .map(|g| serde_json::json!({ "generation": g, "bytes": 1 }))
+                        .collect::<Vec<_>>(),
+                    "compacting": compacting,
+                    "memtables": [],
+                }],
+            }
+        })
+        .to_string()
+    }
+
+    /// `flush_lsm` / `compact_lsm` answer 202 with no body at all.
+    fn accepted() -> http::Response<String> {
+        http::Response::builder()
+            .status(202)
+            .body(String::new())
+            .unwrap()
+    }
+
+    fn ok_json(body: String) -> http::Response<String> {
+        http::Response::builder().status(200).body(body).unwrap()
+    }
+
+    /// A flush landing in an empty L0 finishes on the opening stats read
+    /// alone. Asserting zero compacts is the point: "it returned Ok" is also
+    /// true of a loop that ran a pointless pass.
+    #[tokio::test(start_paused = true)]
+    async fn test_checkpoint_short_circuits_on_empty_l0() {
+        let compacts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let seen = compacts.clone();
+        let table = Table::new_with_handler("my_table", move |request| {
+            let path = request.url().path().to_string();
+            if path.contains("compact_lsm") {
+                seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+                panic!("an already-converged table must issue no compact calls");
+            }
+            if path.contains("flush_lsm") {
+                return accepted();
+            }
+            assert_eq!(path, "/v1/table/my_table/get_lsm_stats/");
+            ok_json(stats_body(&[], false))
+        });
+
+        table.checkpoint_lsm().await.unwrap();
+        assert_eq!(compacts.load(std::sync::atomic::Ordering::SeqCst), 0);
+    }
+
+    /// The loop triggers compaction until every generation that existed at
+    /// the start is gone, one bounded prefix per pass.
+    #[tokio::test(start_paused = true)]
+    async fn test_checkpoint_triggers_until_targets_are_drained() {
+        let compacts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let seen = compacts.clone();
+        let table = Table::new_with_handler("my_table", move |request| {
+            let path = request.url().path().to_string();
+            if path.contains("flush_lsm") {
+                return accepted();
+            }
+            if path.contains("compact_lsm") {
+                seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+                return accepted();
+            }
+            // Each pass drains the oldest generation.
+            let drained = seen.load(std::sync::atomic::Ordering::SeqCst);
+            let left: Vec<u64> = [1u64, 2, 3].into_iter().skip(drained).collect();
+            ok_json(stats_body(&left, false))
+        });
+
+        table.checkpoint_lsm().await.unwrap();
+        assert_eq!(
+            compacts.load(std::sync::atomic::Ordering::SeqCst),
+            3,
+            "one trigger per generation prefix, then stop"
+        );
+    }
+
+    /// Generations created *during* the checkpoint are not waited on, which
+    /// is what lets the loop terminate on a table taking writes where "L0 is
+    /// empty" never becomes true.
+    #[tokio::test(start_paused = true)]
+    async fn test_checkpoint_ignores_generations_created_while_it_runs() {
+        let compacts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let seen = compacts.clone();
+        let table = Table::new_with_handler("my_table", move |request| {
+            let path = request.url().path().to_string();
+            if path.contains("flush_lsm") {
+                return accepted();
+            }
+            if path.contains("compact_lsm") {
+                seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+                return accepted();
+            }
+            // Target is 5. One pass drains it; a writer keeps adding above.
+            let n = seen.load(std::sync::atomic::Ordering::SeqCst);
+            let body = if n == 0 {
+                stats_body(&[5], false)
+            } else {
+                stats_body(&[6, 7], false)
+            };
+            ok_json(body)
+        });
+
+        table.checkpoint_lsm().await.unwrap();
+        assert_eq!(
+            compacts.load(std::sync::atomic::Ordering::SeqCst),
+            1,
+            "the loop must not chase generations written after it started"
+        );
+    }
+
+    /// Contention is a 429 and must be retried. The server keeps it off 503
+    /// precisely so the client can act on the status alone — reading it as
+    /// terminal stops the checkpoint early on a healthy node.
+    #[tokio::test(start_paused = true)]
+    async fn test_checkpoint_retries_contention() {
+        let compacts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let seen = compacts.clone();
+        let table = Table::new_with_handler("my_table", move |request| {
+            let path = request.url().path().to_string();
+            if path.contains("flush_lsm") {
+                return accepted();
+            }
+            if path.contains("compact_lsm") {
+                // First two triggers: every bucket already latched.
+                if seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst) < 2 {
+                    return http::Response::builder()
+                        .status(429)
+                        .body(r#"{"code":21,"error":"Too many concurrent writes"}"#.to_string())
+                        .unwrap();
+                }
+                return accepted();
+            }
+            let accepted_triggers = seen
+                .load(std::sync::atomic::Ordering::SeqCst)
+                .saturating_sub(2);
+            let left: Vec<u64> = if accepted_triggers == 0 {
+                vec![1]
+            } else {
+                vec![]
+            };
+            ok_json(stats_body(&left, false))
+        });
+
+        table
+            .checkpoint_lsm()
+            .await
+            .expect("contention must not abort the checkpoint");
+        assert_eq!(
+            compacts.load(std::sync::atomic::Ordering::SeqCst),
+            3,
+            "assert the retry count, not just the outcome"
+        );
+    }
+
+    /// A transient fault on the poll must not abort the checkpoint. This route
+    /// meets the most contention — it runs every `POLL_INTERVAL` for the
+    /// checkpoint's whole life, with the transport retry layer disabled — yet
+    /// was the one call reached with a bare `?`.
+    #[tokio::test(start_paused = true)]
+    async fn test_checkpoint_retries_a_contended_stats_poll() {
+        let polls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let seen = polls.clone();
+        let table = Table::new_with_handler("my_table", move |request| {
+            let path = request.url().path().to_string();
+            if path.contains("flush_lsm") || path.contains("compact_lsm") {
+                return accepted();
+            }
+            // The opening read lands; the next two polls are latched out.
+            let n = seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+            if (1..3).contains(&n) {
+                return http::Response::builder()
+                    .status(429)
+                    .body(r#"{"code":21,"error":"Too many concurrent writes"}"#.to_string())
+                    .unwrap();
+            }
+            ok_json(stats_body(if n < 4 { &[1] } else { &[] }, false))
+        });
+
+        table
+            .checkpoint_lsm()
+            .await
+            .expect("a contended poll must be retried, not surfaced");
+        assert_eq!(
+            polls.load(std::sync::atomic::Ordering::SeqCst),
+            5,
+            "the two rejected polls must be re-issued, not skipped"
+        );
+    }
+
+    /// Contention and a lost claim draw on separate budgets: five straight
+    /// 429s on `flush`, more than `MAX_REISSUES`, must still converge. On one
+    /// shared counter this spent the re-issue cap and then reported a lost
+    /// claim nothing had ever reported.
+    #[tokio::test(start_paused = true)]
+    async fn test_contention_does_not_exhaust_the_reissue_budget() {
+        let flushes = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let seen = flushes.clone();
+        let table = Table::new_with_handler("my_table", move |request| {
+            let path = request.url().path().to_string();
+            if path.contains("flush_lsm") {
+                if seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst) < 5 {
+                    return http::Response::builder()
+                        .status(429)
+                        .body(r#"{"code":21,"error":"Too many concurrent writes"}"#.to_string())
+                        .unwrap();
+                }
+                return accepted();
+            }
+            if path.contains("compact_lsm") {
+                return accepted();
+            }
+            ok_json(stats_body(&[], false))
+        });
+
+        table
+            .checkpoint_lsm()
+            .await
+            .expect("contention must not be reported as a lost claim");
+        assert_eq!(
+            flushes.load(std::sync::atomic::Ordering::SeqCst),
+            6,
+            "five retries against one seal, then it lands"
+        );
+    }
+
+    /// An exhausted retry budget surfaces the fault that consumed it, not a
+    /// message the loop invented: "429, nine times" points an operator at a
+    /// saturated pool, a generic runtime error points them nowhere.
+    #[tokio::test(start_paused = true)]
+    async fn test_exhausted_retries_surface_the_underlying_fault() {
+        let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let seen = calls.clone();
+        let table = Table::new_with_handler("my_table", move |_request| {
+            seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+            http::Response::builder()
+                .status(429)
+                .body(r#"{"code":21,"error":"Too many concurrent writes"}"#.to_string())
+                .unwrap()
+        });
+
+        let err = table.checkpoint_lsm().await.unwrap_err();
+        assert!(
+            matches!(&err, Error::Http { status_code: Some(s), .. } if s.as_u16() == 429),
+            "the fault that spent the budget must be the one reported: {err:?}"
+        );
+        assert_eq!(
+            calls.load(std::sync::atomic::Ordering::SeqCst),
+            9,
+            "one call plus MAX_RETRIES — the re-issue budget is not spent on top"
+        );
+    }
+
+    /// A draining node is terminal, but the client does not know that from the
+    /// status: draining and a proxy blip are both 503, and telling them apart
+    /// takes parsing the body for a namespace code. So it spends the retry
+    /// budget and then reports what the server said — the drain gate never
+    /// releases, so the answer does not change, and the operator still reads
+    /// "WAL node draining" in the error.
+    #[tokio::test(start_paused = true)]
+    async fn test_draining_surfaces_after_the_retry_budget() {
+        let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let seen = calls.clone();
+        let table = Table::new_with_handler("my_table", move |_request| {
+            seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+            http::Response::builder()
+                .status(503)
+                .body(r#"{"code":19,"error":"WAL node draining"}"#.to_string())
+                .unwrap()
+        });
+
+        let err = table.checkpoint_lsm().await.unwrap_err();
+        let message = err.to_string();
+        assert!(
+            matches!(&err, Error::Http { status_code: Some(s), .. } if s.as_u16() == 503),
+            "the 503 must surface as itself: {err:?}"
+        );
+        assert!(
+            message.contains("WAL node draining"),
+            "the server's own diagnosis must survive to the caller: {message}"
+        );
+        assert_eq!(
+            calls.load(std::sync::atomic::Ordering::SeqCst),
+            9,
+            "one call plus MAX_RETRIES, then it reports rather than spinning"
+        );
+    }
+
+    /// A long stall with nothing compacting must keep waiting, not fail. The
+    /// client cannot judge this: a checkpoint queued behind unrelated tables
+    /// on the pod-wide compactor pool reports exactly these numbers — flat
+    /// generations, an idle latch — as one whose merges are failing. The
+    /// deadline is the caller's.
+    #[tokio::test(start_paused = true)]
+    async fn test_checkpoint_waits_out_a_long_stall_rather_than_failing() {
+        let polls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let seen = polls.clone();
+        let table = Table::new_with_handler("my_table", move |request| {
+            let path = request.url().path().to_string();
+            if path.contains("flush_lsm") || path.contains("compact_lsm") {
+                return accepted();
+            }
+            // Flat for far longer than any bound this loop ever had, with
+            // `compacting: false` throughout — then it drains.
+            let n = seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+            ok_json(stats_body(if n < 40 { &[1, 2] } else { &[] }, false))
+        });
+
+        table
+            .checkpoint_lsm()
+            .await
+            .expect("a stall is the server being slow, not the client's call to make");
+        assert!(
+            polls.load(std::sync::atomic::Ordering::SeqCst) > 40,
+            "the loop must have kept polling well past the old ten-poll bound"
+        );
+    }
+
+    /// A pass already owns the latch on every outstanding bucket, so the loop
+    /// waits rather than piling on triggers it would only refuse. This is the
+    /// sole thing `compacting` is read for.
+    #[tokio::test(start_paused = true)]
+    async fn test_checkpoint_waits_while_a_pass_is_running() {
+        let polls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let compacts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let seen_polls = polls.clone();
+        let seen_compacts = compacts.clone();
+        let table = Table::new_with_handler("my_table", move |request| {
+            let path = request.url().path().to_string();
+            if path.contains("flush_lsm") {
+                return accepted();
+            }
+            if path.contains("compact_lsm") {
+                seen_compacts.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+                return accepted();
+            }
+            // Latched for many polls, then done.
+            let n = seen_polls.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+            ok_json(if n > 15 {
+                stats_body(&[], false)
+            } else {
+                stats_body(&[1], true)
+            })
+        });
+
+        table
+            .checkpoint_lsm()
+            .await
+            .expect("a running pass is progress, not a stall");
+        assert_eq!(
+            compacts.load(std::sync::atomic::Ordering::SeqCst),
+            0,
+            "never trigger against a bucket already compacting"
+        );
+    }
+
+    /// WAL off ⇒ `None`; WAL on ⇒ a fully populated `Some` with no field
+    /// defaulting to a zero it did not measure. `include_generation_rows`
+    /// rides in the body and is off unless asked for.
+    #[tokio::test]
+    async fn test_get_lsm_stats_round_trip() {
+        let table = Table::new_with_handler("my_table", |request| {
+            assert_eq!(request.url().path(), "/v1/table/my_table/get_lsm_stats/");
+            let body = request.body().unwrap().as_bytes().unwrap();
+            let body: serde_json::Value = serde_json::from_slice(body).unwrap();
+            assert_eq!(
+                body["include_generation_rows"], true,
+                "the flag must reach the server, not be silently dropped"
+            );
+            let response = serde_json::json!({
+                "lsm_stats": {
+                    "buckets": [{
+                        "shard_id": "b0",
+                        "status": "Active",
+                        "writer_epoch": 3,
+                        "manifest_version": 11,
+                        "current_generation": 9,
+                        "replay_after_wal_entry_position": 100,
+                        "wal_entry_position_last_seen": 140,
+                        "generations": [{ "generation": 8, "bytes": 4096, "rows": 30 }],
+                        "compacting": false,
+                        "memtables": [
+                            { "generation": 9, "rows": 12, "bytes": 900, "batches": 2,
+                              "indexes": ["vec_idx"] }
+                        ],
+                    }],
+                }
+            });
+            http::Response::builder()
+                .status(200)
+                .body(response.to_string())
+                .unwrap()
+        });
+
+        let stats = table
+            .get_lsm_stats(true)
+            .await
+            .unwrap()
+            .expect("a WAL-backed table reports Some");
+        let bucket = &stats.buckets[0];
+        assert_eq!(bucket.replay_after_wal_entry_position, 100);
+        assert_eq!(bucket.wal_entry_position_last_seen, 140);
+        assert!(!bucket.compacting);
+        assert_eq!(bucket.generations[0].generation, 8);
+        assert_eq!(bucket.generations[0].rows, Some(30));
+        // The line that answers "why is my fresh-tier vector search
+        // brute-force" — an absent index name is the whole explanation.
+        let memtables = bucket.memtables.as_ref().unwrap();
+        assert_eq!(memtables[0].indexes, vec!["vec_idx".to_string()]);
+    }
+
+    /// A 404 arrives as `TableNotFound`, not as a lost claim the loop
+    /// re-issues from flush until its cap. The two are distinguished by
+    /// status: 404 is "no such table", 421 is "this node holds no claim".
+    /// They shared 404 once, and the loop chased a name that never existed.
+    #[tokio::test(start_paused = true)]
+    async fn test_missing_table_is_not_read_as_a_lost_claim() {
+        let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let seen = calls.clone();
+        let table = Table::new_with_handler("my_table", move |_request| {
+            seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+            http::Response::builder()
+                .status(404)
+                .body(r#"{"code":4,"error":"Not found: Table not found: my_table"}"#.to_string())
+                .unwrap()
+        });
+
+        let err = table.checkpoint_lsm().await.unwrap_err();
+        assert!(
+            matches!(err, Error::TableNotFound { .. }),
+            "a missing table must say so: {err:?}"
+        );
+        assert_eq!(
+            calls.load(std::sync::atomic::Ordering::SeqCst),
+            1,
+            "no point re-claiming a table that does not exist"
+        );
+    }
+
+    /// A lost claim — 421, not 404 — does re-issue from flush, the call that
+    /// re-claims and replays.
+    #[tokio::test(start_paused = true)]
+    async fn test_registry_miss_reissues_from_flush() {
+        let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let seen = calls.clone();
+        let table = Table::new_with_handler("my_table", move |request| {
+            let path = request.url().path().to_string();
+            let n = seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+            if path.contains("flush_lsm") {
+                // First flush lands; the claim is then lost, and the
+                // re-issued flush succeeds.
+                return accepted();
+            }
+            if path.contains("compact_lsm") {
+                if n < 4 {
+                    return http::Response::builder()
+                        .status(421)
+                        .body(r#"{"code":19,"error":"table not claimed"}"#.to_string())
+                        .unwrap();
+                }
+                return accepted();
+            }
+            ok_json(stats_body(if n < 6 { &[1] } else { &[] }, false))
+        });
+
+        table
+            .checkpoint_lsm()
+            .await
+            .expect("a lost claim must be recovered by re-flushing, not surfaced");
+    }
+
+    #[tokio::test]
+    async fn test_get_lsm_stats_absent_when_wal_off() {
+        let table = Table::new_with_handler("my_table", |_request| {
+            http::Response::builder()
+                .status(200)
+                .body(serde_json::json!({ "lsm_stats": null }).to_string())
+                .unwrap()
+        });
+        assert!(table.get_lsm_stats(false).await.unwrap().is_none());
+    }
+
    #[tokio::test]
    async fn test_wait_for_index() {
        let table = _make_table_with_indices(0);
@@ -68,10 +68,12 @@ use self::merge::MergeInsertBuilder;
 pub mod add_columns;
 mod add_data;
 pub mod branch_merge;
+pub mod checkpoint;
 mod create_index;
 pub mod datafusion;
 pub(crate) mod dataset;
 pub mod delete;
+pub mod lsm_stats;
 pub mod merge;
 pub mod optimize;
 mod primary_key;
@@ -93,8 +95,8 @@ pub use delete::DeleteResult;
 use futures::future::join_all;
 pub use lance::dataset::refs::{BranchContents, Ref, TagContents, Tags as LanceTags};
 pub use lance::dataset::scanner::DatasetRecordBatchStream;
-use lance::dataset::statistics::DatasetStatisticsExt;
 pub use lance_index::optimize::OptimizeOptions;
+pub use lsm_stats::{BucketStats, GenerationStats, LsmStats, MemtableStats};
 pub use optimize::{CompactionOptions, OptimizeAction, OptimizeStats};
 pub use schema_evolution::{
    AddColumnsResult, AlterColumnsResult, DropColumnsResult, FieldMetadataUpdate,
@@ -368,6 +370,8 @@ pub use self::merge::MergeResult;
 /// date) and [`LsmWriteSpec::with_writer_config_defaults`] (default
 /// `ShardWriter` configuration recorded in the MemWAL index).
 ///
+/// A fresh spec maintains every index on the table, resolved on install.
+///
 /// Install a spec with [`Table::set_lsm_write_spec`] and remove it with
 /// [`Table::unset_lsm_write_spec`]. The actual `merge_insert` dispatch
 /// onto the MemWAL writer is a follow-up.
@@ -382,9 +386,12 @@ pub enum LsmWriteSpec {
    Bucket {
        column: String,
        num_buckets: u32,
-        /// Names of indexes (already created on the table) that the
-        /// MemWAL should maintain in-memory as rows are appended.
-        maintained_indexes: Vec<String>,
+        /// Indexes the MemWAL maintains in-memory as rows are appended.
+        ///
+        /// `None` means every index it can maintain, resolved on install — a
+        /// snapshot, so indexes created later need the spec unset and re-set.
+        /// `Some([])` maintains nothing.
+        maintained_indexes: Option<Vec<String>>,
        /// Default `ShardWriter` configuration recorded in the MemWAL index.
        writer_config_defaults: HashMap<String, String>,
    },
@@ -394,35 +401,41 @@ pub enum LsmWriteSpec {
    /// distinct value of `column` becomes its own shard.
    Identity {
        column: String,
-        /// Names of indexes (already created on the table) that the
-        /// MemWAL should maintain in-memory as rows are appended.
-        maintained_indexes: Vec<String>,
+        /// Indexes the MemWAL maintains in-memory as rows are appended.
+        ///
+        /// `None` means every index it can maintain, resolved on install — a
+        /// snapshot, so indexes created later need the spec unset and re-set.
+        /// `Some([])` maintains nothing.
+        maintained_indexes: Option<Vec<String>>,
        /// Default `ShardWriter` configuration recorded in the MemWAL index.
        writer_config_defaults: HashMap<String, String>,
    },
    /// No sharding — every `merge_insert` call writes to a single MemWAL shard.
    Unsharded {
-        /// Names of indexes (already created on the table) that the
-        /// MemWAL should maintain in-memory as rows are appended.
-        maintained_indexes: Vec<String>,
+        /// Indexes the MemWAL maintains in-memory as rows are appended.
+        ///
+        /// `None` means every index it can maintain, resolved on install — a
+        /// snapshot, so indexes created later need the spec unset and re-set.
+        /// `Some([])` maintains nothing.
+        maintained_indexes: Option<Vec<String>>,
        /// Default `ShardWriter` configuration recorded in the MemWAL index.
        writer_config_defaults: HashMap<String, String>,
    },
 }

 impl LsmWriteSpec {
-    /// Construct a hash-bucket sharding spec with no maintained indexes.
+    /// Construct a hash-bucket sharding spec maintaining every index on the table.
    pub fn bucket(column: impl Into<String>, num_buckets: u32) -> Self {
        Self::Bucket {
            column: column.into(),
            num_buckets,
-            maintained_indexes: Vec::new(),
+            maintained_indexes: None,
            writer_config_defaults: HashMap::new(),
        }
    }

    /// Construct an identity-sharding spec (shard by the raw value of
-    /// `column`) with no maintained indexes.
+    /// `column`) maintaining every index on the table.
    ///
    /// `column` must be a deterministic function of the unenforced primary
    /// key: every row with a given primary key must always produce the same
@@ -434,28 +447,37 @@ impl LsmWriteSpec {
    pub fn identity(column: impl Into<String>) -> Self {
        Self::Identity {
            column: column.into(),
-            maintained_indexes: Vec::new(),
+            maintained_indexes: None,
            writer_config_defaults: HashMap::new(),
        }
    }

-    /// Construct an unsharded spec with no maintained indexes.
+    /// Construct an unsharded spec maintaining every index on the table.
    pub fn unsharded() -> Self {
        Self::Unsharded {
-            maintained_indexes: Vec::new(),
+            maintained_indexes: None,
            writer_config_defaults: HashMap::new(),
        }
    }

-    /// Replace the list of indexes the MemWAL should keep up to date as
-    /// rows are appended. Each name must reference an index that already
-    /// exists on the table at the time `set_lsm_write_spec` is called.
-    pub fn with_maintained_indexes<I, S>(mut self, indexes: I) -> Self
-    where
-        I: IntoIterator<Item = S>,
-        S: Into<String>,
-    {
-        let v: Vec<String> = indexes.into_iter().map(Into::into).collect();
+    /// Set which indexes the MemWAL maintains.
+    ///
+    /// `None` (the default) resolves to every index on the table at install,
+    /// failing if one cannot be maintained — name the set to install anyway. A
+    /// list is verbatim: each name must already exist and be maintainable, and
+    /// an empty list maintains nothing.
+    ///
+    /// ```
+    /// # use lancedb::table::LsmWriteSpec;
+    /// // Every index the table has when the spec is installed:
+    /// LsmWriteSpec::unsharded().with_maintained_indexes(None);
+    /// // Exactly these:
+    /// LsmWriteSpec::unsharded().with_maintained_indexes(vec!["id_idx".to_string()]);
+    /// // None at all:
+    /// LsmWriteSpec::unsharded().with_maintained_indexes(Vec::new());
+    /// ```
+    pub fn with_maintained_indexes(mut self, indexes: impl Into<Option<Vec<String>>>) -> Self {
+        let indexes = indexes.into();
        match &mut self {
            Self::Bucket {
                maintained_indexes, ..
@@ -465,7 +487,7 @@ impl LsmWriteSpec {
            }
            | Self::Unsharded {
                maintained_indexes, ..
-            } => *maintained_indexes = v,
+            } => *maintained_indexes = indexes,
        }
        self
    }
@@ -501,8 +523,9 @@ impl LsmWriteSpec {
        self
    }

-    /// Borrow the list of index names this spec asks MemWAL to maintain.
-    pub fn maintained_indexes(&self) -> &[String] {
+    /// Borrow the list of index names this spec asks MemWAL to maintain, or
+    /// `None` when it asks for every index on the table.
+    pub fn maintained_indexes(&self) -> Option<&[String]> {
        match self {
            Self::Bucket {
                maintained_indexes, ..
@@ -512,7 +535,7 @@ impl LsmWriteSpec {
            }
            | Self::Unsharded {
                maintained_indexes, ..
-            } => maintained_indexes,
+            } => maintained_indexes.as_deref(),
        }
    }

@@ -685,6 +708,31 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
            message: "get_lsm_write_spec is not supported on this table type".into(),
        })
    }
+    /// Seal every bucket's active memtable into L0.
+    ///
+    /// The default implementation returns `NotSupported`.
+    async fn flush_lsm(&self) -> Result<()> {
+        Err(Error::NotSupported {
+            message: "flush_lsm is not supported on this table type".into(),
+        })
+    }
+    /// Trigger a background L0 → base compaction pass per bucket.
+    ///
+    /// The default implementation returns `NotSupported`.
+    async fn compact_lsm(&self) -> Result<()> {
+        Err(Error::NotSupported {
+            message: "compact_lsm is not supported on this table type".into(),
+        })
+    }
+    /// Read live LSM state, or `None` when the LSM write path is not
+    /// enabled for this table.
+    ///
+    /// The default implementation returns `NotSupported`.
+    async fn get_lsm_stats(&self, _include_generation_rows: bool) -> Result<Option<LsmStats>> {
+        Err(Error::NotSupported {
+            message: "get_lsm_stats is not supported on this table type".into(),
+        })
+    }
    /// Drain and close any cached MemWAL shard writers for this table.
    ///
    /// The default implementation is a no-op; table types that maintain
@@ -1685,7 +1733,7 @@ impl Table {
    /// # async fn example(table: &Table) -> Result<(), Box<dyn std::error::Error>> {
    /// table
    ///     .set_lsm_write_spec(
-    ///         LsmWriteSpec::bucket("id", 16).with_maintained_indexes(["id_idx"]),
+    ///         LsmWriteSpec::bucket("id", 16).with_maintained_indexes(vec!["id_idx".to_string()]),
    ///     )
    ///     .await?;
    /// # Ok(())
@@ -1707,9 +1755,10 @@ impl Table {
    ///
    /// Returns `Ok(None)` when the MemWAL LSM write path is not enabled (no
    /// spec has been set, or it was removed with [`Table::unset_lsm_write_spec`]).
-    /// The returned spec — including its [`LsmWriteSpec::maintained_indexes`] and
-    /// [`LsmWriteSpec::writer_config_defaults`] — mirrors what was passed to
-    /// [`Table::set_lsm_write_spec`].
+    /// The returned spec mirrors what was passed to
+    /// [`Table::set_lsm_write_spec`], except that
+    /// [`LsmWriteSpec::maintained_indexes`] always reports the concrete list
+    /// resolved when the spec was set — `None` never round-trips.
    ///
    /// # Example
    ///
@@ -1726,6 +1775,85 @@ impl Table {
        self.inner.get_lsm_write_spec().await
    }

+    /// Converge this table's LSM write path into its base table.
+    ///
+    /// One `flush` to seal every memtable into L0, then compaction triggers
+    /// until every generation that existed at that moment has reached base.
+    /// The loop runs client-side, reading progress from `get_lsm_stats`, so
+    /// there is no held socket and nothing to reconcile if you drop this
+    /// future partway through.
+    ///
+    /// **Best-effort.** Generations created *after* the opening flush are
+    /// deliberately not waited on — that is what lets this terminate on a
+    /// table taking writes. Idempotent and safe on a cadence: an
+    /// already-converged table costs two round trips and triggers nothing.
+    ///
+    /// **No deadline, and the caller owns that.** It returns when the target
+    /// generations are gone, propagates a terminal server fault, and
+    /// otherwise waits however long the server takes. A slow table and a
+    /// stuck one are the same picture from here: the compactor pool is shared
+    /// across every table on the node, so a checkpoint queued behind
+    /// unrelated work is indistinguishable from one that is merging. Wrap
+    /// this in `tokio::time::timeout` for a wall-clock bound; abandoning it
+    /// partway costs nothing.
+    ///
+    /// # Example
+    ///
+    /// ```no_run
+    /// # use lancedb::Table;
+    /// # async fn example(table: &Table) -> Result<(), Box<dyn std::error::Error>> {
+    /// let before = table.get_lsm_stats(false).await?;
+    /// table.checkpoint_lsm().await?;
+    /// let after = table.get_lsm_stats(false).await?;
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub async fn checkpoint_lsm(&self) -> Result<()> {
+        checkpoint::checkpoint_lsm(self).await
+    }
+
+    /// Seal every bucket's active memtable into L0 without touching the
+    /// base table.
+    ///
+    /// Independently useful: flushing makes memtable rows readable from L0 at
+    /// a lower per-query cost. On a node that has not claimed this table it
+    /// claims it and replays the WAL log first — reporting "nothing to flush"
+    /// without replaying would lie about durable data.
+    pub async fn flush_lsm(&self) -> Result<()> {
+        self.inner.flush_lsm().await
+    }
+
+    /// Run one bounded L0 → base compaction pass per bucket, reporting what
+    /// it merged and what is left.
+    ///
+    /// One pass, not convergence: that bounds each request's cost and gives a
+    /// caller driving its own cadence a progress signal per round trip.
+    pub async fn compact_lsm(&self) -> Result<()> {
+        self.inner.compact_lsm().await
+    }
+
+    /// Read live per-bucket LSM state.
+    ///
+    /// Answers "how far behind is my fresh tier", "which bucket is hot", and
+    /// "why is my fresh-tier vector search brute-force". Mutates no table
+    /// state, though on a node that has not claimed this table it claims it,
+    /// exactly as a read would.
+    ///
+    /// `include_generation_rows` reports a row count per L0 generation. Off by
+    /// default: each count opens an uncached Lance dataset, and
+    /// `checkpoint_lsm` polls this needing only generation numbers.
+    ///
+    /// `Ok(None)` only when the LSM write path is not enabled, matching
+    /// [`Table::get_lsm_write_spec`]. Stats is fresh-tier only, so with the
+    /// WAL off there is no manifest to report and a struct of zeros would
+    /// read as measurements.
+    ///
+    /// Do not build a checkpoint's termination on this: the completion
+    /// predicate lives in the `flush` and `compact` responses.
+    pub async fn get_lsm_stats(&self, include_generation_rows: bool) -> Result<Option<LsmStats>> {
+        self.inner.get_lsm_stats(include_generation_rows).await
+    }
+
    /// Drain and close any cached MemWAL shard writers held for this table.
    ///
    /// When an [`LsmWriteSpec`] is installed, `merge_insert` opens MemWAL shard
@@ -3441,9 +3569,24 @@ impl BaseTable for NativeTable {
        let num_rows = self.count_rows(None).await?;
        let num_indices = self.list_indices().await?.len();
        let ds = self.dataset.get().await?;
-        let ds_clone = (*ds).clone();
-        let ds_stats = Arc::new(ds_clone).calculate_data_stats().await?;
-        let total_bytes = ds_stats.fields.iter().map(|f| f.bytes_on_disk).sum::<u64>() as usize;
+        // Sizes come from the manifest. Summing per-field `bytes_on_disk` instead
+        // would open every data file to read its column metadata, which costs one
+        // IO per fragment and reports 0 for legacy v1 storage.
+        //
+        // The manifest summary covers only the fragments' base data files, so
+        // overlay files (recorded on each fragment) and index files (recorded in
+        // the manifest's index section) are added separately.
+        let mut total_bytes = ds.manifest().summary().total_files_size as usize;
+        for frag in ds.manifest().fragments.iter() {
+            for overlay in &frag.overlays {
+                if let Some(size) = overlay.data_file.file_size_bytes.get() {
+                    total_bytes += size.get() as usize;
+                }
+            }
+        }
+        for index in ds.load_indices().await?.iter() {
+            total_bytes += index.total_size_bytes().unwrap_or(0) as usize;
+        }

        let frags = ds.get_fragments();
        let mut sorted_sizes = join_all(
@@ -3515,7 +3658,12 @@ impl BaseTable for NativeTable {
 #[skip_serializing_none]
 #[derive(Debug, Deserialize, PartialEq)]
 pub struct TableStatistics {
-    /// The total number of bytes in the table
+    /// The total size, in bytes, of the table's data files, index files, and
+    /// overlay files
+    ///
+    /// Read from the manifest, so this excludes deletion files and manifests,
+    /// and it excludes any file whose size the manifest does not record
+    /// (tables and indices written before writers persisted file sizes).
    pub total_bytes: usize,

    /// The number of rows in the table
@@ -3576,6 +3724,7 @@ mod tests {
    use super::*;
    use crate::connect;
    use crate::connection::ConnectBuilder;
+    use crate::io::object_store::io_tracking::IoTrackingStore;
    use crate::query::Select;
    use crate::query::{ExecutableQuery, QueryBase};
    use crate::test_utils::connection::new_test_connection;
@@ -4958,7 +5107,7 @@ mod tests {
        // Bucket spec round-trips exactly, including the routing column (recovered
        // from its field id), maintained indexes, and writer config defaults.
        let spec = LsmWriteSpec::bucket("id", 4)
-            .with_maintained_indexes([idx_name])
+            .with_maintained_indexes(vec![idx_name.clone()])
            .with_writer_config_defaults([("durable_write", "false")]);
        table.set_lsm_write_spec(spec.clone()).await.unwrap();
        assert_eq!(table.get_lsm_write_spec().await.unwrap(), Some(spec));
@@ -4968,15 +5117,125 @@ mod tests {
        assert_eq!(table.get_lsm_write_spec().await.unwrap(), None);

        // Identity sharding round-trips (column recovered from the schema).
+        // A spec left at its default maintains every index on the table, so it
+        // reads back naming the one on the table rather than as "infer".
        let spec = LsmWriteSpec::identity("region");
        table.set_lsm_write_spec(spec.clone()).await.unwrap();
-        assert_eq!(table.get_lsm_write_spec().await.unwrap(), Some(spec));
+        assert_eq!(
+            table.get_lsm_write_spec().await.unwrap(),
+            Some(spec.with_maintained_indexes(vec![idx_name.clone()]))
+        );
        table.unset_lsm_write_spec().await.unwrap();

        // Unsharded round-trips (no routing column).
        let spec = LsmWriteSpec::unsharded();
        table.set_lsm_write_spec(spec.clone()).await.unwrap();
-        assert_eq!(table.get_lsm_write_spec().await.unwrap(), Some(spec));
+        assert_eq!(
+            table.get_lsm_write_spec().await.unwrap(),
+            Some(spec.with_maintained_indexes(vec![idx_name]))
+        );
+    }
+
+    /// The maintained set defaults to every index on the table, resolved at
+    /// install. An index the memtable cannot build fails the install rather
+    /// than being dropped: maintaining it would take the table offline for
+    /// writes, dropping it would hide that from the caller.
+    #[tokio::test]
+    async fn test_set_lsm_write_spec_infers_maintained_indexes() {
+        let tmp_dir = tempdir().unwrap();
+        let uri = tmp_dir.path().to_str().unwrap();
+
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("id", DataType::Int64, false),
+            Field::new("tag", DataType::Utf8, true),
+        ]));
+        let batch = RecordBatch::try_new(
+            schema.clone(),
+            vec![
+                Arc::new(arrow_array::Int64Array::from(vec![1, 2, 3])),
+                Arc::new(StringArray::from(vec!["a", "b", "c"])),
+            ],
+        )
+        .unwrap();
+        let reader: Box<dyn arrow_array::RecordBatchReader + Send> =
+            Box::new(RecordBatchIterator::new(vec![Ok(batch)], schema.clone()));
+        let conn = ConnectBuilder::new(uri)
+            .read_consistency_interval(Duration::from_secs(0))
+            .execute()
+            .await
+            .unwrap();
+        let table = conn.create_table("t", reader).execute().await.unwrap();
+
+        table
+            .create_index(&["id"], Index::BTree(Default::default()))
+            .name("id_btree".to_string())
+            .execute()
+            .await
+            .unwrap();
+        table
+            .create_index(&["tag"], Index::Bitmap(Default::default()))
+            .name("tag_bitmap".to_string())
+            .execute()
+            .await
+            .unwrap();
+
+        // Explicitly naming the bitmap index fails before anything commits.
+        let err = table
+            .set_lsm_write_spec(
+                LsmWriteSpec::unsharded().with_maintained_indexes(vec!["tag_bitmap".to_string()]),
+            )
+            .await
+            .unwrap_err();
+        assert!(
+            matches!(err, Error::InvalidInput { ref message } if message.contains("tag_bitmap")),
+            "expected the bitmap index to be rejected, got {err:?}"
+        );
+        assert_eq!(table.get_lsm_write_spec().await.unwrap(), None);
+
+        // The default covers every index, so the bitmap fails it too.
+        let err = table
+            .set_lsm_write_spec(LsmWriteSpec::unsharded())
+            .await
+            .unwrap_err();
+        assert!(
+            matches!(err, Error::InvalidInput { ref message }
+                if message.contains("tag_bitmap") && message.contains("maintained_indexes")),
+            "expected the inferred set to be rejected, got {err:?}"
+        );
+        assert_eq!(table.get_lsm_write_spec().await.unwrap(), None);
+
+        // Naming the maintainable subset installs.
+        table
+            .set_lsm_write_spec(
+                LsmWriteSpec::unsharded().with_maintained_indexes(vec!["id_btree".to_string()]),
+            )
+            .await
+            .unwrap();
+        assert_eq!(
+            table
+                .get_lsm_write_spec()
+                .await
+                .unwrap()
+                .unwrap()
+                .maintained_indexes(),
+            Some(["id_btree".to_string()].as_slice())
+        );
+
+        // Opting out entirely is distinct from the default.
+        table.unset_lsm_write_spec().await.unwrap();
+        table
+            .set_lsm_write_spec(LsmWriteSpec::unsharded().with_maintained_indexes(Vec::new()))
+            .await
+            .unwrap();
+        assert_eq!(
+            table
+                .get_lsm_write_spec()
+                .await
+                .unwrap()
+                .unwrap()
+                .maintained_indexes(),
+            Some([].as_slice())
+        );
    }

    #[tokio::test]
@@ -5024,12 +5283,16 @@ mod tests {

        let res = table.stats().await.unwrap();
        println!("{:#?}", res);
+        // `total_bytes` is the full on-disk size of the 11 data files (this table
+        // has no index or overlay files), so it is well above the 2000 bytes of
+        // column data these 250 int32 pairs hold: each file carries its own footer
+        // and metadata.
        assert_eq!(
            res,
            TableStatistics {
                num_rows: 250,
                num_indices: 0,
-                total_bytes: 2300,
+                total_bytes: 8925,
                fragment_stats: FragmentStatistics {
                    num_fragments: 11,
                    num_small_fragments: 11,
@@ -5069,4 +5332,196 @@ mod tests {
            }
        )
    }
+
+    /// `total_bytes` counts more than the base data files: index files and
+    /// overlay files recorded in the manifest are included too.
+    #[tokio::test]
+    pub async fn test_stats_includes_index_and_overlay_files() {
+        use lance::dataset::WriteDestination;
+        use lance::dataset::transaction::{DataOverlayGroup, Operation};
+        use lance_file::version::{ConcreteFileVersion, LanceFileVersion};
+        use lance_file::writer::FileWriterOptions;
+        use lance_io::utils::CachedFileSize;
+        use lance_table::format::DataFile;
+        use lance_table::format::overlay::{DataOverlayFile, OverlayCoverage};
+        use roaring::RoaringBitmap;
+
+        let tmp_dir = tempdir().unwrap();
+        let uri = tmp_dir.path().to_str().unwrap();
+        let conn = ConnectBuilder::new(uri)
+            .read_consistency_interval(Duration::from_secs(0))
+            .execute()
+            .await
+            .unwrap();
+
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("id", DataType::Int32, false),
+            Field::new("foo", DataType::Int32, true),
+        ]));
+        let batch = RecordBatch::try_new(
+            schema.clone(),
+            vec![
+                Arc::new(Int32Array::from_iter_values(0..100)),
+                Arc::new(Int32Array::from_iter_values(0..100)),
+            ],
+        )
+        .unwrap();
+        let table = conn
+            .create_table("test_stats_extra_files", batch)
+            .execute()
+            .await
+            .unwrap();
+
+        let data_only = table.stats().await.unwrap().total_bytes;
+        assert!(data_only > 0);
+
+        // A scalar index adds index files whose sizes are recorded in the
+        // manifest's index section.
+        table
+            .create_index(&["id"], Index::Auto)
+            .execute()
+            .await
+            .unwrap();
+        let with_index = table.stats().await.unwrap().total_bytes;
+        let dataset = {
+            let native = table.as_native().unwrap();
+            (*native.dataset.get().await.unwrap()).clone()
+        };
+        let index_bytes: usize = dataset
+            .load_indices()
+            .await
+            .unwrap()
+            .iter()
+            .map(|idx| idx.total_size_bytes().unwrap_or(0) as usize)
+            .sum();
+        assert!(index_bytes > 0);
+        assert_eq!(with_index, data_only + index_bytes);
+
+        // Commit an overlay file supplying new `foo` values for the first three
+        // rows of fragment 0. There is no high-level API that writes overlays
+        // yet, so write the overlay's data file and commit the `DataOverlay`
+        // operation by hand.
+        let read_version = dataset.version().version;
+        let fragment_id = dataset.get_fragments()[0].id() as u64;
+        let foo_field_id = dataset.schema().field("foo").unwrap().id;
+        let overlay_schema = dataset.schema().project_by_ids(&[foo_field_id], true);
+        let file_version = ConcreteFileVersion::from(LanceFileVersion::Stable);
+
+        let filename = "overlay.lance".to_string();
+        let store = dataset.object_store(None).await.unwrap();
+        let path = dataset.data_dir().child(filename.clone());
+        let obj_writer = store.create(&path).await.unwrap();
+        let mut writer = lance_file::versions::create_writer(
+            file_version,
+            obj_writer,
+            overlay_schema,
+            FileWriterOptions::default(),
+        )
+        .unwrap();
+        writer
+            .write_column(0, Arc::new(Int32Array::from(vec![1000, 1001, 1002])) as _)
+            .await
+            .unwrap();
+        let summary = writer.finish().await.unwrap();
+        let overlay_bytes = summary.size_bytes as usize;
+        assert!(overlay_bytes > 0);
+
+        let mut data_file = DataFile::new_unstarted(filename, file_version);
+        data_file.fields = writer
+            .field_id_to_column_indices()
+            .iter()
+            .map(|(field_id, _)| *field_id as i32)
+            .collect::<Vec<_>>()
+            .into();
+        data_file.column_indices = writer
+            .field_id_to_column_indices()
+            .iter()
+            .map(|(_, column_index)| *column_index as i32)
+            .collect::<Vec<_>>()
+            .into();
+        data_file.file_size_bytes = CachedFileSize::new(summary.size_bytes);
+
+        let overlay = DataOverlayFile {
+            data_file,
+            coverage: OverlayCoverage::dense(RoaringBitmap::from_iter(0..3)),
+            committed_version: 0,
+        };
+        Dataset::commit(
+            WriteDestination::Dataset(Arc::new(dataset)),
+            Operation::DataOverlay {
+                groups: vec![DataOverlayGroup {
+                    fragment_id,
+                    overlays: vec![overlay],
+                }],
+            },
+            Some(read_version),
+            None,
+            None,
+            Arc::new(Default::default()),
+            false,
+        )
+        .await
+        .unwrap();
+
+        table.checkout_latest().await.unwrap();
+        let with_overlay = table.stats().await.unwrap().total_bytes;
+        assert_eq!(with_overlay, with_index + overlay_bytes);
+    }
+
+    /// `stats()` must stay manifest-only. Summing per-field `bytes_on_disk`
+    /// instead opens every data file, so cost would grow with fragment count.
+    #[tokio::test]
+    pub async fn test_stats_does_not_read_data_files() {
+        let tmp_dir = tempdir().unwrap();
+        let uri = tmp_dir.path().to_str().unwrap();
+
+        let conn = ConnectBuilder::new(uri).execute().await.unwrap();
+
+        let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
+        let batch = RecordBatch::try_new(
+            schema.clone(),
+            vec![Arc::new(Int32Array::from_iter_values(0..10))],
+        )
+        .unwrap();
+
+        conn.create_table("test_stats_io", batch.clone())
+            .execute()
+            .await
+            .unwrap();
+        let table = conn.open_table("test_stats_io").execute().await.unwrap();
+        const NUM_APPENDS: usize = 20;
+        for _ in 0..NUM_APPENDS {
+            table.add(batch.clone()).execute().await.unwrap();
+        }
+
+        // Reopen through a tracking store so the counters cover `stats()` alone and
+        // not the writes above.
+        let (wrapper, io_stats) = IoTrackingStore::new_wrapper();
+        let table = conn
+            .open_table("test_stats_io")
+            .lance_read_params(ReadParams {
+                store_options: Some(ObjectStoreParams {
+                    object_store_wrapper: Some(wrapper),
+                    ..Default::default()
+                }),
+                ..Default::default()
+            })
+            .execute()
+            .await
+            .unwrap();
+        io_stats.lock().unwrap().read_iops = 0;
+
+        let stats = table.stats().await.unwrap();
+        let read_iops = io_stats.lock().unwrap().read_iops;
+
+        assert_eq!(stats.fragment_stats.num_fragments, NUM_APPENDS + 1);
+        assert!(stats.total_bytes > 0);
+        // Reading the fragments' data files would take at least one IOP each.
+        assert!(
+            read_iops < stats.fragment_stats.num_fragments as u64,
+            "stats() issued {} read IOPs across {} fragments",
+            read_iops,
+            stats.fragment_stats.num_fragments
+        );
+    }
 }
@@ -0,0 +1,315 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+//! Converging a table's LSM write path into its base table.
+//!
+//! `checkpoint_lsm` seals once, then triggers compaction and watches
+//! generation numbers until the L0 that existed at the start is gone.
+//!
+//! The loop runs in the client, not the server: `compact_lsm` dispatches a
+//! pass and returns, so nothing holds a socket and a client can vanish
+//! mid-operation with nothing to reconcile. Completion is read from
+//! generation numbers in the shard manifest — durable state, unlike a count
+//! in a compact response, which a concurrent write invalidates.
+//!
+//! The target set is fixed at the start, so generations created *during* the
+//! checkpoint are ignored. That is what lets it terminate under write load,
+//! and what makes it best-effort: it converges the fresh tier as of some
+//! instant. Idempotent, abandonable at any point, safe on a cadence.
+//!
+//! No liveness bound — the caller owns the deadline. The compactor pool is
+//! shared pod-wide, so a checkpoint queued behind unrelated tables looks
+//! exactly like one that is merging.
+
+use std::collections::HashMap;
+use std::future::Future;
+use std::time::Duration;
+
+use crate::{Error, Result, Table};
+
+/// The HTTP status a failed request carried, if it carried one.
+///
+/// `None` for anything with no retry story: a `TableNotFound` that
+/// `check_table_response` already translated, or a connection failure that
+/// never reached the server. Both are terminal.
+fn status_of(e: &Error) -> Option<u16> {
+    #[cfg(feature = "remote")]
+    {
+        match e {
+            Error::Http {
+                status_code: Some(status),
+                ..
+            } => Some(status.as_u16()),
+            _ => None,
+        }
+    }
+    #[cfg(not(feature = "remote"))]
+    {
+        let _ = e;
+        None
+    }
+}
+
+/// 429 (latch held, pool saturated, or the pod replaying its WAL) and 503 (a
+/// draining node, or a proxy between here and it).
+///
+/// The status is the whole signal: the server deliberately keeps contention
+/// off 503, so a latch collision is a 429. A draining node *is* terminal, but
+/// it is also a 503 that stays a 503, so retrying spends one budget and then
+/// reports the server's own message — cheaper than parsing the body for the
+/// namespace code it would take to tell the two apart.
+fn is_retryable(e: &Error) -> bool {
+    matches!(status_of(e), Some(429 | 503))
+}
+
+/// 421: the owning node holds no claim. Only `flush` re-claims and replays,
+/// so this cannot be retried in place — the caller has to start over.
+fn is_lost_claim(e: &Error) -> bool {
+    status_of(e) == Some(421)
+}
+
+/// Interval between `get_lsm_stats` polls. One interval is roughly one
+/// compaction pass, the granularity at which the answer can change.
+///
+/// Fixed rather than configurable, matching `wait_for_index`. It costs
+/// nothing on an already-converged table and at most one interval of tail
+/// latency after the final pass lands.
+const POLL_INTERVAL: Duration = Duration::from_secs(5);
+
+/// Cap on re-issues from `flush` after a 421, so a crash-looping node cannot
+/// turn flush → compact → 421 → flush into a spin.
+///
+/// Deliberately not shared with [`MAX_RETRIES`]: a claim that keeps
+/// evaporating is a broken node, while contention is routine and wants a real
+/// budget. One shared counter let a merely contended table exhaust this cap
+/// and then blame a claim it never lost.
+const MAX_REISSUES: usize = 3;
+
+/// Retryable faults tolerated on a *single* request, reset on every success —
+/// scattered contention across a long checkpoint must not accumulate toward a
+/// cap. Roughly 16s of retrying against the backoff below.
+const MAX_RETRIES: usize = 8;
+
+/// Backoff between retries, doubling up to [`RETRY_BACKOFF_MAX`]. Latch
+/// contention clears in about the time one pass takes, so start small; a
+/// saturated pool wants the ceiling.
+const RETRY_BACKOFF_BASE: Duration = Duration::from_millis(100);
+const RETRY_BACKOFF_MAX: Duration = Duration::from_secs(5);
+
+/// Sleep before re-issuing a retryable request.
+async fn backoff(attempt: usize) {
+    let delay = RETRY_BACKOFF_BASE
+        .saturating_mul(1u32 << attempt.min(8) as u32)
+        .min(RETRY_BACKOFF_MAX);
+    tokio::time::sleep(delay).await;
+}
+
+/// Whether the drain loop finished or needs the table re-claimed first.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum CheckpointOutcome {
+    Done,
+    ReissueFromFlush,
+}
+
+/// What one LSM request produced: its value, or word that the owning node
+/// holds no claim and only `flush` can get it back.
+enum Attempt<T> {
+    Ok(T),
+    ReissueFromFlush,
+}
+
+/// Issue one LSM request, retrying in place while the fault is retryable.
+///
+/// The two recoverable faults have separate budgets: contention clears on its
+/// own and retries here against [`MAX_RETRIES`], while a 421 needs `flush` to
+/// re-claim, which only the caller can drive.
+///
+/// An exhausted budget propagates the last error *as itself* rather than a
+/// synthesized one — "429 after nine tries" beats "checkpoint failed", and a
+/// draining node arrives carrying the server's own message.
+async fn issue<T, F, Fut>(mut call: F) -> Result<Attempt<T>>
+where
+    F: FnMut() -> Fut,
+    Fut: Future<Output = Result<T>>,
+{
+    let mut retries = 0;
+    loop {
+        let e = match call().await {
+            Ok(value) => return Ok(Attempt::Ok(value)),
+            Err(e) => e,
+        };
+        if is_lost_claim(&e) {
+            return Ok(Attempt::ReissueFromFlush);
+        }
+        if !is_retryable(&e) || retries >= MAX_RETRIES {
+            return Err(e);
+        }
+        backoff(retries).await;
+        retries += 1;
+    }
+}
+
+/// Drive [`Table::checkpoint_lsm`]: seal once, fix the target watermark
+/// from the resulting L0, then trigger and poll until it drains.
+pub(crate) async fn checkpoint_lsm(table: &Table) -> Result<()> {
+    for reissue in 0..=MAX_REISSUES {
+        // The seal turns everything written before this call into a
+        // generation, so the watermark has to be read after it. Idempotent:
+        // sealing an empty memtable is a no-op, so a re-issue does not churn
+        // empty generations.
+        match issue(|| table.flush_lsm()).await? {
+            Attempt::Ok(()) => {}
+            Attempt::ReissueFromFlush => {
+                backoff(reissue).await;
+                continue;
+            }
+        }
+
+        let stats = match issue(|| table.get_lsm_stats(false)).await? {
+            Attempt::Ok(stats) => stats,
+            Attempt::ReissueFromFlush => {
+                backoff(reissue).await;
+                continue;
+            }
+        };
+        let Some(stats) = stats else {
+            // Not WAL-backed; `flush_lsm` would have errored first but for a race.
+            return Ok(());
+        };
+        let targets: HashMap<String, u64> = stats
+            .buckets
+            .iter()
+            .filter_map(|b| Some((b.shard_id.clone(), b.newest_generation()?)))
+            .collect();
+        if targets.is_empty() {
+            return Ok(());
+        }
+
+        match drain_to_targets(table, &targets).await? {
+            CheckpointOutcome::Done => return Ok(()),
+            CheckpointOutcome::ReissueFromFlush => {
+                backoff(reissue).await;
+                continue;
+            }
+        }
+    }
+    Err(Error::Runtime {
+        message: "checkpoint_lsm: the owning node kept losing its claim; \
+                  re-issued from flush the maximum number of times"
+            .into(),
+    })
+}
+
+/// Trigger and poll until no bucket holds a generation at or below its
+/// target.
+///
+/// No liveness bound, deliberately. The pod-wide compactor pool (a semaphore
+/// of 2 by default, shared across every table on the node) is taken *inside*
+/// the pass, after the bucket latch, so a checkpoint queued behind unrelated
+/// tables is indistinguishable from one that is merging. An idle-poll counter
+/// here could only ever have fired on a table that would have finished.
+async fn drain_to_targets(
+    table: &Table,
+    targets: &HashMap<String, u64>,
+) -> Result<CheckpointOutcome> {
+    loop {
+        let stats = match issue(|| table.get_lsm_stats(false)).await? {
+            Attempt::Ok(stats) => stats,
+            Attempt::ReissueFromFlush => return Ok(CheckpointOutcome::ReissueFromFlush),
+        };
+        let Some(stats) = stats else {
+            return Ok(CheckpointOutcome::Done);
+        };
+        // `compacting` is the bucket's compaction latch, held from dispatch
+        // until the pass ends — including while it waits on the pod-wide
+        // permit. So it answers one question only: do not pile on. Buckets
+        // with nothing outstanding are skipped, not counted as idle.
+        let mut outstanding = 0;
+        let mut all_compacting = true;
+        for b in &stats.buckets {
+            let Some(target) = targets.get(&b.shard_id) else {
+                continue;
+            };
+            let n = b.outstanding_generations(*target);
+            if n > 0 {
+                outstanding += n;
+                all_compacting &= b.compacting;
+            }
+        }
+        if outstanding == 0 {
+            return Ok(CheckpointOutcome::Done);
+        }
+
+        if !all_compacting {
+            match table.compact_lsm().await {
+                Ok(()) => {}
+                Err(e) if is_lost_claim(&e) => return Ok(CheckpointOutcome::ReissueFromFlush),
+                Err(e) if !is_retryable(&e) => return Err(e),
+                // A 429 here means the server could latch no bucket at all,
+                // which the poll above already handles. Not retried in place:
+                // the latch it would contend for is the one doing the work, so
+                // fall through and re-read — `POLL_INTERVAL` is the backoff.
+                Err(_) => {}
+            }
+        }
+        tokio::time::sleep(POLL_INTERVAL).await;
+    }
+}
+
+#[cfg(all(test, feature = "remote"))]
+mod tests {
+    use super::*;
+
+    fn http(status: u16) -> Error {
+        Error::Http {
+            source: "server said no".into(),
+            request_id: "rid".into(),
+            status_code: reqwest::StatusCode::from_u16(status).ok(),
+        }
+    }
+
+    /// Every status the loop acts on. The two predicates are checked together
+    /// because their overlap is what would be wrong: a status must never be
+    /// both, and 421 in particular must not read as retryable — retrying it in
+    /// place re-issues the call that just said the node holds no claim.
+    #[test]
+    fn taxonomy_round_trips() {
+        for status in [429, 503] {
+            assert!(is_retryable(&http(status)), "{status} must retry");
+            assert!(
+                !is_lost_claim(&http(status)),
+                "{status} is not a lost claim"
+            );
+        }
+        assert!(is_lost_claim(&http(421)), "a lost claim must re-claim");
+        assert!(
+            !is_retryable(&http(421)),
+            "retrying a lost claim in place only asks the same node again"
+        );
+        for status in [400, 404, 409, 500] {
+            assert!(!is_retryable(&http(status)), "{status} is terminal");
+            assert!(!is_lost_claim(&http(status)), "{status} is terminal");
+        }
+    }
+
+    /// An error carrying no status has no retry story and must be terminal —
+    /// a connection that never reached the server, or a `TableNotFound` that
+    /// `check_table_response` translated before the loop saw it.
+    #[test]
+    fn errors_without_a_status_are_terminal() {
+        let no_status = Error::Http {
+            source: "connection reset".into(),
+            request_id: "rid".into(),
+            status_code: None,
+        };
+        assert!(!is_retryable(&no_status));
+        assert!(!is_lost_claim(&no_status));
+
+        let translated = Error::TableNotFound {
+            name: "t".into(),
+            source: "gone".into(),
+        };
+        assert!(!is_retryable(&translated));
+        assert!(!is_lost_claim(&translated));
+    }
+}
@@ -0,0 +1,162 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+//! Live per-bucket LSM state — the shape [`crate::Table::get_lsm_stats`]
+//! returns and [`super::checkpoint`] polls.
+//!
+//! Nothing here is derived: sums and differences (total L0 bytes, WAL lag)
+//! are the caller's to compute. There is no "WAL is off" shape — that case is
+//! `None`, because a struct of zeros would read as measurements.
+
+use serde::Deserialize;
+
+/// One flushed L0 generation.
+#[derive(Debug, Clone, Deserialize)]
+pub struct GenerationStats {
+    pub generation: u64,
+    pub bytes: u64,
+    /// Present only when `include_generation_rows` was requested. Off by
+    /// default because each count opens an uncached Lance dataset, and the
+    /// checkpoint loop polls this route needing only generation numbers.
+    #[serde(default)]
+    pub rows: Option<u64>,
+}
+
+/// One in-memory memtable.
+#[derive(Debug, Clone, Deserialize)]
+pub struct MemtableStats {
+    pub generation: u64,
+    pub rows: u64,
+    pub bytes: u64,
+    pub batches: u64,
+    /// Names of the indexes this memtable carries. An absent name is the whole
+    /// answer to "why is my fresh-tier search on that column brute-force".
+    pub indexes: Vec<String>,
+}
+
+/// Live state of one bucket. A table is N buckets on one node; flattening to
+/// a single number hides the one hot bucket that is usually why someone
+/// opened this endpoint.
+#[derive(Debug, Clone, Deserialize)]
+pub struct BucketStats {
+    pub shard_id: String,
+    /// `Active` | `Sealed` (drop-table 2PC in flight).
+    pub status: String,
+    pub writer_epoch: u64,
+    pub manifest_version: u64,
+    pub current_generation: u64,
+    pub replay_after_wal_entry_position: u64,
+    pub wal_entry_position_last_seen: u64,
+    pub generations: Vec<GenerationStats>,
+    /// Whether a pass owns this bucket's compaction latch right now. Says *a*
+    /// driver is running, not *whose*, and the latch is held from dispatch —
+    /// including while the pass queues for a pod-wide compactor permit. Read
+    /// it as "do not pile on", never as "mine is progressing".
+    pub compacting: bool,
+    /// Oldest first, active last. Absent for a `Sealed` bucket, whose
+    /// in-memory state is torn down.
+    #[serde(default)]
+    pub memtables: Option<Vec<MemtableStats>>,
+}
+
+impl BucketStats {
+    /// The newest flushed generation, or `None` when L0 is empty.
+    pub(crate) fn newest_generation(&self) -> Option<u64> {
+        self.generations.iter().map(|g| g.generation).max()
+    }
+
+    /// How many generations at or below `target` are still in L0.
+    ///
+    /// A count, not a boolean: one pass drains a bounded prefix rather than
+    /// the whole target set, so a boolean would read as "no progress" for
+    /// every pass but the last. Compaction drains oldest-first, so this
+    /// decreases monotonically.
+    pub(crate) fn outstanding_generations(&self, target: u64) -> usize {
+        self.generations
+            .iter()
+            .filter(|g| g.generation <= target)
+            .count()
+    }
+}
+
+/// Live LSM state, one entry per bucket.
+#[derive(Debug, Clone, Deserialize)]
+pub struct LsmStats {
+    pub buckets: Vec<BucketStats>,
+}
+
+/// Server-side JSON envelope for `get_lsm_stats`. `lsm_stats` is null when
+/// the table has no LSM write path.
+#[derive(Debug, Deserialize)]
+pub(crate) struct GetLsmStatsResponse {
+    #[serde(default)]
+    pub lsm_stats: Option<LsmStats>,
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn bucket(shard: &str, generations: &[u64], compacting: bool) -> BucketStats {
+        BucketStats {
+            shard_id: shard.into(),
+            status: "Active".into(),
+            writer_epoch: 1,
+            manifest_version: 1,
+            current_generation: generations.iter().max().copied().unwrap_or(0) + 1,
+            replay_after_wal_entry_position: 0,
+            wal_entry_position_last_seen: 0,
+            generations: generations
+                .iter()
+                .map(|g| GenerationStats {
+                    generation: *g,
+                    bytes: 1,
+                    rows: None,
+                })
+                .collect(),
+            compacting,
+            memtables: None,
+        }
+    }
+
+    /// The target watermark is the newest generation at the start, and a
+    /// generation created after it must not hold the loop open — that is why
+    /// the predicate terminates under write load.
+    #[test]
+    fn newer_generations_do_not_extend_the_target() {
+        let start = bucket("b0", &[7, 8], false);
+        let target = start.newest_generation().expect("L0 is non-empty");
+        assert_eq!(target, 8);
+
+        // Compaction drained 7 and 8; 9 and 10 arrived while it ran.
+        let later = bucket("b0", &[9, 10], false);
+        assert_eq!(
+            later.outstanding_generations(target),
+            0,
+            "generations above the target are somebody else's problem"
+        );
+
+        // Still holding 8 means still outstanding.
+        assert_eq!(
+            bucket("b0", &[8, 9], false).outstanding_generations(target),
+            1
+        );
+    }
+
+    /// The metric counts generations, not buckets: a pass drains a bounded
+    /// prefix, so one bucket going 3 → 2 → 1 → 0 is three steps.
+    #[test]
+    fn progress_is_measured_in_generations() {
+        let target = 3;
+        let counts: Vec<usize> = [&[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());
+    }
+}
@@ -1161,7 +1161,7 @@ mod lsm_tests {
            .unwrap();
        let fts_index = table.list_indices().await.unwrap()[0].name.clone();
        table
-            .set_lsm_write_spec(LsmWriteSpec::unsharded().with_maintained_indexes([fts_index]))
+            .set_lsm_write_spec(LsmWriteSpec::unsharded().with_maintained_indexes(vec![fts_index]))
            .await
            .unwrap();

@@ -1254,7 +1254,7 @@ mod lsm_tests {
            .unwrap();
        let vec_index = table.list_indices().await.unwrap()[0].name.clone();
        table
-            .set_lsm_write_spec(LsmWriteSpec::unsharded().with_maintained_indexes([vec_index]))
+            .set_lsm_write_spec(LsmWriteSpec::unsharded().with_maintained_indexes(vec![vec_index]))
            .await
            .unwrap();

@@ -29,6 +29,7 @@ use arrow_schema::{DataType, Schema as ArrowSchema, SchemaRef};
 use lance::Dataset;
 use lance::dataset::mem_wal::{
    DatasetMemWalExt, ShardWriter, ShardWriterConfig, evaluate_sharding_spec,
+    validate_maintained_indexes,
 };
 use lance::index::DatasetIndexExt;
 use lance_core::datatypes::Schema as LanceSchema;
@@ -37,8 +38,9 @@ use tokio::sync::RwLock;
 use uuid::Uuid;

 use crate::error::{Error, Result};
+use crate::index::IndexConfig;
 use crate::table::merge::{MergeInsertBuilder, MergeResult};
-use crate::table::{LsmWriteSpec, NativeTable};
+use crate::table::{BaseTable, LsmWriteSpec, NativeTable};

 /// Spec id of the sole sharding spec installed by [`set_lsm_write_spec`].
 /// Must match Lance's `InitializeMemWalBuilder` (`SHARDING_SPEC_ID`).
@@ -80,32 +82,44 @@ pub(crate) async fn set_lsm_write_spec(table: &NativeTable, spec: LsmWriteSpec)
        }
    }

+    // Before the builder borrows the dataset clone. `list_indices` merges an
+    // index's segments into one entry, so the result needs no dedup.
+    let maintained_indexes = {
+        let dataset = table.dataset.get().await?;
+        resolve_maintained_indexes(
+            &dataset,
+            &table.list_indices().await?,
+            spec.maintained_indexes(),
+        )
+        .await?
+    };
+
    let mut dataset = (*table.dataset.get().await?).clone();
    let mut builder = dataset.initialize_mem_wal();
-    let (maintained_indexes, writer_config_defaults) = match spec {
+    let writer_config_defaults = match spec {
        LsmWriteSpec::Bucket {
            column,
            num_buckets,
-            maintained_indexes,
            writer_config_defaults,
+            ..
        } => {
            builder = builder.bucket_sharding(column, num_buckets);
-            (maintained_indexes, writer_config_defaults)
+            writer_config_defaults
        }
        LsmWriteSpec::Identity {
            column,
-            maintained_indexes,
            writer_config_defaults,
+            ..
        } => {
            builder = builder.identity_sharding(column);
-            (maintained_indexes, writer_config_defaults)
+            writer_config_defaults
        }
        LsmWriteSpec::Unsharded {
-            maintained_indexes,
            writer_config_defaults,
+            ..
        } => {
            builder = builder.unsharded();
-            (maintained_indexes, writer_config_defaults)
+            writer_config_defaults
        }
    };
    builder = builder.maintained_indexes(maintained_indexes);
@@ -117,6 +131,58 @@ pub(crate) async fn set_lsm_write_spec(table: &NativeTable, spec: LsmWriteSpec)
    Ok(())
 }

+/// Resolve a spec's maintained-index selection against `indices`, as reported
+/// by [`Table::list_indices`](crate::Table::list_indices).
+///
+/// `None` means every index on the table, snapshotted now. Lance validates
+/// either selection against its shard-writer rules, so a spec that installs is
+/// one the MemWAL can open.
+///
+/// An unmaintainable index fails an inferred set rather than being dropped from
+/// it — dropping would leave the caller believing it is maintained.
+async fn resolve_maintained_indexes(
+    dataset: &Dataset,
+    indices: &[IndexConfig],
+    requested: Option<&[String]>,
+) -> Result<Vec<String>> {
+    let Some(requested) = requested else {
+        let all: Vec<String> = indices.iter().map(|index| index.name.clone()).collect();
+        validate_maintained_indexes(dataset, &all)
+            .await
+            .map_err(|source| Error::InvalidInput {
+                message: format!(
+                    "cannot maintain every index on this table: {source}. Set \
+                     maintained_indexes explicitly to choose from {}",
+                    index_name_list(indices),
+                ),
+            })?;
+        return Ok(all);
+    };
+    for name in requested {
+        if !indices.iter().any(|index| &index.name == name) {
+            return Err(Error::InvalidInput {
+                message: format!(
+                    "maintained index '{}' does not exist on this table; it has {}",
+                    name,
+                    index_name_list(indices),
+                ),
+            });
+        }
+    }
+    validate_maintained_indexes(dataset, requested).await?;
+    Ok(requested.to_vec())
+}
+
+/// Index names for an error message.
+fn index_name_list(indices: &[IndexConfig]) -> String {
+    if indices.is_empty() {
+        return "no indexes".to_string();
+    }
+    let mut names: Vec<&str> = indices.iter().map(|index| index.name.as_str()).collect();
+    names.sort_unstable();
+    format!("[{}]", names.join(", "))
+}
+
 // =============================================================================
 // unset_lsm_write_spec
 // =============================================================================