Merge remote-tracking branch 'origin/main' into gatekeeper/fix-1982-1

Merge origin/main into gatekeeper/fix-1982-1
fix(python): support typed embedding vector fields
2026-08-28 08:58:41 +00:00 · 2026-08-07 10:13:11 +00:00 · 2026-08-07 10:12:21 +00:00 · 2026-08-06 00:27:56 +00:00
42 changed files with 251 additions and 2063 deletions
@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "0.37.1-beta.1"
+current_version = "0.37.1-beta.0"
 parse = """(?x)
    (?P<major>0|[1-9]\\d*)\\.
    (?P<minor>0|[1-9]\\d*)\\.
@@ -3455,8 +3455,8 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"

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

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

 [[package]]
 name = "lance-arrow"
-version = "11.0.0-beta.3"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
+version = "11.0.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
 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.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
 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.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -4936,8 +4936,8 @@ dependencies = [

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

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

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

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

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

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

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

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

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

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

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

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

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

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

 [[package]]
 name = "lance-table"
-version = "11.0.0-beta.3"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
+version = "11.0.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5353,7 +5354,6 @@ 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.3"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
+version = "11.0.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -5397,8 +5397,8 @@ dependencies = [

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

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

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

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

 [workspace.dependencies]
-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" }
+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" }
 ahash = "0.8"
 # Note that this one does not include pyarrow
 arrow = { version = "58.0.0", optional = false }
@@ -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.1</version>
+    <version>0.37.1-beta.0</version>
 </dependency>
 ```

@@ -431,10 +431,9 @@ 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 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.
+The returned spec — including its `maintainedIndexes` and
+`writerConfigDefaults` — mirrors what was passed to
+[Table#setLsmWriteSpec](Table.md#setlsmwritespec).

 #### Returns

@@ -807,11 +806,6 @@ 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,9 +34,7 @@ Bucket and identity variants: the sharding column.
 optional maintainedIndexes: string[];
 ```

-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.
+Names of indexes the MemWAL should keep up to date during writes.

 ***

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

-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.
+The total number of bytes in the table
@@ -8,7 +8,7 @@
    <parent>
      <groupId>com.lancedb</groupId>
      <artifactId>lancedb-parent</artifactId>
-      <version>0.37.1-beta.1</version>
+      <version>0.37.1-beta.0</version>
      <relativePath>../pom.xml</relativePath>
    </parent>

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

    <groupId>com.lancedb</groupId>
    <artifactId>lancedb-parent</artifactId>
-    <version>0.37.1-beta.1</version>
+    <version>0.37.1-beta.0</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.3</lance-core.version>
+        <lance-core.version>11.0.0-beta.2</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.1"
+version = "0.37.1-beta.0"
 publish = false
 license.workspace = true
 description.workspace = true
@@ -69,33 +69,6 @@ describe("given a connection", () => {
    await expect(tbl.countRows()).resolves.toBe(1);
  });

-  it("should isolate object-form table creation across databases", async () => {
-    const otherTmpDir = tmp.dirSync({ unsafeCleanup: true });
-    const otherDb = await connect(otherTmpDir.name);
-
-    try {
-      const firstTable = await db.createTable({
-        name: "defaultTable",
-        data: [{ rowId: "id1", vector: Array(384).fill(0) }],
-      });
-      const secondTable = await otherDb.createTable({
-        name: "defaultTable",
-        data: [{ rowId: "id2", vector: Array(384).fill(0) }],
-      });
-
-      await expect(db.tableNames()).resolves.toEqual(["defaultTable"]);
-      await expect(otherDb.tableNames()).resolves.toEqual(["defaultTable"]);
-
-      const firstRows = await firstTable.query().select(["rowId"]).toArray();
-      const secondRows = await secondTable.query().select(["rowId"]).toArray();
-      expect(firstRows.map((row) => row.rowId)).toEqual(["id1"]);
-      expect(secondRows.map((row) => row.rowId)).toEqual(["id2"]);
-    } finally {
-      otherDb.close();
-      otherTmpDir.removeCallback();
-    }
-  });
-
  it("should be able to drop tables`", async () => {
    await db.createTable("test", [{ id: 1 }, { id: 2 }]);
    await db.createTable("test2", [{ id: 1 }, { id: 2 }]);
@@ -277,16 +277,8 @@ describe.each([arrow15, arrow16, arrow17, arrow18])(
        },
        numIndices: 0,
        numRows: 3,
-        // Full on-disk size of the two data files, footers and metadata included.
-        totalBytes: 684,
+        totalBytes: 44,
      });
-
-      // 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,11 +197,7 @@ export interface LsmWriteSpec {
  column?: string;
  /** Bucket variant: the number of buckets, in `[1, 1024]`. */
  numBuckets?: number;
-  /**
-   * 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.
-   */
+  /** Names of indexes the MemWAL should keep up to date during writes. */
  maintainedIndexes?: string[];
  /** Default `ShardWriter` configuration recorded in the MemWAL index. */
  writerConfigDefaults?: Record<string, string>;
@@ -599,11 +595,6 @@ 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
@@ -631,10 +622,9 @@ 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 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.
+   * The returned spec — including its `maintainedIndexes` and
+   * `writerConfigDefaults` — mirrors what was passed to
+   * {@link Table#setLsmWriteSpec}.
   * @returns {Promise<LsmWriteSpec | undefined>}
   */
  abstract getLsmWriteSpec(): Promise<LsmWriteSpec | undefined>;
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-darwin-arm64",
-	"version": "0.37.1-beta.1",
+	"version": "0.37.1-beta.0",
 	"os": ["darwin"],
 	"cpu": ["arm64"],
 	"main": "lancedb.darwin-arm64.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-gnu",
-	"version": "0.37.1-beta.1",
+	"version": "0.37.1-beta.0",
 	"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.1",
+	"version": "0.37.1-beta.0",
 	"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.1",
+	"version": "0.37.1-beta.0",
 	"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.1",
+	"version": "0.37.1-beta.0",
 	"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.1",
+  "version": "0.37.1-beta.0",
  "os": [
    "win32"
  ],
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-win32-x64-msvc",
-	"version": "0.37.1-beta.1",
+	"version": "0.37.1-beta.0",
 	"os": ["win32"],
 	"cpu": ["x64"],
 	"main": "lancedb.win32-x64-msvc.node",
@@ -1,12 +1,12 @@
 {
  "name": "@lancedb/lancedb",
-  "version": "0.37.1-beta.1",
+  "version": "0.37.1-beta.0",
  "lockfileVersion": 3,
  "requires": true,
  "packages": {
    "": {
      "name": "@lancedb/lancedb",
-      "version": "0.37.1-beta.1",
+      "version": "0.37.1-beta.0",
      "cpu": [
        "x64",
        "arm64"
@@ -11,7 +11,7 @@
    "ann"
  ],
  "private": false,
-  "version": "0.37.1-beta.1",
+  "version": "0.37.1-beta.0",
  "main": "dist/index.js",
  "exports": {
    ".": "./dist/index.js",
@@ -772,8 +772,7 @@ pub struct LsmWriteSpec {
    pub column: Option<String>,
    /// Bucket variant: the number of buckets, in `[1, 1024]`.
    pub num_buckets: Option<u32>,
-    /// Indexes the MemWAL keeps up to date. Omitted resolves every
-    /// maintainable index on install; an empty array means none.
+    /// Names of indexes the MemWAL should keep up to date during writes.
    pub maintained_indexes: Option<Vec<String>>,
    /// Default `ShardWriter` configuration recorded in the MemWAL index.
    pub writer_config_defaults: Option<HashMap<String, String>>,
@@ -783,6 +782,7 @@ 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(value.maintained_indexes)
+            .with_maintained_indexes(maintained)
            .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,
+                maintained_indexes: Some(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,
+                maintained_indexes: Some(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,
+                maintained_indexes: Some(maintained_indexes),
                writer_config_defaults: Some(writer_config_defaults),
            },
        }
@@ -1043,10 +1043,7 @@ impl From<lancedb::index::IndexStatistics> for IndexStatistics {

 #[napi(object)]
 pub struct TableStatistics {
-    /// 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.
+    /// The total number of bytes in the table
    pub total_bytes: i64,

    /// The number of rows in the table
@@ -1,6 +1,6 @@
 [package]
 name = "lancedb-python"
-version = "0.37.1-beta.1"
+version = "0.37.1-beta.0"
 publish = false
 edition.workspace = true
 description = "Python bindings for LanceDB"
@@ -355,10 +355,6 @@ 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: ...
@@ -653,10 +649,9 @@ class LsmWriteSpec:
    def identity(column: str) -> "LsmWriteSpec": ...
    @staticmethod
    def unsharded() -> "LsmWriteSpec": ...
-    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_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_writer_config_defaults(self, defaults: Dict[str, str]) -> "LsmWriteSpec":
        """Return a copy of this spec recording the given default
@@ -671,9 +666,7 @@ class LsmWriteSpec:
    @property
    def num_buckets(self) -> Optional[int]: ...
    @property
-    def maintained_indexes(self) -> Optional[List[str]]:
-        """Indexes the MemWAL keeps up to date, or None for every supported one."""
-        ...
+    def maintained_indexes(self) -> List[str]: ...
    @property
    def writer_config_defaults(self) -> Dict[str, str]: ...

@@ -183,7 +183,11 @@ class EmbeddingFunction(BaseModel, ABC):
    def VectorField(self, **kwargs):
        """
        Creates a pydantic Field that can automatically annotate
-        the target vector column for this embedding function
+        the target vector column for this embedding function.
+
+        The field can be annotated as ``list[float]`` for compatibility with
+        static type checkers. LanceDB will infer the fixed vector dimension from
+        this embedding function.
        """
        return Field(json_schema_extra={"vector_column_for": self}, **kwargs)

@@ -87,13 +87,12 @@ class JinaEmbeddings(EmbeddingFunction):
        if isinstance(image, bytes):
            image_dict = {"image": base64.b64encode(image).decode("utf-8")}
        elif isinstance(image, (str, Path)):
-            parsed = urlparse(str(image))
+            parsed = urlparse.urlparse(image)
+            # TODO handle drive letter on windows.
            PIL_Image = attempt_import_or_raise("PIL.Image", "pillow")
            if parsed.scheme == "file":
                pil_image = PIL_Image.open(parsed.path)
-            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.
+            elif parsed.scheme == "":
                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)))
@@ -99,6 +99,17 @@ def Vector(
    ...     pa.field("url", pa.utf8(), False),
    ...     pa.field("embeddings", pa.list_(pa.float32(), 768))
    ... ])
+
+    Notes
+    -----
+    ``Vector`` creates a type dynamically, so calls such as ``Vector(768)`` are
+    not valid static type annotations. For an embedding field, use the standard
+    ``list[float]`` annotation when running mypy; ``VectorField`` supplies the
+    fixed dimension to LanceDB::
+
+        class MyModel(LanceModel):
+            text: str = embeddings.SourceField()
+            vector: list[float] = embeddings.VectorField()
    """

    # TODO: make a public parameterized type.
@@ -379,6 +390,10 @@ def _unwrap_optional_annotation(annotation: Any) -> Any | None:

 def _pydantic_to_arrow_type(field: FieldInfo) -> pa.DataType:
    """Convert a Pydantic FieldInfo to Arrow DataType"""
+    embedding_vector_type = _embedding_vector_to_arrow_type(field)
+    if embedding_vector_type is not None:
+        return embedding_vector_type
+
    unwrapped = _unwrap_optional_annotation(field.annotation)
    if unwrapped is not None:
        return _pydantic_type_to_arrow_type(unwrapped, field)
@@ -392,8 +407,32 @@ def _pydantic_to_arrow_type(field: FieldInfo) -> pa.DataType:
    return _pydantic_type_to_arrow_type(field.annotation, field)


+def _embedding_vector_to_arrow_type(field: FieldInfo) -> pa.DataType | None:
+    """Infer a fixed-size vector type from ``VectorField`` metadata."""
+    if not _is_embedding_vector_annotation(field):
+        return None
+
+    function = get_extras(field, "vector_column_for")
+    return pa.list_(pa.float32(), function.ndims())
+
+
+def _is_embedding_vector_annotation(field: FieldInfo) -> bool:
+    if get_extras(field, "vector_column_for") is None:
+        return False
+
+    annotation = _unwrap_optional_annotation(field.annotation)
+    if annotation is None:
+        annotation = field.annotation
+
+    origin = getattr(annotation, "__origin__", None)
+    args = getattr(annotation, "__args__", ())
+    return origin is list and args == (float,)
+
+
 def is_nullable(field: FieldInfo) -> bool:
    """Check if a Pydantic FieldInfo is nullable."""
+    if _is_embedding_vector_annotation(field):
+        return True
    if _unwrap_optional_annotation(field.annotation) is not None:
        return True
    if isinstance(field.annotation, (_GenericAlias, GenericAlias)):
@@ -3976,28 +3976,6 @@ 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]."""
@@ -4676,13 +4654,6 @@ 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
@@ -4709,73 +4680,12 @@ 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 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.
+        The returned spec — including its ``maintained_indexes`` and
+        ``writer_config_defaults`` — mirrors what was passed to
+        `set_lsm_write_spec`.
        """
        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.

@@ -6341,9 +6251,7 @@ class TableStatistics:
    Attributes
    ----------
    total_bytes: int
-        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.
+        The total number of bytes in the table.
    num_rows: int
        The total number of rows in the table.
    num_indices: int
@@ -631,23 +631,3 @@ 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,9 +83,7 @@ def test_lsm_write_spec_repr():
    assert s.spec_type == "bucket"
    assert s.column == "id"
    assert s.num_buckets == 4
-    # A fresh spec defers its maintained set to install time.
-    assert s.maintained_indexes is None
-    assert s.with_maintained_indexes([]).maintained_indexes == []
+    assert s.maintained_indexes == []
    assert "bucket" in repr(s)
    assert "id" in repr(s)
    assert "4" in repr(s)
@@ -171,23 +169,18 @@ 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). 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".
+    # Identity round-trips (column recovered from the schema).
    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). Opting out is distinct from
-    # the inferred default.
-    table.set_lsm_write_spec(LsmWriteSpec.unsharded().with_maintained_indexes([]))
+    # Unsharded round-trips (no routing column).
+    table.set_lsm_write_spec(LsmWriteSpec.unsharded())
    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().with_maintained_indexes([]))
+    table.set_lsm_write_spec(LsmWriteSpec.unsharded())
    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().with_maintained_indexes([]))
+    table.set_lsm_write_spec(LsmWriteSpec.unsharded())
    with pytest.raises(Exception, match="maintained"):
        table.search([1.0] * VECTOR_DIM).to_arrow()

@@ -9,6 +9,7 @@ from typing import List, Optional, Tuple
 import pyarrow as pa
 import pydantic
 import pytest
+from lancedb.conftest import MockTextEmbeddingFunction
 from lancedb.pydantic import (
    PYDANTIC_VERSION,
    LanceModel,
@@ -426,6 +427,25 @@ def test_bare_vector_raises_clear_error():
        exec("class TestModel(LanceModel):\n    vector: Vector", namespace)


+def test_embedding_vector_list_annotation():
+    embedding = MockTextEmbeddingFunction.create()
+
+    class StaticTypingModel(LanceModel):
+        text: str = embedding.SourceField()
+        vector: list[float] = embedding.VectorField()
+
+    schema = pydantic_to_schema(StaticTypingModel)
+    assert schema == pa.schema(
+        [
+            pa.field("text", pa.utf8(), False),
+            pa.field("vector", pa.list_(pa.float32(), embedding.ndims()), True),
+        ]
+    )
+
+    model = StaticTypingModel(text="hello", vector=[0.0] * embedding.ndims())
+    assert model.vector == [0.0] * embedding.ndims()
+
+
 def test_fixed_size_list_field():
    class TestModel(pydantic.BaseModel):
        vec: Vector(16)
@@ -3713,8 +3713,7 @@ def test_stats(mem_db: DBConnection):
    stats = table.stats()
    print(f"{stats=}")
    assert stats == {
-        # Full on-disk size of the data file, footer and metadata included.
-        "total_bytes": 633,
+        "total_bytes": 60,
        "num_rows": 2,
        "num_indices": 0,
        "fragment_stats": {
@@ -3732,13 +3731,6 @@ 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
@@ -28,72 +28,11 @@ use pyo3::{
    Bound, FromPyObject, Py, PyAny, PyRef, PyResult, Python,
    exceptions::{PyRuntimeError, PyValueError},
    pyclass, pyfunction, pymethods,
-    types::{IntoPyDict, PyAnyMethods, PyBytes, PyDict, PyDictMethods, PyList, PyListMethods},
+    types::{IntoPyDict, PyAnyMethods, PyBytes, PyDict, PyDictMethods},
 };

 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),
@@ -246,22 +185,12 @@ 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(...)`. A fresh spec maintains every index the
-/// MemWAL supports, resolved on install.
+/// `with_writer_config_defaults(...)`.
 #[pyclass(from_py_object)]
 #[derive(Clone, Debug)]
 pub struct LsmWriteSpec {
@@ -301,11 +230,11 @@ impl LsmWriteSpec {
        }
    }

-    /// 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 {
+    /// 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 {
        Self {
            inner: self.inner.clone().with_maintained_indexes(indexes),
        }
@@ -327,29 +256,23 @@ impl LsmWriteSpec {
                maintained_indexes,
                writer_config_defaults,
            } => format!(
-                "LsmWriteSpec.bucket(column={:?}, num_buckets={}, maintained_indexes={}, writer_config_defaults={:?})",
-                column,
-                num_buckets,
-                fmt_maintained(maintained_indexes),
-                writer_config_defaults,
+                "LsmWriteSpec.bucket(column={:?}, num_buckets={}, maintained_indexes={:?}, writer_config_defaults={:?})",
+                column, num_buckets, 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,
-                fmt_maintained(maintained_indexes),
-                writer_config_defaults,
+                "LsmWriteSpec.identity(column={:?}, maintained_indexes={:?}, writer_config_defaults={:?})",
+                column, maintained_indexes, writer_config_defaults,
            ),
            lancedb::table::LsmWriteSpec::Unsharded {
                maintained_indexes,
                writer_config_defaults,
            } => format!(
-                "LsmWriteSpec.unsharded(maintained_indexes={}, writer_config_defaults={:?})",
-                fmt_maintained(maintained_indexes),
-                writer_config_defaults,
+                "LsmWriteSpec.unsharded(maintained_indexes={:?}, writer_config_defaults={:?})",
+                maintained_indexes, writer_config_defaults,
            ),
        }
    }
@@ -384,10 +307,10 @@ impl LsmWriteSpec {
        }
    }

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

    /// Default `ShardWriter` configuration recorded by this spec.
@@ -1416,51 +1339,6 @@ 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 {
@@ -1,6 +1,6 @@
 [package]
 name = "lancedb"
-version = "0.37.1-beta.1"
+version = "0.37.1-beta.0"
 edition.workspace = true
 description = "LanceDB: A serverless, low-latency vector database for AI applications"
 license.workspace = true
@@ -100,8 +100,7 @@ anyhow = "1"
 lance-testing = { workspace = true }
 tempfile = "3.5.0"
 random_word = { version = "0.4.3", features = ["en"] }
-roaring = "0.11.4"
-tokio = { version = "1.23", features = ["io-util", "macros", "net", "rt-multi-thread", "sync", "test-util"] }
+tokio = { version = "1.23", features = ["io-util", "macros", "net", "rt-multi-thread", "sync"] }
 uuid = { version = "1.7.0", features = ["v4"] }
 walkdir = "2"
 aws-sdk-dynamodb = { version = "1.55.0" }
@@ -23,13 +23,11 @@ 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;
@@ -993,18 +991,6 @@ 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 {
@@ -2482,47 +2468,13 @@ 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. A null
-        // `maintained_indexes` asks the server to resolve every maintainable
-        // index at HEAD; a list is verbatim, an empty one meaning none.
+        // 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").
        let sharding = match &spec {
            LsmWriteSpec::Bucket {
                column,
@@ -6599,7 +6551,7 @@ mod tests {
                .unwrap()
        });
        let spec = crate::table::LsmWriteSpec::unsharded()
-            .with_maintained_indexes(vec!["id_idx".to_string()])
+            .with_maintained_indexes(["id_idx"])
            .with_writer_config_defaults([("max_memtable_rows", "1000")]);
        table.set_lsm_write_spec(spec).await.unwrap();
    }
@@ -6618,29 +6570,11 @@ mod tests {
                body["sharding"],
                serde_json::json!({ "mode": "bucket", "column": "id", "num_buckets": 16 })
            );
-            // 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
-            .set_lsm_write_spec(crate::table::LsmWriteSpec::bucket("id", 16))
-            .await
-            .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()),
-            )
+            .set_lsm_write_spec(crate::table::LsmWriteSpec::bucket("id", 16))
            .await
            .unwrap();
    }
@@ -6719,7 +6653,7 @@ mod tests {
            } => {
                assert_eq!(column, "id");
                assert_eq!(num_buckets, 4);
-                assert_eq!(maintained_indexes, Some(vec!["id_idx".to_string()]));
+                assert_eq!(maintained_indexes, vec!["id_idx".to_string()]);
                assert_eq!(
                    writer_config_defaults
                        .get("durable_write")
@@ -6748,499 +6682,6 @@ 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,12 +68,10 @@ use self::merge::MergeInsertBuilder;
 pub mod add_columns;
 mod add_data;
 pub mod branch_merge;
-pub mod checkpoint;
 mod create_index;
 pub mod datafusion;
 pub(crate) mod dataset;
 pub mod delete;
-pub mod lsm_stats;
 pub mod merge;
 pub mod optimize;
 mod primary_key;
@@ -95,8 +93,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,
@@ -370,8 +368,6 @@ 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.
@@ -386,12 +382,9 @@ pub enum LsmWriteSpec {
    Bucket {
        column: String,
        num_buckets: u32,
-        /// 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>>,
+        /// Names of indexes (already created on the table) that the
+        /// MemWAL should maintain in-memory as rows are appended.
+        maintained_indexes: Vec<String>,
        /// Default `ShardWriter` configuration recorded in the MemWAL index.
        writer_config_defaults: HashMap<String, String>,
    },
@@ -401,41 +394,35 @@ pub enum LsmWriteSpec {
    /// distinct value of `column` becomes its own shard.
    Identity {
        column: 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>>,
+        /// Names of indexes (already created on the table) that the
+        /// MemWAL should maintain in-memory as rows are appended.
+        maintained_indexes: 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 {
-        /// 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>>,
+        /// Names of indexes (already created on the table) that the
+        /// MemWAL should maintain in-memory as rows are appended.
+        maintained_indexes: Vec<String>,
        /// Default `ShardWriter` configuration recorded in the MemWAL index.
        writer_config_defaults: HashMap<String, String>,
    },
 }

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

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

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

-    /// 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();
+    /// 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();
        match &mut self {
            Self::Bucket {
                maintained_indexes, ..
@@ -487,7 +465,7 @@ impl LsmWriteSpec {
            }
            | Self::Unsharded {
                maintained_indexes, ..
-            } => *maintained_indexes = indexes,
+            } => *maintained_indexes = v,
        }
        self
    }
@@ -523,9 +501,8 @@ impl LsmWriteSpec {
        self
    }

-    /// 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]> {
+    /// Borrow the list of index names this spec asks MemWAL to maintain.
+    pub fn maintained_indexes(&self) -> &[String] {
        match self {
            Self::Bucket {
                maintained_indexes, ..
@@ -535,7 +512,7 @@ impl LsmWriteSpec {
            }
            | Self::Unsharded {
                maintained_indexes, ..
-            } => maintained_indexes.as_deref(),
+            } => maintained_indexes,
        }
    }

@@ -708,31 +685,6 @@ 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
@@ -1733,7 +1685,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(vec!["id_idx".to_string()]),
+    ///         LsmWriteSpec::bucket("id", 16).with_maintained_indexes(["id_idx"]),
    ///     )
    ///     .await?;
    /// # Ok(())
@@ -1755,10 +1707,9 @@ 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 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.
+    /// The returned spec — including its [`LsmWriteSpec::maintained_indexes`] and
+    /// [`LsmWriteSpec::writer_config_defaults`] — mirrors what was passed to
+    /// [`Table::set_lsm_write_spec`].
    ///
    /// # Example
    ///
@@ -1775,85 +1726,6 @@ 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
@@ -3569,24 +3441,9 @@ 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?;
-        // 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 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;

        let frags = ds.get_fragments();
        let mut sorted_sizes = join_all(
@@ -3658,12 +3515,7 @@ impl BaseTable for NativeTable {
 #[skip_serializing_none]
 #[derive(Debug, Deserialize, PartialEq)]
 pub struct TableStatistics {
-    /// 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).
+    /// The total number of bytes in the table
    pub total_bytes: usize,

    /// The number of rows in the table
@@ -3724,7 +3576,6 @@ 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;
@@ -5107,7 +4958,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(vec![idx_name.clone()])
+            .with_maintained_indexes([idx_name])
            .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));
@@ -5117,125 +4968,15 @@ 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.with_maintained_indexes(vec![idx_name.clone()]))
-        );
+        assert_eq!(table.get_lsm_write_spec().await.unwrap(), Some(spec));
        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.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())
-        );
+        assert_eq!(table.get_lsm_write_spec().await.unwrap(), Some(spec));
    }

    #[tokio::test]
@@ -5283,16 +5024,12 @@ 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: 8925,
+                total_bytes: 2300,
                fragment_stats: FragmentStatistics {
                    num_fragments: 11,
                    num_small_fragments: 11,
@@ -5332,196 +5069,4 @@ 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
-        );
-    }
 }
@@ -1,315 +0,0 @@
-// 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));
-    }
-}
@@ -1,162 +0,0 @@
-// 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(vec![fts_index]))
+            .set_lsm_write_spec(LsmWriteSpec::unsharded().with_maintained_indexes([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![vec_index]))
+            .set_lsm_write_spec(LsmWriteSpec::unsharded().with_maintained_indexes([vec_index]))
            .await
            .unwrap();

@@ -29,7 +29,6 @@ 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;
@@ -38,9 +37,8 @@ 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::{BaseTable, LsmWriteSpec, NativeTable};
+use crate::table::{LsmWriteSpec, NativeTable};

 /// Spec id of the sole sharding spec installed by [`set_lsm_write_spec`].
 /// Must match Lance's `InitializeMemWalBuilder` (`SHARDING_SPEC_ID`).
@@ -82,44 +80,32 @@ 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 writer_config_defaults = match spec {
+    let (maintained_indexes, writer_config_defaults) = match spec {
        LsmWriteSpec::Bucket {
            column,
            num_buckets,
+            maintained_indexes,
            writer_config_defaults,
-            ..
        } => {
            builder = builder.bucket_sharding(column, num_buckets);
-            writer_config_defaults
+            (maintained_indexes, writer_config_defaults)
        }
        LsmWriteSpec::Identity {
            column,
+            maintained_indexes,
            writer_config_defaults,
-            ..
        } => {
            builder = builder.identity_sharding(column);
-            writer_config_defaults
+            (maintained_indexes, writer_config_defaults)
        }
        LsmWriteSpec::Unsharded {
+            maintained_indexes,
            writer_config_defaults,
-            ..
        } => {
            builder = builder.unsharded();
-            writer_config_defaults
+            (maintained_indexes, writer_config_defaults)
        }
    };
    builder = builder.maintained_indexes(maintained_indexes);
@@ -131,58 +117,6 @@ 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
 // =============================================================================
Author	SHA1	Message	Date
Gatefixer	42405cd6d6	Merge remote-tracking branch 'origin/main' into gatekeeper/fix-1982-1	2026-08-07 10:13:11 +00:00
Gatefixer	610541f5df	Merge origin/main into gatekeeper/fix-1982-1	2026-08-07 10:12:21 +00:00
Gatefixer	cb5a8a64bf	fix(python): support typed embedding vector fields	2026-08-06 00:27:56 +00:00