fix(storage): recover mirrored manifest creates

Closes #3704

## Problem

Transforms can fail on bad data (e.g. nulls/NaNs from incomplete user
surveys). Today any transform exception aborts iteration, and there is
no way to skip invalid rows during loading.

## Solution

New `on_transform_error` parameter on `StreamingDataset`:

- `"raise"` (default, matches current behavior and the convention in
tf.data / WebDataset / Ray Data)
- `"skip"` — drop the failing rows and continue
- `"warn"` — like skip, plus a logged warning per failing batch
- a WebDataset-style callable `handler(exc) -> bool`, so users can skip
only expected error types

Key design points:

- **Row-granular skipping**: when a batch fails, the transform is re-run
on single-row slices so only the rows that actually fail are dropped
(avoids Ray-style whole-block loss). Skips are counted in a new
`rows_skipped` property.
- **No crash on uneven skips**: the round-robin loop now ends the epoch
at the last cycle where every split still has a row, instead of hitting
`IndexError` when a split runs dry early.
- **Exact resumability under skips**: checkpoints are now
position-based. `state_dict` gains `positions_consumed_per_split` (exact
for owned splits), and a new `merge_state_dicts` static method combines
per-rank states via elementwise max for elastic resume across topology
changes. Old checkpoints without the new key still load. Positions equal
sample counts when nothing is skipped, so existing behavior is
unchanged.
- **Guardrail**: transforms returning the wrong number of rows now raise
a clear `ValueError` instead of silently corrupting split accounting.

### Answers to the issue's open questions

- *Can we do this?* Yes — all transforms funnel through one guarded call
in the Stage 2 pipeline.
- *What do other libraries do?* tf.data `ignore_errors()`, WebDataset
`handler=`, Ray `max_errored_blocks`; MosaicML StreamingDataset offers
nothing (skipping conflicts with its determinism model). This design
follows the common conventions: raise by default, opt-in skipping,
count/log drops.
- *Error handling or pre-filtering?* Both: the existing `filter=`
remains the recommended tool for predictable bad data (splits are built
post-filter, so all guarantees hold — now documented);
`on_transform_error` covers failures not expressible as a predicate.
- *Impact on splits / elastic determinism?* Per-split sample sequences
stay deterministic (skips are data-dependent, not topology-dependent).
With unequal bad-row counts across splits the last few global steps of
an epoch can differ across topologies (bounded by the skew), which is
documented on the parameter. With equal counts per split, full
determinism is preserved — covered by a test.

## Testing

15 new tests in `test_elastic_dataloader.py` covering: default raise,
invalid values, uniform and uneven skips (including epoch-end
truncation), warn logging, selective callable handlers, wrong-row-count
guardrail, determinism across runs and across world sizes (1/2/3/4) with
skips, exact mid-epoch resume with skips on the same topology, elastic
resume via `merge_state_dicts` (ws=2 → ws=1), merge validation, and
backward-compat loading of old checkpoints.

Note: relying on CI for the test run — my local machine OOMs during the
final link of the native extension. The change itself is pure Python.

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

.bumpversion.toml

@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "0.37.1-beta.0"
+current_version = "0.37.1-beta.1"
 parse = """(?x)
     (?P<major>0|[1-9]\\d*)\\.
     (?P<minor>0|[1-9]\\d*)\\.

.github/workflows/docs-link-check.yml

@@ -36,7 +36,9 @@ jobs:
     permissions:
       contents: read
     outputs:
+      checker_outcome: ${{ steps.lychee.outcome }}
       exit_code: ${{ steps.lychee.outputs.exit_code }}
+      status: ${{ steps.validate.outputs.status }}
     steps:
       - name: Checkout
         uses: actions/checkout@v6
@@ -50,6 +52,7 @@ jobs:
 
       - name: Check links
         id: lychee
+        continue-on-error: true
         uses: lycheeverse/lychee-action@e7477775783ea5526144ba13e8db5eec57747ce8 # v2.9.0
         with:
           # Restricted to http(s) on purpose. Much of docs/src is generated
@@ -68,38 +71,50 @@ jobs:
           format: json
           output: ./lychee/out.json
           jobSummary: false
-          # The report, not a red build, is the signal for broken links. The
-          # validation step below still fails the run if the check itself
-          # breaks.
+          # The report issue, not a red workflow run, is the signal for link
+          # findings and checker failures alike.
           fail: false
 
       - name: Validate report
+        id: validate
         # lychee does not reserve exit code 2 for broken links: its CLI
         # parser also exits 2 on an invalid option, before any link was
         # checked or any report written. Only a parseable report whose
-        # counts agree with the exit code counts as a link verdict; anything
-        # else fails here, and the report job below is skipped entirely, so
-        # the tracking issue is never touched. Exit 2 covers timeouts as
-        # well as errors, and a timed-out host is exactly the transient
-        # unavailability this report exists to surface, so both count as
-        # findings. Requiring total > 0 also catches a glob that silently
-        # stopped matching any file.
-        if: steps.lychee.outputs.exit_code == 0 || steps.lychee.outputs.exit_code == 2
+        # counts agree with a completed exit code (0 or 2) counts as a link
+        # verdict. Everything else becomes a checker-error report instead of
+        # failing the workflow. Exit 2 covers timeouts as well as errors, and a
+        # timed-out host is exactly the transient unavailability this report
+        # exists to surface, so both count as findings. Requiring total > 0
+        # also catches a glob that silently stopped matching any file.
+        if: always()
         env:
+          CHECKER_OUTCOME: ${{ steps.lychee.outcome }}
           EXIT_CODE: ${{ steps.lychee.outputs.exit_code }}
         run: |
-          jq -e --argjson code "$EXIT_CODE" '
-            (.total > 0) and
-            (if $code == 0
-             then .errors == 0 and .timeouts == 0
-               and (.error_map | length == 0) and (.timeout_map | length == 0)
-             else (.errors + .timeouts) > 0
-               and ((.error_map | length) + (.timeout_map | length)) > 0
-             end)
-          ' ./lychee/out.json
+          status=checker-error
+          if [[ "$CHECKER_OUTCOME" == success ]] &&
+             [[ "$EXIT_CODE" == 0 || "$EXIT_CODE" == 2 ]] &&
+             jq -e --argjson code "$EXIT_CODE" '
+               (.total > 0) and
+               (if $code == 0
+                then .errors == 0 and .timeouts == 0
+                  and (.error_map | length == 0) and (.timeout_map | length == 0)
+                else (.errors + .timeouts) > 0
+                  and ((.error_map | length) + (.timeout_map | length)) > 0
+                end)
+             ' ./lychee/out.json
+          then
+            if [[ "$EXIT_CODE" == 0 ]]; then
+              status=healthy
+            else
+              status=findings
+            fi
+          fi
+          echo "status=$status" >> "$GITHUB_OUTPUT"
+          echo "Validated link check as $status"
 
       - name: Upload report
-        if: steps.lychee.outputs.exit_code == 2
+        if: steps.validate.outputs.status == 'findings'
         uses: actions/upload-artifact@v7
         with:
           name: link-report
@@ -115,26 +130,11 @@ jobs:
     permissions:
       issues: write
     env:
+      CHECKER_OUTCOME: ${{ needs.scan.outputs.checker_outcome }}
       EXIT_CODE: ${{ needs.scan.outputs.exit_code }}
+      STATUS: ${{ needs.scan.outputs.status }}
       GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
     steps:
-      - name: Classify checker result
-        # lychee exits 0 when every link resolves and 2 when links fail,
-        # both already cross-checked against the report by the scan job's
-        # validation step. Anything else (1 runtime, 3 bad config) means the
-        # check never produced a link verdict, which must surface as a failed
-        # run rather than be published as "broken documentation links".
-        run: |
-          case "$EXIT_CODE" in
-            0|2)
-              echo "lychee exit code $EXIT_CODE"
-              ;;
-            *)
-              echo "::error::lychee exited with '$EXIT_CODE': the link check did not complete. Leaving the report issue untouched."
-              exit 1
-              ;;
-          esac
-
       - name: Find existing report issue
         id: report
         # Matched on title alone, and through search rather than a listing:
@@ -144,7 +144,7 @@ jobs:
         # Closed issues are included because a healthy run closes the report:
         # an open-only lookup would forget that identity and the next failing
         # run would open a duplicate. The oldest match stays the canonical
-        # report and is reopened below when links break again.
+        # report and is reopened below when a problem recurs.
         run: |
           match=$(gh issue list --repo "$GITHUB_REPOSITORY" --state all \
             --search "in:title \"$REPORT_TITLE\" author:app/github-actions" \
@@ -154,14 +154,14 @@ jobs:
           echo "state=$(jq -r '.state // empty' <<<"$match")" >> "$GITHUB_OUTPUT"
 
       - name: Download report
-        if: env.EXIT_CODE == 2
+        if: env.STATUS == 'findings'
         uses: actions/download-artifact@v8
         with:
           name: link-report
           path: ./lychee
 
       - name: Compose report
-        if: env.EXIT_CODE == 2
+        if: env.STATUS == 'findings'
         run: |
           run_url="$GITHUB_SERVER_URL/$GITHUB_REPOSITORY/actions/runs/$GITHUB_RUN_ID"
           {
@@ -185,22 +185,41 @@ jobs:
             ' ./lychee/out.json
           } > ./lychee/issue.md
 
+      - name: Compose checker error report
+        if: env.STATUS == 'checker-error'
+        run: |
+          mkdir -p ./lychee
+          run_url="$GITHUB_SERVER_URL/$GITHUB_REPOSITORY/actions/runs/$GITHUB_RUN_ID"
+          {
+            echo "The documentation link check did not complete in [the latest run]($run_url)."
+            echo
+            echo "This issue is rewritten by every scheduled run and closed automatically once a trustworthy run finds that all links resolve."
+            echo
+            echo "The checker did not produce a trustworthy link verdict. Treat the previous result, if any, as stale until a later run completes."
+            echo
+            echo "* Action outcome: \`$CHECKER_OUTCOME\`"
+            echo "* Exit code: \`${EXIT_CODE:-not reported}\`"
+            echo "* Verdict validation: \`failed\`"
+          } > ./lychee/issue.md
+
       - name: Reopen report issue
         # A healthy run closes the report, and the issue action below only
         # rewrites the body of whatever number it is given. Without an
-        # explicit reopen, the 2 -> 0 -> 2 sequence would keep rewriting a
-        # closed issue while links are broken. A CLOSED state implies the
-        # lookup found a canonical issue, so no separate emptiness check.
-        if: env.EXIT_CODE == 2 && steps.report.outputs.state == 'CLOSED'
+        # explicit reopen, a later finding or checker error would rewrite a
+        # closed issue. A CLOSED state implies the lookup found a canonical
+        # issue, so no separate emptiness check.
+        if: >-
+          env.STATUS != 'healthy' &&
+          steps.report.outputs.state == 'CLOSED'
         env:
           ISSUE_NUMBER: ${{ steps.report.outputs.number }}
         run: |
           run_url="$GITHUB_SERVER_URL/$GITHUB_REPOSITORY/actions/runs/$GITHUB_RUN_ID"
           gh issue reopen "$ISSUE_NUMBER" --repo "$GITHUB_REPOSITORY" \
-            --comment "Broken documentation links found again in [the latest run]($run_url)."
+            --comment "The documentation link checker reported a problem again in [the latest run]($run_url)."
 
-      - name: Report broken links
-        if: env.EXIT_CODE == 2
+      - name: Report link-check problem
+        if: env.STATUS != 'healthy'
         uses: peter-evans/create-issue-from-file@fca9117c27cdc29c6c4db3b86c48e4115a786710 # v6.0.0
         with:
           # Empty on the first failing run, which creates the issue; afterwards
@@ -213,7 +232,9 @@ jobs:
       - name: Close report issue once links are healthy
         # An OPEN state implies the lookup found a canonical issue; a report
         # that is already closed needs nothing.
-        if: env.EXIT_CODE == 0 && steps.report.outputs.state == 'OPEN'
+        if: >-
+          env.STATUS == 'healthy' &&
+          steps.report.outputs.state == 'OPEN'
         env:
           ISSUE_NUMBER: ${{ steps.report.outputs.number }}
         run: |

Cargo.lock

@@ -5411,7 +5411,7 @@ dependencies = [
 
 [[package]]
 name = "lancedb"
-version = "0.37.1-beta.0"
+version = "0.37.1-beta.1"
 dependencies = [
  "ahash",
  "anyhow",
@@ -5447,7 +5447,6 @@ dependencies = [
  "datafusion-physical-plan",
  "datafusion-sql",
  "futures",
- "goosefs-sdk",
  "half",
  "hf-hub",
  "http 1.5.0",
@@ -5480,6 +5479,7 @@ dependencies = [
  "random_word",
  "regex",
  "reqwest 0.12.28",
+ "roaring",
  "rstest",
  "semver",
  "serde",
@@ -5499,7 +5499,7 @@ dependencies = [
 
 [[package]]
 name = "lancedb-nodejs"
-version = "0.37.1-beta.0"
+version = "0.37.1-beta.1"
 dependencies = [
  "arrow-array",
  "arrow-buffer",
@@ -5524,7 +5524,7 @@ dependencies = [
 
 [[package]]
 name = "lancedb-python"
-version = "0.37.1-beta.0"
+version = "0.37.1-beta.1"
 dependencies = [
  "arrow",
  "async-trait",

docs/src/java/java.md

@@ -14,7 +14,7 @@ Add the following dependency to your `pom.xml`:
 <dependency>
     <groupId>com.lancedb</groupId>
     <artifactId>lancedb-core</artifactId>
-    <version>0.37.1-beta.0</version>
+    <version>0.37.1-beta.1</version>
 </dependency>
 ```
 

docs/src/js/classes/Table.md

@@ -431,9 +431,10 @@ Read the [LsmWriteSpec](../interfaces/LsmWriteSpec.md) currently installed on th
 
 Resolves to `undefined` when the MemWAL LSM write path is not enabled (no
 spec has been set, or it was removed with [Table#unsetLsmWriteSpec](Table.md#unsetlsmwritespec)).
-The returned spec — including its `maintainedIndexes` and
-`writerConfigDefaults` — mirrors what was passed to
-[Table#setLsmWriteSpec](Table.md#setlsmwritespec).
+The returned spec mirrors what was passed to
+[Table#setLsmWriteSpec](Table.md#setlsmwritespec), except that `maintainedIndexes` always
+reports the concrete list resolved when the spec was set — `undefined`
+never round-trips.
 
 #### Returns
 
@@ -806,6 +807,11 @@ All variants require the table to have an unenforced primary key
 ([Table#setUnenforcedPrimaryKey](Table.md#setunenforcedprimarykey)); bucket sharding additionally
 requires it to be the single column being bucketed.
 
+Omitting `maintainedIndexes` maintains every index on the table, resolved
+here, failing if one cannot be maintained — name them to install anyway.
+Naming them pins an exact set, and a still-building index is rejected
+rather than quietly omitted.
+
 #### Parameters
 
 * **spec**: [`LsmWriteSpec`](../interfaces/LsmWriteSpec.md)

docs/src/js/interfaces/LsmWriteSpec.md

@@ -34,7 +34,9 @@ Bucket and identity variants: the sharding column.
 optional maintainedIndexes: string[];
 ```
 
-Names of indexes the MemWAL should keep up to date during writes.
+Indexes the MemWAL keeps up to date. Omit to maintain every supported
+index, resolved on install — a snapshot, so indexes created later are not
+maintained. Pass `[]` for none.
 
 ***
 

docs/src/js/interfaces/TableStatistics.md

@@ -44,4 +44,7 @@ The number of rows in the table
 totalBytes: number;
 ```
 
-The total number of bytes in the table
+The total size, in bytes, of the table's data files, index files, and
+overlay files
+
+Read from the manifest, so this excludes deletion files and manifests.

java/lancedb-core/pom.xml

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

java/pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>com.lancedb</groupId>
     <artifactId>lancedb-parent</artifactId>
-    <version>0.37.1-beta.0</version>
+    <version>0.37.1-beta.1</version>
     <packaging>pom</packaging>
     <name>${project.artifactId}</name>
     <description>LanceDB Java SDK Parent POM</description>

nodejs/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "lancedb-nodejs"
 edition.workspace = true
-version = "0.37.1-beta.0"
+version = "0.37.1-beta.1"
 publish = false
 license.workspace = true
 description.workspace = true

nodejs/__test__/arrow.test.ts

@@ -6,7 +6,9 @@ import * as arrow17 from "apache-arrow-17";
 import * as arrow18 from "apache-arrow-18";
 
 import {
+  Vector as CurrentVector,
   convertToTable,
+  tableFromIPC as currentTableFromIPC,
   fromBufferToRecordBatch,
   fromDataToBuffer,
   fromRecordBatchToBuffer,
@@ -19,6 +21,7 @@ import {
   FunctionOptions,
 } from "../lancedb/embedding/embedding_function";
 import { EmbeddingFunctionConfig } from "../lancedb/embedding/registry";
+import { sanitizeTable } from "../lancedb/sanitize";
 
 // biome-ignore lint/suspicious/noExplicitAny: skip
 function sampleRecords(): Array<Record<string, any>> {
@@ -64,7 +67,11 @@ describe.each([arrow15, arrow16, arrow17, arrow18])(
       tableFromIPC,
       DataType,
       Dictionary,
+      RecordBatch: ArrowRecordBatch,
+      Table: ArrowTable,
       Uint8: ArrowUint8,
+      makeData: arrowMakeData,
+      vectorFromArray,
       // biome-ignore lint/suspicious/noExplicitAny: <explanation>
     } = <any>arrow;
     type Schema = ApacheArrow["Schema"];
@@ -1054,6 +1061,114 @@ describe.each([arrow15, arrow16, arrow17, arrow18])(
     });
 
     describe("when using two versions of arrow", function () {
+      it("preserves a dictionary shared by multiple fields", async function () {
+        const values = ["alpha", "beta", "alpha"];
+        const dictionaryVector = vectorFromArray(values);
+        const batch = new ArrowRecordBatch({
+          first: dictionaryVector.data[0],
+          second: dictionaryVector.data[0],
+        });
+        const table = new ArrowTable([batch]);
+
+        const sanitized = sanitizeTable(table);
+        expect([...sanitized.getChild("first")!]).toEqual(values);
+        expect([...sanitized.getChild("second")!]).toEqual(values);
+        const firstType = sanitized.schema.fields[0].type as {
+          dictionary: unknown;
+        };
+        const secondType = sanitized.schema.fields[1].type as {
+          dictionary: unknown;
+        };
+        expect(secondType.dictionary).toBe(firstType.dictionary);
+        expect(sanitized.batches[0].data.children[1].dictionary).toBe(
+          sanitized.batches[0].data.children[0].dictionary,
+        );
+
+        const buf = await fromDataToBuffer(table);
+        const actual = currentTableFromIPC(buf);
+        expect([...actual.getChild("first")!]).toEqual(values);
+        expect([...actual.getChild("second")!]).toEqual(values);
+      });
+
+      it("preserves shared dictionary data from another Arrow version", async function () {
+        const values = ["alpha", "beta", "alpha"];
+        const dictionaryVector = vectorFromArray(values);
+        const firstBatch = new ArrowRecordBatch({
+          label: dictionaryVector.slice(0, 2).data[0],
+        });
+        const secondBatch = new ArrowRecordBatch({
+          label: dictionaryVector.slice(2).data[0],
+        });
+        const table = new ArrowTable([firstBatch, secondBatch]);
+
+        const sanitized = sanitizeTable(table);
+        expect([...sanitized.getChild("label")!]).toEqual(values);
+
+        const dictionaries = sanitized.batches.map(
+          (batch) => batch.data.children[0].dictionary,
+        );
+        expect(dictionaries[0]).toBeInstanceOf(CurrentVector);
+        expect(dictionaries[1]).toBe(dictionaries[0]);
+
+        const buf = await fromDataToBuffer(table);
+        const actual = currentTableFromIPC(buf);
+        expect([...actual.getChild("label")!]).toEqual(values);
+      });
+
+      it("preserves shared chunks in growing dictionaries", async function () {
+        const type = new Dictionary(new Utf8(), new Int32(), 42, false);
+        const firstDictionary = vectorFromArray(["alpha", "beta"], new Utf8());
+        const secondDictionary = firstDictionary.concat(
+          vectorFromArray(["gamma"], new Utf8()),
+        );
+        const firstData = arrowMakeData({
+          type,
+          data: Int32Array.from([0, 1]),
+          dictionary: firstDictionary,
+        });
+        const secondData = arrowMakeData({
+          type,
+          data: Int32Array.from([2]),
+          dictionary: secondDictionary,
+        });
+        const table = new ArrowTable([
+          new ArrowRecordBatch({ label: firstData }),
+          new ArrowRecordBatch({ label: secondData }),
+        ]);
+
+        const sanitized = sanitizeTable(table);
+        const expected = ["alpha", "beta", "gamma"];
+        expect([...sanitized.getChild("label")!]).toEqual(expected);
+        const firstLocalDictionary =
+          sanitized.batches[0].data.children[0].dictionary!;
+        const secondLocalDictionary =
+          sanitized.batches[1].data.children[0].dictionary!;
+        expect(secondLocalDictionary.data[0]).toBe(
+          firstLocalDictionary.data[0],
+        );
+
+        const buf = await fromTableToBuffer(sanitized);
+        const actual = currentTableFromIPC(buf);
+        expect([...actual.getChild("label")!]).toEqual(expected);
+      });
+
+      it("can serialize list data from another Arrow version", async function () {
+        const values = [["anime", "action"], [], null];
+        const vector = vectorFromArray(
+          values,
+          new List(new Field("item", new Utf8(), true)),
+        );
+        const table = new ArrowTable({ tags: vector });
+
+        const buf = await fromDataToBuffer(table);
+        const actual = currentTableFromIPC(buf);
+        const actualTags = actual.getChild("tags");
+
+        expect(actualTags?.get(0)?.toJSON()).toEqual(values[0]);
+        expect(actualTags?.get(1)?.toJSON()).toEqual(values[1]);
+        expect(actualTags?.get(2)).toBeNull();
+      });
+
       it("can still import data", async function () {
         const schema = new arrow15.Schema([
           new arrow15.Field("id", new arrow15.Int32()),

nodejs/__test__/table.test.ts

@@ -277,8 +277,16 @@ describe.each([arrow15, arrow16, arrow17, arrow18])(
         },
         numIndices: 0,
         numRows: 3,
-        totalBytes: 44,
+        // Full on-disk size of the two data files, footers and metadata included.
+        totalBytes: 684,
       });
+
+      // Index files count toward totalBytes too (only deletion files and
+      // manifests are excluded).
+      await table.createIndex("id", { config: Index.btree() });
+      const statsWithIndex = await table.stats();
+      expect(statsWithIndex.numIndices).toBe(1);
+      expect(statsWithIndex.totalBytes).toBeGreaterThan(684);
     });
 
     it("should overwrite data if asked", async () => {

nodejs/lancedb/sanitize.ts

@@ -9,7 +9,7 @@
 // comes from the exact same library instance.  This is not always the case
 // and so we must sanitize the input to ensure that it is compatible.
 
-import { BufferType, Data } from "apache-arrow";
+import { BufferType, Data, Vector } from "apache-arrow";
 import type { IntBitWidth, TKeys, TimeBitWidth } from "apache-arrow/type";
 import {
   Binary,
@@ -74,6 +74,20 @@ import {
   Utf8,
 } from "./arrow";
 
+type SanitizationContext = {
+  types: WeakMap<object, DataType>;
+  vectors: WeakMap<object, Vector>;
+  data: WeakMap<object, Data<DataType>>;
+};
+
+function createSanitizationContext(): SanitizationContext {
+  return {
+    types: new WeakMap(),
+    vectors: new WeakMap(),
+    data: new WeakMap(),
+  };
+}
+
 export function sanitizeMetadata(
   metadataLike?: unknown,
 ): Map<string, string> | undefined {
@@ -186,6 +200,13 @@ export function sanitizeInterval(typeLike: object) {
 }
 
 export function sanitizeList(typeLike: object) {
+  return sanitizeListWithContext(typeLike, createSanitizationContext());
+}
+
+function sanitizeListWithContext(
+  typeLike: object,
+  context: SanitizationContext,
+) {
   if (!("children" in typeLike) || !Array.isArray(typeLike.children)) {
     throw Error(
       "Expected a List type to have an array-like `children` property",
@@ -194,19 +215,35 @@ export function sanitizeList(typeLike: object) {
   if (typeLike.children.length !== 1) {
     throw Error("Expected a List type to have exactly one child");
   }
-  return new List(sanitizeField(typeLike.children[0]));
+  return new List(sanitizeFieldWithContext(typeLike.children[0], context));
 }
 
 export function sanitizeStruct(typeLike: object) {
+  return sanitizeStructWithContext(typeLike, createSanitizationContext());
+}
+
+function sanitizeStructWithContext(
+  typeLike: object,
+  context: SanitizationContext,
+) {
   if (!("children" in typeLike) || !Array.isArray(typeLike.children)) {
     throw Error(
       "Expected a Struct type to have an array-like `children` property",
     );
   }
-  return new Struct(typeLike.children.map((child) => sanitizeField(child)));
+  return new Struct(
+    typeLike.children.map((child) => sanitizeFieldWithContext(child, context)),
+  );
 }
 
 export function sanitizeUnion(typeLike: object) {
+  return sanitizeUnionWithContext(typeLike, createSanitizationContext());
+}
+
+function sanitizeUnionWithContext(
+  typeLike: object,
+  context: SanitizationContext,
+) {
   if (
     !("typeIds" in typeLike) ||
     !("mode" in typeLike) ||
@@ -226,7 +263,7 @@ export function sanitizeUnion(typeLike: object) {
     typeLike.mode,
     // biome-ignore lint/suspicious/noExplicitAny: skip
     typeLike.typeIds as any,
-    typeLike.children.map((child) => sanitizeField(child)),
+    typeLike.children.map((child) => sanitizeFieldWithContext(child, context)),
   );
 }
 
@@ -234,6 +271,19 @@ export function sanitizeTypedUnion(
   typeLike: object,
   // eslint-disable-next-line @typescript-eslint/naming-convention
   UnionType: typeof DenseUnion | typeof SparseUnion,
+) {
+  return sanitizeTypedUnionWithContext(
+    typeLike,
+    UnionType,
+    createSanitizationContext(),
+  );
+}
+
+function sanitizeTypedUnionWithContext(
+  typeLike: object,
+  // eslint-disable-next-line @typescript-eslint/naming-convention
+  UnionType: typeof DenseUnion | typeof SparseUnion,
+  context: SanitizationContext,
 ) {
   if (!("typeIds" in typeLike)) {
     throw Error(
@@ -248,7 +298,7 @@ export function sanitizeTypedUnion(
 
   return new UnionType(
     typeLike.typeIds as Int32Array | number[],
-    typeLike.children.map((child) => sanitizeField(child)),
+    typeLike.children.map((child) => sanitizeFieldWithContext(child, context)),
   );
 }
 
@@ -262,6 +312,16 @@ export function sanitizeFixedSizeBinary(typeLike: object) {
 }
 
 export function sanitizeFixedSizeList(typeLike: object) {
+  return sanitizeFixedSizeListWithContext(
+    typeLike,
+    createSanitizationContext(),
+  );
+}
+
+function sanitizeFixedSizeListWithContext(
+  typeLike: object,
+  context: SanitizationContext,
+) {
   if (!("listSize" in typeLike) || typeof typeLike.listSize !== "number") {
     throw Error("Expected a FixedSizeList type to have a `listSize` property");
   }
@@ -275,11 +335,18 @@ export function sanitizeFixedSizeList(typeLike: object) {
   }
   return new FixedSizeList(
     typeLike.listSize,
-    sanitizeField(typeLike.children[0]),
+    sanitizeFieldWithContext(typeLike.children[0], context),
   );
 }
 
 export function sanitizeMap(typeLike: object) {
+  return sanitizeMapWithContext(typeLike, createSanitizationContext());
+}
+
+function sanitizeMapWithContext(
+  typeLike: object,
+  context: SanitizationContext,
+) {
   if (!("children" in typeLike) || !Array.isArray(typeLike.children)) {
     throw Error(
       "Expected a Map type to have an array-like `children` property",
@@ -292,7 +359,10 @@ export function sanitizeMap(typeLike: object) {
     throw Error("Expected a Map type to have exactly one child");
   }
 
-  return new Map_(sanitizeField(typeLike.children[0]), typeLike.keysSorted);
+  return new Map_(
+    sanitizeFieldWithContext(typeLike.children[0], context),
+    typeLike.keysSorted,
+  );
 }
 
 export function sanitizeDuration(typeLike: object) {
@@ -303,6 +373,13 @@ export function sanitizeDuration(typeLike: object) {
 }
 
 export function sanitizeDictionary(typeLike: object) {
+  return sanitizeDictionaryWithContext(typeLike, createSanitizationContext());
+}
+
+function sanitizeDictionaryWithContext(
+  typeLike: object,
+  context: SanitizationContext,
+) {
   if (!("id" in typeLike) || typeof typeLike.id !== "number") {
     throw Error("Expected a Dictionary type to have an `id` property");
   }
@@ -316,8 +393,8 @@ export function sanitizeDictionary(typeLike: object) {
     throw Error("Expected a Dictionary type to have an `isOrdered` property");
   }
   return new Dictionary(
-    sanitizeType(typeLike.dictionary),
-    sanitizeType(typeLike.indices) as TKeys,
+    sanitizeTypeWithContext(typeLike.dictionary, context),
+    sanitizeTypeWithContext(typeLike.indices, context) as TKeys,
     typeLike.id,
     typeLike.isOrdered,
   );
@@ -325,12 +402,23 @@ export function sanitizeDictionary(typeLike: object) {
 
 // biome-ignore lint/suspicious/noExplicitAny: skip
 export function sanitizeType(typeLike: unknown): DataType<any> {
+  return sanitizeTypeWithContext(typeLike, createSanitizationContext());
+}
+
+function sanitizeTypeWithContext(
+  typeLike: unknown,
+  context: SanitizationContext,
+): DataType {
   if (typeof typeLike === "string") {
     return dataTypeFromName(typeLike);
   }
   if (typeof typeLike !== "object" || typeLike === null) {
     throw Error("Expected a Type but object was null/undefined");
   }
+  const cached = context.types.get(typeLike);
+  if (cached !== undefined) {
+    return cached;
+  }
   if (
     !("typeId" in typeLike) ||
     !(
@@ -349,6 +437,16 @@ export function sanitizeType(typeLike: unknown): DataType<any> {
     throw Error("Type's typeId property was not a function or number");
   }
 
+  const type = sanitizeTypeById(typeLike, typeId, context);
+  context.types.set(typeLike, type);
+  return type;
+}
+
+function sanitizeTypeById(
+  typeLike: object,
+  typeId: Type,
+  context: SanitizationContext,
+): DataType {
   switch (typeId) {
     case Type.NONE:
       throw Error("Received a Type with a typeId of NONE");
@@ -375,21 +473,21 @@ export function sanitizeType(typeLike: unknown): DataType<any> {
     case Type.Interval:
       return sanitizeInterval(typeLike);
     case Type.List:
-      return sanitizeList(typeLike);
+      return sanitizeListWithContext(typeLike, context);
     case Type.Struct:
-      return sanitizeStruct(typeLike);
+      return sanitizeStructWithContext(typeLike, context);
     case Type.Union:
-      return sanitizeUnion(typeLike);
+      return sanitizeUnionWithContext(typeLike, context);
     case Type.FixedSizeBinary:
       return sanitizeFixedSizeBinary(typeLike);
     case Type.FixedSizeList:
-      return sanitizeFixedSizeList(typeLike);
+      return sanitizeFixedSizeListWithContext(typeLike, context);
     case Type.Map:
-      return sanitizeMap(typeLike);
+      return sanitizeMapWithContext(typeLike, context);
     case Type.Duration:
       return sanitizeDuration(typeLike);
     case Type.Dictionary:
-      return sanitizeDictionary(typeLike);
+      return sanitizeDictionaryWithContext(typeLike, context);
     case Type.Int8:
       return new Int8();
     case Type.Int16:
@@ -433,9 +531,9 @@ export function sanitizeType(typeLike: unknown): DataType<any> {
     case Type.TimestampSecond:
       return sanitizeTypedTimestamp(typeLike, TimestampSecond);
     case Type.DenseUnion:
-      return sanitizeTypedUnion(typeLike, DenseUnion);
+      return sanitizeTypedUnionWithContext(typeLike, DenseUnion, context);
     case Type.SparseUnion:
-      return sanitizeTypedUnion(typeLike, SparseUnion);
+      return sanitizeTypedUnionWithContext(typeLike, SparseUnion, context);
     case Type.IntervalDayTime:
       return new IntervalDayTime();
     case Type.IntervalYearMonth:
@@ -454,6 +552,13 @@ export function sanitizeType(typeLike: unknown): DataType<any> {
 }
 
 export function sanitizeField(fieldLike: unknown): Field {
+  return sanitizeFieldWithContext(fieldLike, createSanitizationContext());
+}
+
+function sanitizeFieldWithContext(
+  fieldLike: unknown,
+  context: SanitizationContext,
+): Field {
   if (fieldLike instanceof Field) {
     return fieldLike;
   }
@@ -471,7 +576,7 @@ export function sanitizeField(fieldLike: unknown): Field {
   }
   let type: DataType;
   try {
-    type = sanitizeType(fieldLike.type);
+    type = sanitizeTypeWithContext(fieldLike.type, context);
   } catch (error: unknown) {
     throw Error(
       `Unable to sanitize type for field: ${fieldLike.name} due to error: ${error}`,
@@ -501,6 +606,13 @@ export function sanitizeField(fieldLike: unknown): Field {
  * than lancedb is using.
  */
 export function sanitizeSchema(schemaLike: SchemaLike): Schema {
+  return sanitizeSchemaWithContext(schemaLike, createSanitizationContext());
+}
+
+function sanitizeSchemaWithContext(
+  schemaLike: SchemaLike,
+  context: SanitizationContext,
+): Schema {
   if (schemaLike instanceof Schema) {
     return schemaLike;
   }
@@ -522,7 +634,7 @@ export function sanitizeSchema(schemaLike: SchemaLike): Schema {
     );
   }
   const sanitizedFields = schemaLike.fields.map((field) =>
-    sanitizeField(field),
+    sanitizeFieldWithContext(field, context),
   );
   return new Schema(sanitizedFields, metadata);
 }
@@ -544,13 +656,18 @@ export function sanitizeTable(tableLike: TableLike): Table {
       "The table passed in does not appear to be a table (no 'columns' property)",
     );
   }
-  const schema = sanitizeSchema(tableLike.schema);
-
-  const batches = tableLike.batches.map(sanitizeRecordBatch);
+  const context = createSanitizationContext();
+  const schema = sanitizeSchemaWithContext(tableLike.schema, context);
+  const batches = tableLike.batches.map((batch) =>
+    sanitizeRecordBatch(batch, context),
+  );
   return new Table(schema, batches);
 }
 
-function sanitizeRecordBatch(batchLike: RecordBatchLike): RecordBatch {
+function sanitizeRecordBatch(
+  batchLike: RecordBatchLike,
+  context: SanitizationContext,
+): RecordBatch {
   if (batchLike instanceof RecordBatch) {
     return batchLike;
   }
@@ -567,19 +684,43 @@ function sanitizeRecordBatch(batchLike: RecordBatchLike): RecordBatch {
       "The record batch passed in does not appear to be a record batch (no 'data' property)",
     );
   }
-  const schema = sanitizeSchema(batchLike.schema);
-  const data = sanitizeData(batchLike.data);
+  const schema = sanitizeSchemaWithContext(batchLike.schema, context);
+  const data = sanitizeData(batchLike.data, context) as Data<Struct>;
   return new RecordBatch(schema, data);
 }
+
+type DictionaryVectorLike = {
+  data: readonly DataLike[];
+};
+
+type DictionaryDataLike = DataLike & {
+  dictionary?: DictionaryVectorLike;
+};
+
 function sanitizeData(
   dataLike: DataLike,
-  // biome-ignore lint/suspicious/noExplicitAny: <explanation>
-): import("apache-arrow").Data<Struct<any>> {
+  context: SanitizationContext,
+): Data<DataType> {
   if (dataLike instanceof Data) {
     return dataLike;
   }
-  return new Data(
-    dataLike.type,
+  const cachedData = context.data.get(dataLike);
+  if (cachedData !== undefined) {
+    return cachedData;
+  }
+  const dictionaryLike = (dataLike as DictionaryDataLike).dictionary;
+  let dictionary: Vector | undefined;
+  if (dictionaryLike !== undefined) {
+    dictionary = context.vectors.get(dictionaryLike);
+    if (dictionary === undefined) {
+      dictionary = new Vector(
+        dictionaryLike.data.map((data) => sanitizeData(data, context)),
+      );
+      context.vectors.set(dictionaryLike, dictionary);
+    }
+  }
+  const data = new Data(
+    sanitizeTypeWithContext(dataLike.type, context),
     dataLike.offset,
     dataLike.length,
     dataLike.nullCount,
@@ -589,7 +730,11 @@ function sanitizeData(
       [BufferType.VALIDITY]: dataLike.nullBitmap,
       [BufferType.TYPE]: dataLike.typeIds,
     },
+    dataLike.children.map((child) => sanitizeData(child, context)),
+    dictionary,
   );
+  context.data.set(dataLike, data);
+  return data;
 }
 
 const constructorsByTypeName = {

nodejs/lancedb/table.ts

@@ -197,7 +197,11 @@ export interface LsmWriteSpec {
   column?: string;
   /** Bucket variant: the number of buckets, in `[1, 1024]`. */
   numBuckets?: number;
-  /** Names of indexes the MemWAL should keep up to date during writes. */
+  /**
+   * Indexes the MemWAL keeps up to date. Omit to maintain every supported
+   * index, resolved on install — a snapshot, so indexes created later are not
+   * maintained. Pass `[]` for none.
+   */
   maintainedIndexes?: string[];
   /** Default `ShardWriter` configuration recorded in the MemWAL index. */
   writerConfigDefaults?: Record<string, string>;
@@ -595,6 +599,11 @@ export abstract class Table {
    * All variants require the table to have an unenforced primary key
    * ({@link Table#setUnenforcedPrimaryKey}); bucket sharding additionally
    * requires it to be the single column being bucketed.
+   *
+   * Omitting `maintainedIndexes` maintains every index on the table, resolved
+   * here, failing if one cannot be maintained — name them to install anyway.
+   * Naming them pins an exact set, and a still-building index is rejected
+   * rather than quietly omitted.
    * @param {LsmWriteSpec} spec The sharding spec to install.
    * @returns {Promise<void>}
    * @example
@@ -622,9 +631,10 @@ export abstract class Table {
    *
    * Resolves to `undefined` when the MemWAL LSM write path is not enabled (no
    * spec has been set, or it was removed with {@link Table#unsetLsmWriteSpec}).
-   * The returned spec — including its `maintainedIndexes` and
-   * `writerConfigDefaults` — mirrors what was passed to
-   * {@link Table#setLsmWriteSpec}.
+   * The returned spec mirrors what was passed to
+   * {@link Table#setLsmWriteSpec}, except that `maintainedIndexes` always
+   * reports the concrete list resolved when the spec was set — `undefined`
+   * never round-trips.
    * @returns {Promise<LsmWriteSpec | undefined>}
    */
   abstract getLsmWriteSpec(): Promise<LsmWriteSpec | undefined>;

nodejs/npm/darwin-arm64/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-darwin-arm64",
-	"version": "0.37.1-beta.0",
+	"version": "0.37.1-beta.1",
 	"os": ["darwin"],
 	"cpu": ["arm64"],
 	"main": "lancedb.darwin-arm64.node",

nodejs/npm/linux-arm64-gnu/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-gnu",
-	"version": "0.37.1-beta.0",
+	"version": "0.37.1-beta.1",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-gnu.node",

nodejs/npm/linux-arm64-musl/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-musl",
-	"version": "0.37.1-beta.0",
+	"version": "0.37.1-beta.1",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-musl.node",

nodejs/npm/linux-x64-gnu/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-gnu",
-	"version": "0.37.1-beta.0",
+	"version": "0.37.1-beta.1",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-gnu.node",

nodejs/npm/linux-x64-musl/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-musl",
-	"version": "0.37.1-beta.0",
+	"version": "0.37.1-beta.1",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-musl.node",

nodejs/npm/win32-arm64-msvc/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@lancedb/lancedb-win32-arm64-msvc",
-  "version": "0.37.1-beta.0",
+  "version": "0.37.1-beta.1",
   "os": [
     "win32"
   ],

nodejs/npm/win32-x64-msvc/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-win32-x64-msvc",
-	"version": "0.37.1-beta.0",
+	"version": "0.37.1-beta.1",
 	"os": ["win32"],
 	"cpu": ["x64"],
 	"main": "lancedb.win32-x64-msvc.node",

nodejs/package-lock.json

@@ -1,12 +1,12 @@
 {
   "name": "@lancedb/lancedb",
-  "version": "0.37.1-beta.0",
+  "version": "0.37.1-beta.1",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "@lancedb/lancedb",
-      "version": "0.37.1-beta.0",
+      "version": "0.37.1-beta.1",
       "cpu": [
         "x64",
         "arm64"

nodejs/package.json

@@ -11,7 +11,7 @@
     "ann"
   ],
   "private": false,
-  "version": "0.37.1-beta.0",
+  "version": "0.37.1-beta.1",
   "main": "dist/index.js",
   "exports": {
     ".": "./dist/index.js",

nodejs/src/table.rs

@@ -772,7 +772,8 @@ pub struct LsmWriteSpec {
     pub column: Option<String>,
     /// Bucket variant: the number of buckets, in `[1, 1024]`.
     pub num_buckets: Option<u32>,
-    /// Names of indexes the MemWAL should keep up to date during writes.
+    /// Indexes the MemWAL keeps up to date. Omitted resolves every
+    /// maintainable index on install; an empty array means none.
     pub maintained_indexes: Option<Vec<String>>,
     /// Default `ShardWriter` configuration recorded in the MemWAL index.
     pub writer_config_defaults: Option<HashMap<String, String>>,
@@ -782,7 +783,6 @@ impl TryFrom<LsmWriteSpec> for lancedb::table::LsmWriteSpec {
     type Error = napi::Error;
 
     fn try_from(value: LsmWriteSpec) -> napi::Result<Self> {
-        let maintained = value.maintained_indexes.unwrap_or_default();
         let writer_config_defaults = value.writer_config_defaults.unwrap_or_default();
         let spec = match value.spec_type.as_str() {
             "bucket" => {
@@ -809,7 +809,7 @@ impl TryFrom<LsmWriteSpec> for lancedb::table::LsmWriteSpec {
             }
         };
         Ok(spec
-            .with_maintained_indexes(maintained)
+            .with_maintained_indexes(value.maintained_indexes)
             .with_writer_config_defaults(writer_config_defaults))
     }
 }
@@ -827,7 +827,7 @@ impl From<lancedb::table::LsmWriteSpec> for LsmWriteSpec {
                 spec_type: "bucket".to_string(),
                 column: Some(column),
                 num_buckets: Some(num_buckets),
-                maintained_indexes: Some(maintained_indexes),
+                maintained_indexes,
                 writer_config_defaults: Some(writer_config_defaults),
             },
             Native::Identity {
@@ -838,7 +838,7 @@ impl From<lancedb::table::LsmWriteSpec> for LsmWriteSpec {
                 spec_type: "identity".to_string(),
                 column: Some(column),
                 num_buckets: None,
-                maintained_indexes: Some(maintained_indexes),
+                maintained_indexes,
                 writer_config_defaults: Some(writer_config_defaults),
             },
             Native::Unsharded {
@@ -848,7 +848,7 @@ impl From<lancedb::table::LsmWriteSpec> for LsmWriteSpec {
                 spec_type: "unsharded".to_string(),
                 column: None,
                 num_buckets: None,
-                maintained_indexes: Some(maintained_indexes),
+                maintained_indexes,
                 writer_config_defaults: Some(writer_config_defaults),
             },
         }
@@ -1043,7 +1043,10 @@ impl From<lancedb::index::IndexStatistics> for IndexStatistics {
 
 #[napi(object)]
 pub struct TableStatistics {
-    /// The total number of bytes in the table
+    /// The total size, in bytes, of the table's data files, index files, and
+    /// overlay files
+    ///
+    /// Read from the manifest, so this excludes deletion files and manifests.
     pub total_bytes: i64,
 
     /// The number of rows in the table

python/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "lancedb-python"
-version = "0.37.1-beta.0"
+version = "0.37.1-beta.1"
 publish = false
 edition.workspace = true
 description = "Python bindings for LanceDB"

python/python/lancedb/_lancedb.pyi

@@ -653,9 +653,10 @@ class LsmWriteSpec:
     def identity(column: str) -> "LsmWriteSpec": ...
     @staticmethod
     def unsharded() -> "LsmWriteSpec": ...
-    def with_maintained_indexes(self, indexes: List[str]) -> "LsmWriteSpec":
-        """Return a copy of this spec asking the MemWAL to keep the named
-        indexes up to date as rows are appended."""
+    def with_maintained_indexes(self, indexes: Optional[List[str]]) -> "LsmWriteSpec":
+        """Set which indexes the MemWAL keeps up to date. None resolves every
+        index on the table at install, failing if one cannot be maintained;
+        a list is verbatim, empty means none."""
         ...
     def with_writer_config_defaults(self, defaults: Dict[str, str]) -> "LsmWriteSpec":
         """Return a copy of this spec recording the given default
@@ -670,7 +671,9 @@ class LsmWriteSpec:
     @property
     def num_buckets(self) -> Optional[int]: ...
     @property
-    def maintained_indexes(self) -> List[str]: ...
+    def maintained_indexes(self) -> Optional[List[str]]:
+        """Indexes the MemWAL keeps up to date, or None for every supported one."""
+        ...
     @property
     def writer_config_defaults(self) -> Dict[str, str]: ...
 

python/python/lancedb/embeddings/jinaai.py

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

python/python/lancedb/streaming.py

@@ -11,6 +11,11 @@ Provides StreamingDataset, a PyTorch IterableDataset that guarantees:
 - **Resumability**: state_dict / load_state_dict capture per-split consumption
   counts so training can resume from an exact mid-epoch position even when the
   distributed topology changes between runs.
+
+Transform failures on bad rows (e.g. nulls or NaNs from incomplete data) can
+be tolerated with ``on_transform_error="skip"``; see the parameter
+documentation on StreamingDataset for how this interacts with the guarantees
+above.
 """
 
 import ctypes
@@ -22,7 +27,7 @@ import time
 from collections import deque
 from concurrent.futures import ThreadPoolExecutor
 from multiprocessing import RawArray
-from typing import Any, Callable, Iterator, Optional
+from typing import Any, Callable, Iterator, Optional, Union
 
 from torch.utils.data import IterableDataset, get_worker_info
 
@@ -127,6 +132,49 @@ class StreamingDataset(IterableDataset):
         Maximum number of transforms to run concurrently.  Must be greater
         than zero.  When ``None`` (the default), uses ``os.cpu_count()`` or 1
         when the CPU count is unavailable.
+    on_transform_error:
+        What to do when the transform raises an exception:
+
+        - ``"raise"`` (the default): the exception propagates and iteration
+          aborts.
+        - ``"skip"``: the failing rows are dropped and iteration continues.
+        - ``"warn"``: like ``"skip"``, but a warning is logged for each
+          failing batch.
+        - a callable ``handler(exc) -> bool``: called with the exception;
+          return ``True`` to skip the failing rows or ``False`` to re-raise.
+          Useful to skip only expected error types (compatible with
+          ``webdataset.handlers`` style handlers).
+
+        When a batch fails, the transform is re-invoked on each single-row
+        slice of the batch so that only the rows that actually fail are
+        dropped.  Transforms should therefore be deterministic and accept
+        batches of any size (including one row).  Skipped rows are counted in
+        ``rows_skipped``.
+
+        Skipping weakens the elastic-determinism guarantee at the end of the
+        epoch: splits that lose more rows than others run dry earlier, and
+        each rank's iterator ends at the last cycle where every split *it
+        owns* still has a row.  Because bad rows are not distributed evenly
+        across splits, this means one rank's iterator can yield noticeably
+        fewer or more steps than another rank's *in the same run* — there is
+        no cross-rank coordination that stops every rank at the same global
+        step.  This is generally safe for asynchronous or single-rank use,
+        but synchronous distributed training (e.g. ranks that call
+        ``all_reduce`` every step) can hang or deadlock if one rank's
+        iterator is exhausted while others are still stepping; callers doing
+        synchronous multi-rank training with ``on_transform_error != "raise"``
+        are responsible for their own cross-rank stopping mechanism (e.g.
+        broadcasting a stop signal on ``StopIteration``).  The final few
+        global steps can also differ across topologies (bounded by the skew
+        in bad-row counts across splits).  The sequence of samples yielded
+        from each split remains deterministic.  Mid-epoch
+        checkpoints remain exact provided the transform fails
+        deterministically; in multi-rank training each rank must save its
+        own ``state_dict`` and the states must be combined with
+        ``merge_state_dicts`` before resuming on a different topology.
+        Prefer the ``filter`` parameter when bad rows can be expressed as a
+        SQL predicate (e.g. ``"col IS NOT NULL"``) — filtering happens before
+        splits are built, so every guarantee is fully preserved.
     worker_info_override:
         If set, used in place of ``torch.utils.data.get_worker_info()`` to
         determine the DataLoader worker assignment.  Intended for unit tests
@@ -152,6 +200,7 @@ class StreamingDataset(IterableDataset):
         filter: Optional[str] = None,
         transform: Optional[Callable] = None,
         transform_parallelism: Optional[int] = None,
+        on_transform_error: Union[str, Callable[[Exception], bool]] = "raise",
         connection_factory: Optional[Callable[[str], Any]] = None,
         worker_info_override=None,
     ):
@@ -167,6 +216,13 @@ class StreamingDataset(IterableDataset):
             )
         if transform_parallelism is not None and transform_parallelism <= 0:
             raise ValueError("transform_parallelism must be greater than 0")
+        if on_transform_error not in ("raise", "skip", "warn") and not callable(
+            on_transform_error
+        ):
+            raise ValueError(
+                "on_transform_error must be 'raise', 'skip', 'warn', or a "
+                f"callable, got {on_transform_error!r}"
+            )
 
         self._table = table
         self._num_splits = num_splits
@@ -182,6 +238,7 @@ class StreamingDataset(IterableDataset):
         self._filter = filter
         self._transform = transform
         self._transform_parallelism = transform_parallelism
+        self._on_transform_error = on_transform_error
         self._connection_factory = connection_factory
         self._worker_info_override = worker_info_override
 
@@ -199,19 +256,28 @@ class StreamingDataset(IterableDataset):
         # in the main process.  RawArray is picklable via the forkserver
         # reduction protocol so it survives the dataset pickle round-trip.
         # Layout: [unscanned_rows, raw_rows, cooked_rows, consumed_rows,
-        #          bytes_loaded, fetch_time_us, transform_time_us]
-        self._worker_stats: RawArray = RawArray(ctypes.c_int64, 7)
+        #          bytes_loaded, fetch_time_us, transform_time_us,
+        #          rows_skipped]
+        self._worker_stats: RawArray = RawArray(ctypes.c_int64, 8)
 
         # Cumulative bytes of Arrow buffer data fetched across all iterations.
         self._bytes_loaded: int = 0
         # Cumulative seconds spent in LanceDB I/O and in transform functions.
         self._fetch_time: float = 0.0
         self._transform_time: float = 0.0
+        # Cumulative rows dropped by on_transform_error across all iterations.
+        self._rows_skipped: int = 0
 
         # Number of samples each split has already been consumed.  At global
         # step boundaries all splits have consumed this many samples, so a
         # single scalar captures the topology-independent checkpoint state.
         self._resume_offset: int = 0
+        # Permutation position each split has consumed through, keyed by
+        # global split index.  Equal to _resume_offset for every split unless
+        # on_transform_error skipped rows, in which case skipped positions
+        # push the watermark of the affected splits further ahead.  Splits
+        # this instance has never iterated have no entry.
+        self._resume_positions: dict[int, int] = {}
 
         # Build the permutation table once, deterministically.
         builder = permutation_builder(table)
@@ -275,6 +341,7 @@ class StreamingDataset(IterableDataset):
         # Set identity transform on each Permutation so __getitems__ returns
         # the raw RecordBatch.  Stage 2 applies the real transform.
         permutations: list[Permutation] = []
+        initial_positions: list[int] = []
         for split_idx in my_splits:
             perm = Permutation.from_tables(
                 self._table, self._perm_table, split=split_idx
@@ -282,14 +349,20 @@ class StreamingDataset(IterableDataset):
             if self._columns is not None:
                 perm = perm.select_columns(self._columns)
             perm = perm.with_transform(lambda batch: batch)
-            if self._resume_offset > 0:
-                perm = perm.with_skip(self._resume_offset)
+            start_pos = self._resume_positions.get(split_idx, self._resume_offset)
+            if start_pos > 0:
+                perm = perm.with_skip(start_pos)
+            initial_positions.append(start_pos)
             permutations.append(perm)
 
         n = len(permutations)
         split_sizes = [perm.num_rows for perm in permutations]
         initial_offset = self._resume_offset
         local_consumed = [0] * n
+        # Permutation position each split has consumed through (absolute,
+        # i.e. counted from the start of the unskipped split).  Runs ahead of
+        # initial + local_consumed when rows are skipped.
+        pos_consumed = list(initial_positions)
 
         batch_size = self._read_batch_size
         max_prefetch = self._prefetch_batches
@@ -302,12 +375,14 @@ class StreamingDataset(IterableDataset):
             self._transform if self._transform is not None else Transforms.arrow2python
         )
 
-        # Per-split pipeline state.
+        # Per-split pipeline state.  Batches are paired with the absolute
+        # permutation position of their first row so that skipped rows can be
+        # accounted for in pos_consumed.
         fetch_head = [0] * n
-        io_pending = [deque() for _ in range(n)]  # Future[RecordBatch]
-        raw_batches = [deque() for _ in range(n)]  # RecordBatch — fetched, awaiting tx
-        tx_pending = [deque() for _ in range(n)]  # Future[list[Any]]
-        cooked = [deque() for _ in range(n)]  # rows ready to yield
+        io_pending = [deque() for _ in range(n)]  # (abs_start, Future[RecordBatch])
+        raw_batches = [deque() for _ in range(n)]  # (abs_start, RecordBatch)
+        tx_pending = [deque() for _ in range(n)]  # Future[list[(abs_pos, row)]]
+        cooked = [deque() for _ in range(n)]  # (abs_pos, row) ready to yield
 
         # Limit simultaneous transforms to transform_workers across all splits.
         tx_semaphore = threading.Semaphore(transform_workers)
@@ -330,7 +405,8 @@ class StreamingDataset(IterableDataset):
             fetch_head[i] += fetch
             perm_i = permutations[i]
             indices = list(range(start, start + fetch))
-            io_pending[i].append(io_pool.submit(_io_call, perm_i, indices))
+            abs_start = initial_positions[i] + start
+            io_pending[i].append((abs_start, io_pool.submit(_io_call, perm_i, indices)))
 
         def _fill_io(i: int) -> None:
             while len(io_pending[i]) < max_prefetch and fetch_head[i] < split_sizes[i]:
@@ -338,15 +414,72 @@ class StreamingDataset(IterableDataset):
 
         def _drain_io(i: int) -> None:
             """Move completed I/O futures into raw_batches non-blockingly."""
-            while io_pending[i] and io_pending[i][0].done():
-                raw_batches[i].append(io_pending[i].popleft().result())
+            while io_pending[i] and io_pending[i][0][1].done():
+                abs_start, fut = io_pending[i].popleft()
+                raw_batches[i].append((abs_start, fut.result()))
 
         # ── Stage 2 helpers ───────────────────────────────────────────────────
 
-        def _tx_call_guarded(batch):
+        on_error = self._on_transform_error
+
+        def _should_skip(exc: Exception) -> bool:
+            if on_error == "raise":
+                return False
+            if callable(on_error):
+                return bool(on_error(exc))
+            return True  # "skip" or "warn"
+
+        def _check_row_count(rows: list, num_rows: int) -> None:
+            if len(rows) != num_rows:
+                raise ValueError(
+                    f"transform returned {len(rows)} rows for a batch of "
+                    f"{num_rows}; transforms must return exactly one output "
+                    "row per input row.  To drop bad rows, raise inside the "
+                    "transform and pass on_transform_error='skip'."
+                )
+
+        def _transform_isolated(abs_start, batch, batch_exc):
+            """Re-run the transform on single-row slices, dropping failures."""
+            out = []
+            skipped = 0
+            first_exc = None
+            for j in range(batch.num_rows):
+                try:
+                    rows = list(final_transform(batch.slice(j, 1)))
+                except Exception as exc:
+                    if not _should_skip(exc):
+                        raise
+                    skipped += 1
+                    if first_exc is None:
+                        first_exc = exc
+                    continue
+                _check_row_count(rows, 1)
+                out.append((abs_start + j, rows[0]))
+            self._rows_skipped += skipped
+            if skipped and on_error == "warn":
+                logger.warning(
+                    "Skipped %d of %d rows whose transform failed (first error: %r)",
+                    skipped,
+                    batch.num_rows,
+                    first_exc if first_exc is not None else batch_exc,
+                )
+            return out
+
+        def _transform_batch(abs_start, batch):
+            """Apply the transform, returning [(abs_pos, row), ...]."""
+            try:
+                rows = list(final_transform(batch))
+            except Exception as exc:
+                if not _should_skip(exc):
+                    raise
+                return _transform_isolated(abs_start, batch, exc)
+            _check_row_count(rows, batch.num_rows)
+            return [(abs_start + j, row) for j, row in enumerate(rows)]
+
+        def _tx_call_guarded(abs_start, batch):
             try:
                 t0 = time.perf_counter()
-                result = final_transform(batch)
+                result = _transform_batch(abs_start, batch)
                 self._transform_time += time.perf_counter() - t0
                 return result
             finally:
@@ -355,8 +488,8 @@ class StreamingDataset(IterableDataset):
         def _try_submit_tx(i: int) -> None:
             """Submit transforms for raw_batches[i] up to available capacity."""
             while raw_batches[i] and tx_semaphore.acquire(blocking=False):
-                batch = raw_batches[i].popleft()
-                tx_pending[i].append(tx_pool.submit(_tx_call_guarded, batch))
+                abs_start, batch = raw_batches[i].popleft()
+                tx_pending[i].append(tx_pool.submit(_tx_call_guarded, abs_start, batch))
 
         def _drain_tx(i: int) -> None:
             """Move completed transform futures into cooked non-blockingly."""
@@ -384,11 +517,14 @@ class StreamingDataset(IterableDataset):
                     # Acquire a transform slot (may block briefly if all
                     # transform_workers are busy with other splits).
                     tx_semaphore.acquire()
-                    batch = raw_batches[i].popleft()
-                    tx_pending[i].append(tx_pool.submit(_tx_call_guarded, batch))
+                    abs_start, batch = raw_batches[i].popleft()
+                    tx_pending[i].append(
+                        tx_pool.submit(_tx_call_guarded, abs_start, batch)
+                    )
                 elif io_pending[i]:
                     # Block on the oldest in-flight I/O fetch.
-                    raw_batches[i].append(io_pending[i].popleft().result())
+                    abs_start, fut = io_pending[i].popleft()
+                    raw_batches[i].append((abs_start, fut.result()))
                     _advance(i)
                 else:
                     break  # split exhausted
@@ -407,15 +543,28 @@ class StreamingDataset(IterableDataset):
                         _fill_io(i)
 
                     while True:
-                        # Stop when any split is exhausted (all exhaust
-                        # simultaneously: equal split sizes + round-robin).
-                        if any(local_consumed[i] >= split_sizes[i] for i in range(n)):
+                        # A cycle only runs if every split can still produce a
+                        # row.  Without skips all splits exhaust simultaneously
+                        # (equal split sizes + round-robin); when
+                        # on_transform_error drops rows a split can run dry
+                        # early, ending the epoch at the last complete cycle.
+                        # This check only sees splits owned by this rank/worker
+                        # (my_splits) — there is no cross-rank coordination, so
+                        # a different rank with fewer skipped rows keeps going;
+                        # see the on_transform_error docstring.
+                        exhausted = False
+                        for i in range(n):
+                            _ensure_cooked(i)
+                            if not cooked[i]:
+                                exhausted = True
+                                break
+                        if exhausted:
                             break
 
                         for i in range(n):
-                            _ensure_cooked(i)
-                            row = cooked[i].popleft()
+                            pos, row = cooked[i].popleft()
                             local_consumed[i] += 1
+                            pos_consumed[i] = pos + 1
                             _advance(i)
 
                             # After the last split in each cycle: update the
@@ -424,21 +573,39 @@ class StreamingDataset(IterableDataset):
                             # even when __iter__ runs in a worker process.
                             if i == n - 1:
                                 self._resume_offset = initial_offset + local_consumed[i]
+                                for j, split_idx in enumerate(my_splits):
+                                    self._resume_positions[split_idx] = pos_consumed[j]
                                 ws = self._worker_stats
                                 ws[0] = sum(
                                     split_sizes[j] - fetch_head[j] for j in range(n)
                                 )
                                 ws[1] = sum(
-                                    batch.num_rows for q in raw_batches for batch in q
+                                    batch.num_rows
+                                    for q in raw_batches
+                                    for _, batch in q
                                 )
                                 ws[2] = sum(len(q) for q in cooked)
                                 ws[3] = sum(local_consumed)
                                 ws[4] = self._bytes_loaded
                                 ws[5] = int(self._fetch_time * 1_000_000)
                                 ws[6] = int(self._transform_time * 1_000_000)
+                                ws[7] = self._rows_skipped
 
                             yield row
                 finally:
+                    # Final stats flush: the per-cycle write above never runs
+                    # when iteration ends mid-cycle (e.g. a split whose rows
+                    # were all skipped before completing a single cycle), so
+                    # counters like rows_skipped would otherwise be stale.
+                    ws = self._worker_stats
+                    ws[0] = sum(split_sizes[j] - fetch_head[j] for j in range(n))
+                    ws[1] = 0  # queue-depth properties document 0 when idle
+                    ws[2] = 0
+                    ws[3] = sum(local_consumed)
+                    ws[4] = self._bytes_loaded
+                    ws[5] = int(self._fetch_time * 1_000_000)
+                    ws[6] = int(self._transform_time * 1_000_000)
+                    ws[7] = self._rows_skipped
                     self._raw_batches_ref = None
                     self._cooked_ref = None
                     self._fetch_head_ref = None
@@ -492,7 +659,7 @@ class StreamingDataset(IterableDataset):
         batches.  Returns 0 when not iterating.
         """
         if self._raw_batches_ref is not None:
-            return sum(batch.num_rows for q in self._raw_batches_ref for batch in q)
+            return sum(batch.num_rows for q in self._raw_batches_ref for _, batch in q)
         return int(self._worker_stats[1])
 
     @property
@@ -522,6 +689,19 @@ class StreamingDataset(IterableDataset):
             )
         return int(self._worker_stats[0])
 
+    @property
+    def rows_skipped(self) -> int:
+        """Number of rows dropped because their transform raised an exception.
+
+        Only ever non-zero when ``on_transform_error`` is set to ``"skip"``,
+        ``"warn"``, or a callable that returned ``True``.  Accumulates across
+        multiple iterations of the same dataset instance and is never reset
+        automatically.
+        """
+        if self._raw_batches_ref is not None:
+            return self._rows_skipped
+        return int(self._worker_stats[7])
+
     @property
     def consumed_rows(self) -> int:
         """Number of rows already yielded to the caller across all splits.
@@ -587,12 +767,27 @@ class StreamingDataset(IterableDataset):
         every split has been consumed the same number of times (by the
         round-robin design), so the per-split count is a single uniform value
         that is identical across all ranks and DataLoader workers.
+
+        ``positions_consumed_per_split`` records how far into each split's
+        permutation iteration has advanced.  It only differs from
+        ``samples_consumed_per_split`` when ``on_transform_error`` skipped
+        rows, in which case entries are exact for the splits this instance
+        iterated and a lower bound (the sample count) for splits owned by
+        other ranks or workers.  Combine the state dicts from all ranks with
+        [merge_state_dicts][lancedb.streaming.StreamingDataset.merge_state_dicts]
+        to recover the exact value for every split before resuming on a
+        different topology.
         """
+        positions = [
+            self._resume_positions.get(split, self._resume_offset)
+            for split in range(self._num_splits)
+        ]
         return {
             "shuffle_seed": self._shuffle_seed,
             "num_splits": self._num_splits,
             "epoch": self._epoch,
             "samples_consumed_per_split": [self._resume_offset] * self._num_splits,
+            "positions_consumed_per_split": positions,
         }
 
     def load_state_dict(self, state: dict) -> None:
@@ -618,3 +813,96 @@ class StreamingDataset(IterableDataset):
             self._resume_offset = consumed[0] if consumed else 0
         else:
             self._resume_offset = int(consumed)
+        # Older checkpoints predate positions_consumed_per_split; without
+        # skipped rows positions equal sample counts, so falling back to
+        # _resume_offset (the .get default in __iter__) is exact.
+        positions = state.get("positions_consumed_per_split")
+        if positions is None:
+            self._resume_positions = {}
+        else:
+            self._resume_positions = {
+                split: int(pos) for split, pos in enumerate(positions)
+            }
+
+    @staticmethod
+    def merge_state_dicts(states: list[dict]) -> dict:
+        """Merge state dicts saved by different ranks into one exact state.
+
+        Only needed when ``on_transform_error`` skips rows in multi-rank
+        training: each rank then knows the exact permutation position only for
+        its own splits, and records a lower bound for the rest.  Because
+        exactly one rank owns each split, the elementwise maximum across all
+        ranks' ``positions_consumed_per_split`` recovers the exact position of
+        every split.  Without skipped rows every rank's state is already
+        identical and merging is a no-op.
+
+        Raises ``ValueError`` if the states are empty or were not produced by
+        the same run (mismatched seed, split count, epoch, or sample counts).
+
+        The merge is always all-to-all and topology-agnostic: collect the
+        ``state_dict()`` from every rank of the *previous* run into one list,
+        merge that whole list, and hand the identical merged result to every
+        rank of the *next* run — regardless of whether the rank count grew,
+        shrank, or stayed the same. There is no pairwise or subset merging
+        step, because each split's exact position is only known to whichever
+        rank owned that split, and the elementwise maximum needs every rank's
+        contribution to be correct.
+
+        For example, checkpointing 8 ranks and resuming on 4 (the same
+        pattern applies when growing, e.g. 4 ranks resuming on 8)::
+
+            states = [ds.state_dict() for ds in previous_run_datasets]  # 8
+            merged = StreamingDataset.merge_state_dicts(states)
+            for ds in resumed_datasets:  # now only 4 ranks
+                ds.load_state_dict(merged)  # same dict on every rank
+
+        The rank count on either side never affects the merge itself, since
+        ``merge_state_dicts`` only cares about the list of states it is
+        given.  Each split's position is recovered by elementwise maximum;
+        here rank 0 owned split 0 (and skipped two rows there) while rank 1
+        owned split 1 (and skipped one row):
+
+        >>> rank0 = {
+        ...     "shuffle_seed": 0, "num_splits": 2, "epoch": 0,
+        ...     "samples_consumed_per_split": [3, 3],
+        ...     "positions_consumed_per_split": [5, 3],
+        ... }
+        >>> rank1 = {
+        ...     "shuffle_seed": 0, "num_splits": 2, "epoch": 0,
+        ...     "samples_consumed_per_split": [3, 3],
+        ...     "positions_consumed_per_split": [3, 4],
+        ... }
+        >>> merged = StreamingDataset.merge_state_dicts([rank0, rank1])
+        >>> merged["positions_consumed_per_split"]
+        [5, 4]
+        """
+        if not states:
+            raise ValueError("merge_state_dicts requires at least one state dict")
+        first = states[0]
+        for state in states[1:]:
+            for key in ("shuffle_seed", "num_splits", "epoch"):
+                if state[key] != first[key]:
+                    raise ValueError(
+                        f"{key} mismatch across state dicts: "
+                        f"{state[key]} != {first[key]}"
+                    )
+            if (
+                state["samples_consumed_per_split"]
+                != first["samples_consumed_per_split"]
+            ):
+                raise ValueError(
+                    "samples_consumed_per_split mismatch across state dicts; "
+                    "state_dict() must be called at the same global step "
+                    "boundary on every rank"
+                )
+        merged = dict(first)
+        all_positions = [
+            state.get(
+                "positions_consumed_per_split", state["samples_consumed_per_split"]
+            )
+            for state in states
+        ]
+        merged["positions_consumed_per_split"] = [
+            max(per_split) for per_split in zip(*all_positions)
+        ]
+        return merged

python/python/lancedb/table.py

@@ -4676,6 +4676,13 @@ class AsyncTable:
         via [`set_unenforced_primary_key`]; bucket sharding additionally
         requires it to be the single column being bucketed.
 
+        By default the MemWAL maintains every index on the table, resolved
+        here — a snapshot, so an index created afterwards needs the spec unset
+        and set again. This fails if one cannot be maintained; name the set
+        with ``with_maintained_indexes`` to install anyway. That pins an exact
+        set (a still-building index is rejected, not omitted); ``[]`` maintains
+        none.
+
         Parameters
         ----------
         spec : LsmWriteSpec
@@ -4702,9 +4709,9 @@ class AsyncTable:
 
         Returns ``None`` when the MemWAL LSM write path is not enabled (no
         spec has been set, or it was removed with `unset_lsm_write_spec`).
-        The returned spec — including its ``maintained_indexes`` and
-        ``writer_config_defaults`` — mirrors what was passed to
-        `set_lsm_write_spec`.
+        The returned spec mirrors what was passed to `set_lsm_write_spec`,
+        except that ``maintained_indexes`` always reports the concrete list
+        resolved when the spec was set — ``None`` never round-trips.
         """
         return await self._inner.get_lsm_write_spec()
 
@@ -6334,7 +6341,9 @@ class TableStatistics:
     Attributes
     ----------
     total_bytes: int
-        The total number of bytes in the table.
+        The total size, in bytes, of the table's data files, index files, and
+        overlay files. Read from the manifest, so this excludes deletion files
+        and manifests.
     num_rows: int
         The total number of rows in the table.
     num_indices: int

python/python/tests/test_elastic_dataloader.py

@@ -1456,6 +1456,408 @@ def test_shuffle_clump_size_yields_all_rows(lance_table):
     )
 
 
+# ---------------------------------------------------------------------------
+# on_transform_error tests
+# ---------------------------------------------------------------------------
+
+
+class BadRowError(ValueError):
+    """Raised by the failing transforms below when a batch contains a bad id."""
+
+
+def _failing_transform(bad_ids: set):
+    """A transform that raises BadRowError whenever the batch has a bad id.
+
+    Raises on the full batch and on any single-row slice containing a bad id,
+    so per-row isolation drops exactly the bad rows.
+    """
+
+    def transform(batch: pa.RecordBatch) -> list:
+        ids = batch.column("id").to_pylist()
+        bad = sorted(set(ids) & bad_ids)
+        if bad:
+            raise BadRowError(f"bad ids in batch: {bad}")
+        return [{"id": i} for i in ids]
+
+    return transform
+
+
+def _sequential_split_members(table) -> list[list[int]]:
+    """Return each split's ids in yield order for shuffle=False.
+
+    With a single rank and no workers the round-robin yields one row per split
+    per cycle, so item k of a clean run belongs to split k % NUM_SPLITS.
+    """
+    ds = StreamingDataset(table, num_splits=NUM_SPLITS, shuffle=False)
+    members: list[list[int]] = [[] for _ in range(NUM_SPLITS)]
+    for k, row in enumerate(ds):
+        members[k % NUM_SPLITS].append(row["id"])
+    return members
+
+
+def test_on_transform_error_default_raises(lance_table):
+    """By default a transform exception propagates and aborts iteration."""
+    ds = StreamingDataset(
+        lance_table,
+        num_splits=NUM_SPLITS,
+        shuffle_seed=SHUFFLE_SEED,
+        transform=_failing_transform({7}),
+    )
+    with pytest.raises(BadRowError):
+        list(ds)
+
+
+def test_on_transform_error_invalid_value(lance_table):
+    with pytest.raises(ValueError, match="on_transform_error"):
+        StreamingDataset(lance_table, num_splits=NUM_SPLITS, on_transform_error="bogus")
+
+
+def test_on_transform_error_skip_drops_bad_rows(lance_table):
+    """With one bad row per split, 'skip' yields every good row exactly once
+    and counts the dropped rows in rows_skipped."""
+    members = _sequential_split_members(lance_table)
+    bad_ids = {members[i][4] for i in range(NUM_SPLITS)}
+
+    ds = StreamingDataset(
+        lance_table,
+        num_splits=NUM_SPLITS,
+        shuffle=False,
+        transform=_failing_transform(bad_ids),
+        on_transform_error="skip",
+    )
+    assert ds.rows_skipped == 0
+
+    ids = [row["id"] for row in ds]
+
+    assert sorted(ids) == sorted(set(range(NUM_ROWS)) - bad_ids)
+    assert ds.rows_skipped == NUM_SPLITS
+
+
+def test_on_transform_error_skip_uneven_ends_at_last_complete_cycle(lance_table):
+    """When one split loses more rows than the others, the epoch ends at the
+    last cycle where every split still has a row — no crash, no bad rows, and
+    every step remains one sample per split."""
+    members = _sequential_split_members(lance_table)
+    bad_ids = set(members[0][:3])  # all 3 bad rows in split 0
+
+    ds = StreamingDataset(
+        lance_table,
+        num_splits=NUM_SPLITS,
+        shuffle=False,
+        transform=_failing_transform(bad_ids),
+        on_transform_error="skip",
+    )
+    items = [row["id"] for row in ds]
+
+    rows_per_split = NUM_ROWS // NUM_SPLITS
+    expected_cycles = rows_per_split - len(bad_ids)
+    assert len(items) == expected_cycles * NUM_SPLITS
+    assert len(set(items)) == len(items), "duplicate samples yielded"
+    assert not set(items) & bad_ids, "a bad row was yielded"
+    # Split 0 contributed exactly its surviving rows, in order, one per cycle.
+    survivors = [i for i in members[0] if i not in bad_ids]
+    assert items[0::NUM_SPLITS] == survivors[:expected_cycles]
+
+
+def test_on_transform_error_warn_logs(lance_table, caplog):
+    """'warn' skips like 'skip' but logs a warning for the failing batch."""
+    members = _sequential_split_members(lance_table)
+    bad_ids = {members[i][3] for i in range(NUM_SPLITS)}
+
+    ds = StreamingDataset(
+        lance_table,
+        num_splits=NUM_SPLITS,
+        shuffle=False,
+        transform=_failing_transform(bad_ids),
+        on_transform_error="warn",
+    )
+    with caplog.at_level(logging.WARNING, logger="lancedb.streaming"):
+        items = list(ds)
+
+    assert len(items) == NUM_ROWS - NUM_SPLITS
+    assert ds.rows_skipped == NUM_SPLITS
+    assert "Skipped" in caplog.text
+    assert "BadRowError" in caplog.text
+
+
+def test_on_transform_error_callable_selective(lance_table):
+    """A callable handler can skip expected errors and re-raise the rest."""
+    members = _sequential_split_members(lance_table)
+    bad_ids = {members[i][0] for i in range(NUM_SPLITS)}
+
+    handled: list[Exception] = []
+
+    def handler(exc: Exception) -> bool:
+        handled.append(exc)
+        return isinstance(exc, BadRowError)
+
+    ds = StreamingDataset(
+        lance_table,
+        num_splits=NUM_SPLITS,
+        shuffle=False,
+        transform=_failing_transform(bad_ids),
+        on_transform_error=handler,
+    )
+    items = list(ds)
+    assert len(items) == NUM_ROWS - NUM_SPLITS
+    assert handled and all(isinstance(exc, BadRowError) for exc in handled)
+
+    def broken_transform(batch: pa.RecordBatch) -> list:
+        raise TypeError("boom")
+
+    ds2 = StreamingDataset(
+        lance_table,
+        num_splits=NUM_SPLITS,
+        shuffle=False,
+        transform=broken_transform,
+        on_transform_error=handler,
+    )
+    with pytest.raises(TypeError, match="boom"):
+        list(ds2)
+
+
+def test_transform_wrong_row_count_raises(lance_table):
+    """A transform that returns the wrong number of rows is an error even with
+    on_transform_error='skip' — silent shrinkage would corrupt accounting."""
+
+    def drops_rows(batch: pa.RecordBatch) -> list:
+        return batch.column("id").to_pylist()[:-1]
+
+    ds = StreamingDataset(
+        lance_table,
+        num_splits=NUM_SPLITS,
+        shuffle_seed=SHUFFLE_SEED,
+        transform=drops_rows,
+        on_transform_error="skip",
+    )
+    with pytest.raises(ValueError, match="one output row per input row"):
+        list(ds)
+
+
+def test_skip_deterministic_across_runs(lance_table):
+    """With a fixed seed, skipping produces the identical sample sequence on
+    every run — skips are data-dependent, not run-dependent."""
+    bad_ids = {5, 17, 46}
+
+    def run() -> tuple[list[int], int]:
+        ds = StreamingDataset(
+            lance_table,
+            num_splits=NUM_SPLITS,
+            shuffle_seed=SHUFFLE_SEED,
+            transform=_failing_transform(bad_ids),
+            on_transform_error="skip",
+        )
+        return [row["id"] for row in ds], ds.rows_skipped
+
+    ids_a, skipped_a = run()
+    ids_b, skipped_b = run()
+    assert ids_a == ids_b
+    assert skipped_a == skipped_b
+    assert not set(ids_a) & bad_ids
+
+
+def test_skip_elastic_det_across_world_sizes(lance_table):
+    """With equal bad-row counts per split, skipping preserves the full
+    elastic-determinism guarantee: identical global batches at every step for
+    every compatible world_size."""
+    members = _sequential_split_members(lance_table)
+    bad_ids = {members[i][6] for i in range(NUM_SPLITS)}
+
+    def collect(world_size: int) -> list[frozenset[int]]:
+        micro = GLOBAL_BATCH_SIZE // world_size
+        iters = [
+            iter(
+                StreamingDataset(
+                    lance_table,
+                    num_splits=NUM_SPLITS,
+                    shuffle=False,
+                    rank=rank,
+                    world_size=world_size,
+                    transform=_failing_transform(bad_ids),
+                    on_transform_error="skip",
+                )
+            )
+            for rank in range(world_size)
+        ]
+        _STOP = object()
+        batches: list[frozenset[int]] = []
+        while True:
+            step_samples: set[int] = set()
+            exhausted = 0
+            for it in iters:
+                for _ in range(micro):
+                    val = next(it, _STOP)
+                    if val is _STOP:
+                        exhausted += 1
+                        break
+                    step_samples.add(val["id"])
+            if exhausted == len(iters):
+                break
+            assert exhausted == 0, (
+                "Rank iterators exhausted at different steps despite equal "
+                "bad-row counts per split"
+            )
+            batches.append(frozenset(step_samples))
+        return batches
+
+    reference = collect(1)
+    assert len(reference) == NUM_ROWS // NUM_SPLITS - 1
+    for ws in (2, 3, 4):
+        assert collect(ws) == reference, f"world_size={ws} diverged"
+
+
+def test_resumability_with_skips_same_topology(lance_table):
+    """Checkpointing mid-epoch with skipped rows resumes exactly: no sample
+    repeated, no sample lost, skipped rows stay skipped."""
+    members = _sequential_split_members(lance_table)
+    # Uneven skips: positions diverge across splits (2 bad in split 0, 1 in
+    # split 5), which only a position-based checkpoint can resume exactly.
+    bad_ids = {members[0][2], members[0][3], members[5][7]}
+    kwargs = dict(
+        num_splits=NUM_SPLITS,
+        shuffle=False,
+        transform=_failing_transform(bad_ids),
+        on_transform_error="skip",
+    )
+
+    reference = [row["id"] for row in StreamingDataset(lance_table, **kwargs)]
+    rows_per_split = NUM_ROWS // NUM_SPLITS
+    assert len(reference) == (rows_per_split - 2) * NUM_SPLITS
+
+    steps = 3
+    ds = StreamingDataset(lance_table, **kwargs)
+    it = iter(ds)
+    consumed = [next(it)["id"] for _ in range(steps * NUM_SPLITS)]
+    checkpoint = ds.state_dict()
+    it.close()
+
+    # Split 0 skipped positions 2 and 3 within its first 3 yields; split 5's
+    # bad row is beyond the checkpoint.  Everything else is at 3 = the sample
+    # count.
+    positions = checkpoint["positions_consumed_per_split"]
+    assert positions[0] == 5
+    assert positions[1:] == [3] * (NUM_SPLITS - 1)
+    assert checkpoint["samples_consumed_per_split"] == [3] * NUM_SPLITS
+
+    ds2 = StreamingDataset(lance_table, **kwargs)
+    ds2.load_state_dict(checkpoint)
+    resumed = [row["id"] for row in ds2]
+
+    assert consumed == reference[: steps * NUM_SPLITS]
+    assert resumed == reference[steps * NUM_SPLITS :]
+
+
+def test_resumability_with_skips_elastic_merge(lance_table):
+    """Elastic resume with skips: each rank's checkpoint knows exact positions
+    only for its own splits; merge_state_dicts recovers the global state, and
+    a run on a different world_size continues exactly."""
+    members = _sequential_split_members(lance_table)
+    # Bad rows early in split 0 (rank 0) and split 6 (rank 1 of a ws=2 run) so
+    # both ranks' position vectors diverge before the checkpoint.
+    bad_ids = {members[0][0], members[0][2], members[6][1]}
+    kwargs = dict(
+        num_splits=NUM_SPLITS,
+        shuffle=False,
+        transform=_failing_transform(bad_ids),
+        on_transform_error="skip",
+    )
+
+    reference = [row["id"] for row in StreamingDataset(lance_table, **kwargs)]
+
+    steps = 3
+    world_size = 2
+    micro = GLOBAL_BATCH_SIZE // world_size
+    datasets = [
+        StreamingDataset(lance_table, rank=rank, world_size=world_size, **kwargs)
+        for rank in range(world_size)
+    ]
+    iters = [iter(ds) for ds in datasets]
+    seen: list[frozenset[int]] = []
+    for _ in range(steps):
+        step_samples = set()
+        for it in iters:
+            for _ in range(micro):
+                step_samples.add(next(it)["id"])
+        seen.append(frozenset(step_samples))
+    states = [ds.state_dict() for ds in datasets]
+    for it in iters:
+        it.close()
+
+    merged = StreamingDataset.merge_state_dicts(states)
+    expected_positions = [3] * NUM_SPLITS
+    expected_positions[0] = 5  # skipped positions 0 and 2
+    expected_positions[6] = 4  # skipped position 1
+    assert merged["positions_consumed_per_split"] == expected_positions
+
+    # The first 3 global batches match the world_size=1 reference.
+    ref_batches = [
+        frozenset(reference[s * NUM_SPLITS : (s + 1) * NUM_SPLITS])
+        for s in range(len(reference) // NUM_SPLITS)
+    ]
+    assert seen == ref_batches[:steps]
+
+    # Resume on world_size=1 from the merged state.
+    ds_resume = StreamingDataset(lance_table, **kwargs)
+    ds_resume.load_state_dict(merged)
+    resumed = [row["id"] for row in ds_resume]
+    assert resumed == reference[steps * NUM_SPLITS :]
+
+
+def test_rows_skipped_flushed_when_split_entirely_bad(lance_table):
+    """A split whose rows all fail never completes a cycle, so the epoch ends
+    immediately — but rows_skipped must still report the drops after the
+    iterator exits (the shared-memory counter is flushed on exhaustion)."""
+    members = _sequential_split_members(lance_table)
+    bad_ids = set(members[0])  # every row of split 0 is bad
+
+    ds = StreamingDataset(
+        lance_table,
+        num_splits=NUM_SPLITS,
+        shuffle=False,
+        transform=_failing_transform(bad_ids),
+        on_transform_error="skip",
+    )
+    assert list(ds) == []
+    assert ds.rows_skipped == len(bad_ids)
+
+
+def test_merge_state_dicts_validates_consistency(lance_table):
+    ds = StreamingDataset(lance_table, num_splits=NUM_SPLITS, shuffle_seed=SHUFFLE_SEED)
+    state = ds.state_dict()
+    other = dict(state, shuffle_seed=SHUFFLE_SEED + 1)
+    with pytest.raises(ValueError, match="shuffle_seed mismatch"):
+        StreamingDataset.merge_state_dicts([state, other])
+    with pytest.raises(ValueError, match="at least one"):
+        StreamingDataset.merge_state_dicts([])
+
+
+def test_load_state_dict_without_positions_key(lance_table):
+    """Checkpoints from before positions_consumed_per_split existed still
+    resume exactly (positions equal sample counts when nothing is skipped)."""
+    reference = [
+        row["id"]
+        for row in StreamingDataset(
+            lance_table, num_splits=NUM_SPLITS, shuffle_seed=SHUFFLE_SEED
+        )
+    ]
+
+    steps = 4
+    ds = StreamingDataset(lance_table, num_splits=NUM_SPLITS, shuffle_seed=SHUFFLE_SEED)
+    it = iter(ds)
+    for _ in range(steps * NUM_SPLITS):
+        next(it)
+    checkpoint = ds.state_dict()
+    it.close()
+    del checkpoint["positions_consumed_per_split"]
+
+    ds2 = StreamingDataset(
+        lance_table, num_splits=NUM_SPLITS, shuffle_seed=SHUFFLE_SEED
+    )
+    ds2.load_state_dict(checkpoint)
+    resumed = [row["id"] for row in ds2]
+    assert resumed == reference[steps * NUM_SPLITS :]
+
+
 def test_num_splits_defaults_to_world_size(lance_table):
     """Omitting num_splits gives world_size splits (one per rank)."""
     ds = StreamingDataset(

python/python/tests/test_embeddings.py

@@ -631,3 +631,23 @@ def test_url_retrieve_downloads_image():
     image_bytes = url_retrieve(image_url)
     img = Image.open(io.BytesIO(image_bytes))
     assert img.size[0] > 0 and img.size[1] > 0
+
+
+def test_jina_generate_image_input_dict_local_path(tmp_path):
+    """
+    JinaEmbeddings._generate_image_input_dict must accept a local image path
+    (str or Path), not just bytes. Previously it crashed with
+    `AttributeError: 'function' object has no attribute 'urlparse'` on any
+    str/Path input because it called `urlparse.urlparse(image)` instead of
+    `urlparse(image)` (urlparse was imported as a function, not a module).
+    """
+    Image = pytest.importorskip("PIL.Image")
+    from lancedb.embeddings.jinaai import JinaEmbeddings
+
+    image_path = tmp_path / "test.png"
+    Image.new("RGB", (4, 4), color="red").save(image_path, format="PNG")
+
+    for image in (str(image_path), image_path):
+        image_dict = JinaEmbeddings._generate_image_input_dict(image)
+        assert "image" in image_dict
+        assert isinstance(image_dict["image"], str) and len(image_dict["image"]) > 0

python/python/tests/test_lsm_write_spec.py

@@ -83,7 +83,9 @@ def test_lsm_write_spec_repr():
     assert s.spec_type == "bucket"
     assert s.column == "id"
     assert s.num_buckets == 4
-    assert s.maintained_indexes == []
+    # A fresh spec defers its maintained set to install time.
+    assert s.maintained_indexes is None
+    assert s.with_maintained_indexes([]).maintained_indexes == []
     assert "bucket" in repr(s)
     assert "id" in repr(s)
     assert "4" in repr(s)
@@ -169,18 +171,23 @@ def test_get_lsm_write_spec(tmp_path):
     table.unset_lsm_write_spec()
     assert table.get_lsm_write_spec() is None
 
-    # Identity round-trips (column recovered from the schema).
+    # Identity round-trips (column recovered from the schema). Leaving the
+    # maintained set to be inferred picks up the index on the table, so the
+    # spec reads back naming it rather than as "infer".
     table.set_lsm_write_spec(LsmWriteSpec.identity("id"))
     spec = table.get_lsm_write_spec()
     assert spec.spec_type == "identity"
     assert spec.column == "id"
+    assert spec.maintained_indexes == [idx_name]
     table.unset_lsm_write_spec()
 
-    # Unsharded round-trips (no routing column).
-    table.set_lsm_write_spec(LsmWriteSpec.unsharded())
+    # Unsharded round-trips (no routing column). Opting out is distinct from
+    # the inferred default.
+    table.set_lsm_write_spec(LsmWriteSpec.unsharded().with_maintained_indexes([]))
     spec = table.get_lsm_write_spec()
     assert spec.spec_type == "unsharded"
     assert spec.column is None
+    assert spec.maintained_indexes == []
 
 
 @pytest.mark.asyncio

python/python/tests/test_merge_insert_lsm.py

@@ -544,7 +544,7 @@ def test_lsm_read_fts_unmaintained_index_errors(tmp_path):
     table.create_index("text", config=FTS())
     # No maintained indexes: the active memtable FTS arm cannot serve un-compacted
     # docs, so the search would silently omit them — reject instead.
-    table.set_lsm_write_spec(LsmWriteSpec.unsharded())
+    table.set_lsm_write_spec(LsmWriteSpec.unsharded().with_maintained_indexes([]))
     with pytest.raises(Exception, match="maintained"):
         table.search("fox", query_type="fts", fts_columns="text").to_arrow()
 
@@ -631,7 +631,7 @@ def test_lsm_read_vector_unmaintained_index_errors(tmp_path):
     )
     # Spec with NO maintained indexes: the base vector index's catch-up is untracked,
     # so the scanner rejects rather than risk dropping compacted-but-unindexed rows.
-    table.set_lsm_write_spec(LsmWriteSpec.unsharded())
+    table.set_lsm_write_spec(LsmWriteSpec.unsharded().with_maintained_indexes([]))
     with pytest.raises(Exception, match="maintained"):
         table.search([1.0] * VECTOR_DIM).to_arrow()
 

python/python/tests/test_table.py

@@ -3713,7 +3713,8 @@ def test_stats(mem_db: DBConnection):
     stats = table.stats()
     print(f"{stats=}")
     assert stats == {
-        "total_bytes": 60,
+        # Full on-disk size of the data file, footer and metadata included.
+        "total_bytes": 633,
         "num_rows": 2,
         "num_indices": 0,
         "fragment_stats": {
@@ -3731,6 +3732,13 @@ def test_stats(mem_db: DBConnection):
         },
     }
 
+    # Index files count toward total_bytes too (only deletion files and
+    # manifests are excluded).
+    table.create_index("id", config=BTree())
+    stats_with_index = table.stats()
+    assert stats_with_index["num_indices"] == 1
+    assert stats_with_index["total_bytes"] > stats["total_bytes"]
+
 
 def test_create_table_empty_list_with_schema(mem_db: DBConnection):
     """Test creating table with empty list data and schema

python/src/index.rs

@@ -289,7 +289,7 @@ struct IvfHnswFlatParams {
     target_partition_size: Option<u32>,
 }
 
-#[pyclass(get_all)]
+#[pyclass(module = "lancedb._lancedb", get_all)]
 /// A description of an index currently configured on a column
 pub struct IndexConfig {
     /// The type of the index

python/src/session.rs

@@ -11,7 +11,7 @@ use pyo3::{PyResult, pyclass, pymethods};
 /// Sessions allow you to configure cache sizes for index and metadata caches,
 /// which can significantly impact memory use and performance. They can
 /// also be re-used across multiple connections to share the same cache state.
-#[pyclass(from_py_object)]
+#[pyclass(module = "lancedb._lancedb", from_py_object)]
 #[derive(Clone)]
 pub struct Session {
     pub(crate) inner: Arc<LanceSession>,

python/src/table.rs

@@ -246,12 +246,22 @@ impl From<lancedb::table::MergeResult> for MergeResult {
     }
 }
 
+/// Render for `__repr__`, so the default reads as Python's `None` rather than
+/// Rust's `Some([..])`.
+fn fmt_maintained(maintained: &Option<Vec<String>>) -> String {
+    match maintained {
+        Some(names) => format!("{:?}", names),
+        None => "None".to_string(),
+    }
+}
+
 /// Specification selecting Lance's MemWAL LSM-style write path for
 /// `merge_insert`.
 ///
 /// Constructed via the `bucket(...)`, `identity(...)`, or `unsharded()`
 /// classmethods, then optionally chain `with_maintained_indexes(...)` and
-/// `with_writer_config_defaults(...)`.
+/// `with_writer_config_defaults(...)`. A fresh spec maintains every index the
+/// MemWAL supports, resolved on install.
 #[pyclass(from_py_object)]
 #[derive(Clone, Debug)]
 pub struct LsmWriteSpec {
@@ -291,11 +301,11 @@ impl LsmWriteSpec {
         }
     }
 
-    /// Replace the list of indexes the MemWAL should keep up to date as
-    /// rows are appended. Each name must reference an index that
-    /// already exists on the table at the time `set_lsm_write_spec`
-    /// is called.
-    pub fn with_maintained_indexes(&self, indexes: Vec<String>) -> Self {
+    /// Set which indexes the MemWAL maintains. `None` (the default)
+    /// resolves every supported index on install; a list is verbatim,
+    /// and an empty list maintains nothing.
+    #[pyo3(signature = (indexes))]
+    pub fn with_maintained_indexes(&self, indexes: Option<Vec<String>>) -> Self {
         Self {
             inner: self.inner.clone().with_maintained_indexes(indexes),
         }
@@ -317,23 +327,29 @@ impl LsmWriteSpec {
                 maintained_indexes,
                 writer_config_defaults,
             } => format!(
-                "LsmWriteSpec.bucket(column={:?}, num_buckets={}, maintained_indexes={:?}, writer_config_defaults={:?})",
-                column, num_buckets, maintained_indexes, writer_config_defaults,
+                "LsmWriteSpec.bucket(column={:?}, num_buckets={}, maintained_indexes={}, writer_config_defaults={:?})",
+                column,
+                num_buckets,
+                fmt_maintained(maintained_indexes),
+                writer_config_defaults,
             ),
             lancedb::table::LsmWriteSpec::Identity {
                 column,
                 maintained_indexes,
                 writer_config_defaults,
             } => format!(
-                "LsmWriteSpec.identity(column={:?}, maintained_indexes={:?}, writer_config_defaults={:?})",
-                column, maintained_indexes, writer_config_defaults,
+                "LsmWriteSpec.identity(column={:?}, maintained_indexes={}, writer_config_defaults={:?})",
+                column,
+                fmt_maintained(maintained_indexes),
+                writer_config_defaults,
             ),
             lancedb::table::LsmWriteSpec::Unsharded {
                 maintained_indexes,
                 writer_config_defaults,
             } => format!(
-                "LsmWriteSpec.unsharded(maintained_indexes={:?}, writer_config_defaults={:?})",
-                maintained_indexes, writer_config_defaults,
+                "LsmWriteSpec.unsharded(maintained_indexes={}, writer_config_defaults={:?})",
+                fmt_maintained(maintained_indexes),
+                writer_config_defaults,
             ),
         }
     }
@@ -368,10 +384,10 @@ impl LsmWriteSpec {
         }
     }
 
-    /// Names of indexes the MemWAL should keep up to date during writes.
+    /// Indexes the MemWAL keeps up to date, or `None` for every supported one.
     #[getter]
-    pub fn maintained_indexes(&self) -> Vec<String> {
-        self.inner.maintained_indexes().to_vec()
+    pub fn maintained_indexes(&self) -> Option<Vec<String>> {
+        self.inner.maintained_indexes().map(<[String]>::to_vec)
     }
 
     /// Default `ShardWriter` configuration recorded by this spec.
@@ -563,7 +579,7 @@ impl PyBlobFile {
     }
 }
 
-#[pyclass(get_all, from_py_object)]
+#[pyclass(module = "lancedb._lancedb", get_all, from_py_object)]
 #[derive(Clone, Debug)]
 pub struct FtsToken {
     pub text: String,

rust/lancedb/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "lancedb"
-version = "0.37.1-beta.0"
+version = "0.37.1-beta.1"
 edition.workspace = true
 description = "LanceDB: A serverless, low-latency vector database for AI applications"
 license.workspace = true
@@ -49,8 +49,6 @@ lance-namespace = { workspace = true }
 lance-namespace-impls = { workspace = true }
 metrics = { workspace = true, optional = true }
 metrics-util = { workspace = true, optional = true }
-# Pin the GooseFS SDK to the version required by Lance's OpenDAL dependency.
-goosefs-sdk = { version = "=0.1.9", optional = true }
 moka = { workspace = true }
 pin-project = { workspace = true }
 tokio = { version = "1.23", features = ["rt-multi-thread", "sync"] }
@@ -100,6 +98,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"] }
 uuid = { version = "1.7.0", features = ["v4"] }
 walkdir = "2"
@@ -135,7 +134,6 @@ azure = [
 ]
 cos = ["lance/tencent", "lance-io/tencent"]
 goosefs = [
-    "dep:goosefs-sdk",
     "lance/goosefs",
     "lance-io/goosefs",
     "lance-namespace-impls/dir-goosefs",

rust/lancedb/src/io/object_store.rs

@@ -8,7 +8,7 @@ use std::{fmt::Formatter, sync::Arc};
 use futures::{StreamExt, TryFutureExt, stream::BoxStream};
 use lance::io::WrappingObjectStore;
 use object_store::{
-    CopyOptions, Error, GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta,
+    CopyMode, CopyOptions, Error, GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta,
     ObjectStore, ObjectStoreExt, PutMultipartOptions, PutOptions, PutPayload, PutResult, Result,
     UploadPart, path::Path,
 };
@@ -52,7 +52,7 @@ impl PrimaryOnly for Path {
 /// store. We have primary store that is durable but slow, and a secondary
 /// store that is fast but not asdurable
 ///
-/// Note: this object store does not mirror writes to *.manifest files
+/// Note: this object store does not mirror writes to `_latest.manifest`.
 #[async_trait]
 impl ObjectStore for MirroringObjectStore {
     async fn put_opts(
@@ -137,6 +137,16 @@ impl ObjectStore for MirroringObjectStore {
             // or may be evicted before the copy begins.
             match self.secondary.copy_opts(from, to, options.clone()).await {
                 Ok(()) | Err(Error::NotFound { .. }) => {}
+                // The secondary is a non-authoritative cache. A process can
+                // leave an orphaned manifest there if it exits before creating
+                // the durable primary object, so let the primary decide the
+                // outcome of create-only manifest copies.
+                Err(Error::AlreadyExists { .. } | Error::Precondition { .. })
+                    if options.mode == CopyMode::Create
+                        && to
+                            .filename()
+                            .map(|name| name.ends_with(".manifest"))
+                            .unwrap_or(false) => {}
                 Err(err) => return Err(err),
             }
             self.primary.copy_opts(from, to, options).await
@@ -340,6 +350,64 @@ mod test {
         ));
     }
 
+    #[tokio::test]
+    async fn test_create_manifest_recovers_from_orphaned_secondary() {
+        let primary: Arc<dyn ObjectStore> = Arc::new(InMemory::new());
+        let secondary: Arc<dyn ObjectStore> = Arc::new(InMemory::new());
+        let store = MirroringObjectStore {
+            primary: primary.clone(),
+            secondary: secondary.clone(),
+        };
+        let staging = Path::from("_versions/1.manifest-staging");
+        let finalized = Path::from("_versions/1.manifest");
+
+        primary
+            .put(&staging, "manifest contents".into())
+            .await
+            .unwrap();
+        secondary
+            .put(&staging, "manifest contents".into())
+            .await
+            .unwrap();
+        secondary
+            .copy_if_not_exists(&staging, &finalized)
+            .await
+            .expect("simulate a crash after secondary create and before primary create");
+
+        store
+            .copy_if_not_exists(&staging, &finalized)
+            .await
+            .expect("an orphaned secondary manifest must not block primary creation");
+
+        let copied = primary
+            .get(&finalized)
+            .await
+            .unwrap()
+            .bytes()
+            .await
+            .unwrap();
+        assert_eq!(copied, "manifest contents");
+
+        assert!(matches!(
+            store.copy_if_not_exists(&staging, &finalized).await,
+            Err(Error::AlreadyExists { .. } | Error::Precondition { .. })
+        ));
+
+        let non_manifest = Path::from("data/existing.lance");
+        secondary
+            .copy_if_not_exists(&staging, &non_manifest)
+            .await
+            .unwrap();
+        assert!(matches!(
+            store.copy_if_not_exists(&staging, &non_manifest).await,
+            Err(Error::AlreadyExists { .. } | Error::Precondition { .. })
+        ));
+        assert!(matches!(
+            primary.head(&non_manifest).await,
+            Err(Error::NotFound { .. })
+        ));
+    }
+
     // This test is ignored because lance 3.0 introduced LocalWriter optimization
     // that bypasses the object store wrapper for local writes. The mirroring feature
     // still works for remote/cloud storage, but can't be tested with local storage.

rust/lancedb/src/remote/table.rs

@@ -2520,9 +2520,9 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
         self.check_mutable().await?;
 
         // Map the spec onto the server's request DTO. `sharding` is internally
-        // tagged on `mode` to mirror sophon's `Sharding` enum; `maintained_indexes`
-        // and `writer_config_defaults` are sent verbatim (an empty list means "no
-        // maintained indexes", not "default to all").
+        // tagged on `mode` to mirror sophon's `Sharding` enum. A null
+        // `maintained_indexes` asks the server to resolve every maintainable
+        // index at HEAD; a list is verbatim, an empty one meaning none.
         let sharding = match &spec {
             LsmWriteSpec::Bucket {
                 column,
@@ -6599,7 +6599,7 @@ mod tests {
                 .unwrap()
         });
         let spec = crate::table::LsmWriteSpec::unsharded()
-            .with_maintained_indexes(["id_idx"])
+            .with_maintained_indexes(vec!["id_idx".to_string()])
             .with_writer_config_defaults([("max_memtable_rows", "1000")]);
         table.set_lsm_write_spec(spec).await.unwrap();
     }
@@ -6618,7 +6618,8 @@ mod tests {
                 body["sharding"],
                 serde_json::json!({ "mode": "bucket", "column": "id", "num_buckets": 16 })
             );
-            assert_eq!(body["maintained_indexes"], serde_json::json!([]));
+            // An unpinned maintained set sends null: resolve server-side.
+            assert_eq!(body["maintained_indexes"], serde_json::Value::Null);
             http::Response::builder().status(200).body("{}").unwrap()
         });
         table
@@ -6627,6 +6628,23 @@ mod tests {
             .unwrap();
     }
 
+    /// `[]` (none) must stay distinguishable on the wire from null (all).
+    #[tokio::test]
+    async fn test_set_lsm_write_spec_no_maintained_indexes() {
+        let table = Table::new_with_handler("my_table", |request| {
+            let body = request.body().unwrap().as_bytes().unwrap();
+            let body: serde_json::Value = serde_json::from_slice(body).unwrap();
+            assert_eq!(body["maintained_indexes"], serde_json::json!([]));
+            http::Response::builder().status(200).body("{}").unwrap()
+        });
+        table
+            .set_lsm_write_spec(
+                crate::table::LsmWriteSpec::bucket("id", 16).with_maintained_indexes(Vec::new()),
+            )
+            .await
+            .unwrap();
+    }
+
     #[tokio::test]
     async fn test_set_lsm_write_spec_identity() {
         let table = Table::new_with_handler("my_table", |request| {
@@ -6701,7 +6719,7 @@ mod tests {
             } => {
                 assert_eq!(column, "id");
                 assert_eq!(num_buckets, 4);
-                assert_eq!(maintained_indexes, vec!["id_idx".to_string()]);
+                assert_eq!(maintained_indexes, Some(vec!["id_idx".to_string()]));
                 assert_eq!(
                     writer_config_defaults
                         .get("durable_write")

rust/lancedb/src/table.rs

@@ -95,7 +95,6 @@ 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};
@@ -371,6 +370,8 @@ pub use self::merge::MergeResult;
 /// date) and [`LsmWriteSpec::with_writer_config_defaults`] (default
 /// `ShardWriter` configuration recorded in the MemWAL index).
 ///
+/// A fresh spec maintains every index on the table, resolved on install.
+///
 /// Install a spec with [`Table::set_lsm_write_spec`] and remove it with
 /// [`Table::unset_lsm_write_spec`]. The actual `merge_insert` dispatch
 /// onto the MemWAL writer is a follow-up.
@@ -385,9 +386,12 @@ pub enum LsmWriteSpec {
     Bucket {
         column: String,
         num_buckets: u32,
-        /// Names of indexes (already created on the table) that the
-        /// MemWAL should maintain in-memory as rows are appended.
-        maintained_indexes: Vec<String>,
+        /// Indexes the MemWAL maintains in-memory as rows are appended.
+        ///
+        /// `None` means every index it can maintain, resolved on install — a
+        /// snapshot, so indexes created later need the spec unset and re-set.
+        /// `Some([])` maintains nothing.
+        maintained_indexes: Option<Vec<String>>,
         /// Default `ShardWriter` configuration recorded in the MemWAL index.
         writer_config_defaults: HashMap<String, String>,
     },
@@ -397,35 +401,41 @@ pub enum LsmWriteSpec {
     /// distinct value of `column` becomes its own shard.
     Identity {
         column: String,
-        /// Names of indexes (already created on the table) that the
-        /// MemWAL should maintain in-memory as rows are appended.
-        maintained_indexes: Vec<String>,
+        /// Indexes the MemWAL maintains in-memory as rows are appended.
+        ///
+        /// `None` means every index it can maintain, resolved on install — a
+        /// snapshot, so indexes created later need the spec unset and re-set.
+        /// `Some([])` maintains nothing.
+        maintained_indexes: Option<Vec<String>>,
         /// Default `ShardWriter` configuration recorded in the MemWAL index.
         writer_config_defaults: HashMap<String, String>,
     },
     /// No sharding — every `merge_insert` call writes to a single MemWAL shard.
     Unsharded {
-        /// Names of indexes (already created on the table) that the
-        /// MemWAL should maintain in-memory as rows are appended.
-        maintained_indexes: Vec<String>,
+        /// Indexes the MemWAL maintains in-memory as rows are appended.
+        ///
+        /// `None` means every index it can maintain, resolved on install — a
+        /// snapshot, so indexes created later need the spec unset and re-set.
+        /// `Some([])` maintains nothing.
+        maintained_indexes: Option<Vec<String>>,
         /// Default `ShardWriter` configuration recorded in the MemWAL index.
         writer_config_defaults: HashMap<String, String>,
     },
 }
 
 impl LsmWriteSpec {
-    /// Construct a hash-bucket sharding spec with no maintained indexes.
+    /// Construct a hash-bucket sharding spec maintaining every index on the table.
     pub fn bucket(column: impl Into<String>, num_buckets: u32) -> Self {
         Self::Bucket {
             column: column.into(),
             num_buckets,
-            maintained_indexes: Vec::new(),
+            maintained_indexes: None,
             writer_config_defaults: HashMap::new(),
         }
     }
 
     /// Construct an identity-sharding spec (shard by the raw value of
-    /// `column`) with no maintained indexes.
+    /// `column`) maintaining every index on the table.
     ///
     /// `column` must be a deterministic function of the unenforced primary
     /// key: every row with a given primary key must always produce the same
@@ -437,28 +447,37 @@ impl LsmWriteSpec {
     pub fn identity(column: impl Into<String>) -> Self {
         Self::Identity {
             column: column.into(),
-            maintained_indexes: Vec::new(),
+            maintained_indexes: None,
             writer_config_defaults: HashMap::new(),
         }
     }
 
-    /// Construct an unsharded spec with no maintained indexes.
+    /// Construct an unsharded spec maintaining every index on the table.
     pub fn unsharded() -> Self {
         Self::Unsharded {
-            maintained_indexes: Vec::new(),
+            maintained_indexes: None,
             writer_config_defaults: HashMap::new(),
         }
     }
 
-    /// Replace the list of indexes the MemWAL should keep up to date as
-    /// rows are appended. Each name must reference an index that already
-    /// exists on the table at the time `set_lsm_write_spec` is called.
-    pub fn with_maintained_indexes<I, S>(mut self, indexes: I) -> Self
-    where
-        I: IntoIterator<Item = S>,
-        S: Into<String>,
-    {
-        let v: Vec<String> = indexes.into_iter().map(Into::into).collect();
+    /// Set which indexes the MemWAL maintains.
+    ///
+    /// `None` (the default) resolves to every index on the table at install,
+    /// failing if one cannot be maintained — name the set to install anyway. A
+    /// list is verbatim: each name must already exist and be maintainable, and
+    /// an empty list maintains nothing.
+    ///
+    /// ```
+    /// # use lancedb::table::LsmWriteSpec;
+    /// // Every index the table has when the spec is installed:
+    /// LsmWriteSpec::unsharded().with_maintained_indexes(None);
+    /// // Exactly these:
+    /// LsmWriteSpec::unsharded().with_maintained_indexes(vec!["id_idx".to_string()]);
+    /// // None at all:
+    /// LsmWriteSpec::unsharded().with_maintained_indexes(Vec::new());
+    /// ```
+    pub fn with_maintained_indexes(mut self, indexes: impl Into<Option<Vec<String>>>) -> Self {
+        let indexes = indexes.into();
         match &mut self {
             Self::Bucket {
                 maintained_indexes, ..
@@ -468,7 +487,7 @@ impl LsmWriteSpec {
             }
             | Self::Unsharded {
                 maintained_indexes, ..
-            } => *maintained_indexes = v,
+            } => *maintained_indexes = indexes,
         }
         self
     }
@@ -504,8 +523,9 @@ impl LsmWriteSpec {
         self
     }
 
-    /// Borrow the list of index names this spec asks MemWAL to maintain.
-    pub fn maintained_indexes(&self) -> &[String] {
+    /// Borrow the list of index names this spec asks MemWAL to maintain, or
+    /// `None` when it asks for every index on the table.
+    pub fn maintained_indexes(&self) -> Option<&[String]> {
         match self {
             Self::Bucket {
                 maintained_indexes, ..
@@ -515,7 +535,7 @@ impl LsmWriteSpec {
             }
             | Self::Unsharded {
                 maintained_indexes, ..
-            } => maintained_indexes,
+            } => maintained_indexes.as_deref(),
         }
     }
 
@@ -1713,7 +1733,7 @@ impl Table {
     /// # async fn example(table: &Table) -> Result<(), Box<dyn std::error::Error>> {
     /// table
     ///     .set_lsm_write_spec(
-    ///         LsmWriteSpec::bucket("id", 16).with_maintained_indexes(["id_idx"]),
+    ///         LsmWriteSpec::bucket("id", 16).with_maintained_indexes(vec!["id_idx".to_string()]),
     ///     )
     ///     .await?;
     /// # Ok(())
@@ -1735,9 +1755,10 @@ impl Table {
     ///
     /// Returns `Ok(None)` when the MemWAL LSM write path is not enabled (no
     /// spec has been set, or it was removed with [`Table::unset_lsm_write_spec`]).
-    /// The returned spec — including its [`LsmWriteSpec::maintained_indexes`] and
-    /// [`LsmWriteSpec::writer_config_defaults`] — mirrors what was passed to
-    /// [`Table::set_lsm_write_spec`].
+    /// The returned spec mirrors what was passed to
+    /// [`Table::set_lsm_write_spec`], except that
+    /// [`LsmWriteSpec::maintained_indexes`] always reports the concrete list
+    /// resolved when the spec was set — `None` never round-trips.
     ///
     /// # Example
     ///
@@ -3548,9 +3569,24 @@ impl BaseTable for NativeTable {
         let num_rows = self.count_rows(None).await?;
         let num_indices = self.list_indices().await?.len();
         let ds = self.dataset.get().await?;
-        let ds_clone = (*ds).clone();
-        let ds_stats = Arc::new(ds_clone).calculate_data_stats().await?;
-        let total_bytes = ds_stats.fields.iter().map(|f| f.bytes_on_disk).sum::<u64>() as usize;
+        // Sizes come from the manifest. Summing per-field `bytes_on_disk` instead
+        // would open every data file to read its column metadata, which costs one
+        // IO per fragment and reports 0 for legacy v1 storage.
+        //
+        // The manifest summary covers only the fragments' base data files, so
+        // overlay files (recorded on each fragment) and index files (recorded in
+        // the manifest's index section) are added separately.
+        let mut total_bytes = ds.manifest().summary().total_files_size as usize;
+        for frag in ds.manifest().fragments.iter() {
+            for overlay in &frag.overlays {
+                if let Some(size) = overlay.data_file.file_size_bytes.get() {
+                    total_bytes += size.get() as usize;
+                }
+            }
+        }
+        for index in ds.load_indices().await?.iter() {
+            total_bytes += index.total_size_bytes().unwrap_or(0) as usize;
+        }
 
         let frags = ds.get_fragments();
         let mut sorted_sizes = join_all(
@@ -3622,7 +3658,12 @@ impl BaseTable for NativeTable {
 #[skip_serializing_none]
 #[derive(Debug, Deserialize, PartialEq)]
 pub struct TableStatistics {
-    /// The total number of bytes in the table
+    /// The total size, in bytes, of the table's data files, index files, and
+    /// overlay files
+    ///
+    /// Read from the manifest, so this excludes deletion files and manifests,
+    /// and it excludes any file whose size the manifest does not record
+    /// (tables and indices written before writers persisted file sizes).
     pub total_bytes: usize,
 
     /// The number of rows in the table
@@ -3683,6 +3724,7 @@ mod tests {
     use super::*;
     use crate::connect;
     use crate::connection::ConnectBuilder;
+    use crate::io::object_store::io_tracking::IoTrackingStore;
     use crate::query::Select;
     use crate::query::{ExecutableQuery, QueryBase};
     use crate::test_utils::connection::new_test_connection;
@@ -5065,7 +5107,7 @@ mod tests {
         // Bucket spec round-trips exactly, including the routing column (recovered
         // from its field id), maintained indexes, and writer config defaults.
         let spec = LsmWriteSpec::bucket("id", 4)
-            .with_maintained_indexes([idx_name])
+            .with_maintained_indexes(vec![idx_name.clone()])
             .with_writer_config_defaults([("durable_write", "false")]);
         table.set_lsm_write_spec(spec.clone()).await.unwrap();
         assert_eq!(table.get_lsm_write_spec().await.unwrap(), Some(spec));
@@ -5075,15 +5117,125 @@ mod tests {
         assert_eq!(table.get_lsm_write_spec().await.unwrap(), None);
 
         // Identity sharding round-trips (column recovered from the schema).
+        // A spec left at its default maintains every index on the table, so it
+        // reads back naming the one on the table rather than as "infer".
         let spec = LsmWriteSpec::identity("region");
         table.set_lsm_write_spec(spec.clone()).await.unwrap();
-        assert_eq!(table.get_lsm_write_spec().await.unwrap(), Some(spec));
+        assert_eq!(
+            table.get_lsm_write_spec().await.unwrap(),
+            Some(spec.with_maintained_indexes(vec![idx_name.clone()]))
+        );
         table.unset_lsm_write_spec().await.unwrap();
 
         // Unsharded round-trips (no routing column).
         let spec = LsmWriteSpec::unsharded();
         table.set_lsm_write_spec(spec.clone()).await.unwrap();
-        assert_eq!(table.get_lsm_write_spec().await.unwrap(), Some(spec));
+        assert_eq!(
+            table.get_lsm_write_spec().await.unwrap(),
+            Some(spec.with_maintained_indexes(vec![idx_name]))
+        );
+    }
+
+    /// The maintained set defaults to every index on the table, resolved at
+    /// install. An index the memtable cannot build fails the install rather
+    /// than being dropped: maintaining it would take the table offline for
+    /// writes, dropping it would hide that from the caller.
+    #[tokio::test]
+    async fn test_set_lsm_write_spec_infers_maintained_indexes() {
+        let tmp_dir = tempdir().unwrap();
+        let uri = tmp_dir.path().to_str().unwrap();
+
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("id", DataType::Int64, false),
+            Field::new("tag", DataType::Utf8, true),
+        ]));
+        let batch = RecordBatch::try_new(
+            schema.clone(),
+            vec![
+                Arc::new(arrow_array::Int64Array::from(vec![1, 2, 3])),
+                Arc::new(StringArray::from(vec!["a", "b", "c"])),
+            ],
+        )
+        .unwrap();
+        let reader: Box<dyn arrow_array::RecordBatchReader + Send> =
+            Box::new(RecordBatchIterator::new(vec![Ok(batch)], schema.clone()));
+        let conn = ConnectBuilder::new(uri)
+            .read_consistency_interval(Duration::from_secs(0))
+            .execute()
+            .await
+            .unwrap();
+        let table = conn.create_table("t", reader).execute().await.unwrap();
+
+        table
+            .create_index(&["id"], Index::BTree(Default::default()))
+            .name("id_btree".to_string())
+            .execute()
+            .await
+            .unwrap();
+        table
+            .create_index(&["tag"], Index::Bitmap(Default::default()))
+            .name("tag_bitmap".to_string())
+            .execute()
+            .await
+            .unwrap();
+
+        // Explicitly naming the bitmap index fails before anything commits.
+        let err = table
+            .set_lsm_write_spec(
+                LsmWriteSpec::unsharded().with_maintained_indexes(vec!["tag_bitmap".to_string()]),
+            )
+            .await
+            .unwrap_err();
+        assert!(
+            matches!(err, Error::InvalidInput { ref message } if message.contains("tag_bitmap")),
+            "expected the bitmap index to be rejected, got {err:?}"
+        );
+        assert_eq!(table.get_lsm_write_spec().await.unwrap(), None);
+
+        // The default covers every index, so the bitmap fails it too.
+        let err = table
+            .set_lsm_write_spec(LsmWriteSpec::unsharded())
+            .await
+            .unwrap_err();
+        assert!(
+            matches!(err, Error::InvalidInput { ref message }
+                if message.contains("tag_bitmap") && message.contains("maintained_indexes")),
+            "expected the inferred set to be rejected, got {err:?}"
+        );
+        assert_eq!(table.get_lsm_write_spec().await.unwrap(), None);
+
+        // Naming the maintainable subset installs.
+        table
+            .set_lsm_write_spec(
+                LsmWriteSpec::unsharded().with_maintained_indexes(vec!["id_btree".to_string()]),
+            )
+            .await
+            .unwrap();
+        assert_eq!(
+            table
+                .get_lsm_write_spec()
+                .await
+                .unwrap()
+                .unwrap()
+                .maintained_indexes(),
+            Some(["id_btree".to_string()].as_slice())
+        );
+
+        // Opting out entirely is distinct from the default.
+        table.unset_lsm_write_spec().await.unwrap();
+        table
+            .set_lsm_write_spec(LsmWriteSpec::unsharded().with_maintained_indexes(Vec::new()))
+            .await
+            .unwrap();
+        assert_eq!(
+            table
+                .get_lsm_write_spec()
+                .await
+                .unwrap()
+                .unwrap()
+                .maintained_indexes(),
+            Some([].as_slice())
+        );
     }
 
     #[tokio::test]
@@ -5131,12 +5283,16 @@ mod tests {
 
         let res = table.stats().await.unwrap();
         println!("{:#?}", res);
+        // `total_bytes` is the full on-disk size of the 11 data files (this table
+        // has no index or overlay files), so it is well above the 2000 bytes of
+        // column data these 250 int32 pairs hold: each file carries its own footer
+        // and metadata.
         assert_eq!(
             res,
             TableStatistics {
                 num_rows: 250,
                 num_indices: 0,
-                total_bytes: 2300,
+                total_bytes: 8925,
                 fragment_stats: FragmentStatistics {
                     num_fragments: 11,
                     num_small_fragments: 11,
@@ -5176,4 +5332,196 @@ mod tests {
             }
         )
     }
+
+    /// `total_bytes` counts more than the base data files: index files and
+    /// overlay files recorded in the manifest are included too.
+    #[tokio::test]
+    pub async fn test_stats_includes_index_and_overlay_files() {
+        use lance::dataset::WriteDestination;
+        use lance::dataset::transaction::{DataOverlayGroup, Operation};
+        use lance_file::version::{ConcreteFileVersion, LanceFileVersion};
+        use lance_file::writer::FileWriterOptions;
+        use lance_io::utils::CachedFileSize;
+        use lance_table::format::DataFile;
+        use lance_table::format::overlay::{DataOverlayFile, OverlayCoverage};
+        use roaring::RoaringBitmap;
+
+        let tmp_dir = tempdir().unwrap();
+        let uri = tmp_dir.path().to_str().unwrap();
+        let conn = ConnectBuilder::new(uri)
+            .read_consistency_interval(Duration::from_secs(0))
+            .execute()
+            .await
+            .unwrap();
+
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("id", DataType::Int32, false),
+            Field::new("foo", DataType::Int32, true),
+        ]));
+        let batch = RecordBatch::try_new(
+            schema.clone(),
+            vec![
+                Arc::new(Int32Array::from_iter_values(0..100)),
+                Arc::new(Int32Array::from_iter_values(0..100)),
+            ],
+        )
+        .unwrap();
+        let table = conn
+            .create_table("test_stats_extra_files", batch)
+            .execute()
+            .await
+            .unwrap();
+
+        let data_only = table.stats().await.unwrap().total_bytes;
+        assert!(data_only > 0);
+
+        // A scalar index adds index files whose sizes are recorded in the
+        // manifest's index section.
+        table
+            .create_index(&["id"], Index::Auto)
+            .execute()
+            .await
+            .unwrap();
+        let with_index = table.stats().await.unwrap().total_bytes;
+        let dataset = {
+            let native = table.as_native().unwrap();
+            (*native.dataset.get().await.unwrap()).clone()
+        };
+        let index_bytes: usize = dataset
+            .load_indices()
+            .await
+            .unwrap()
+            .iter()
+            .map(|idx| idx.total_size_bytes().unwrap_or(0) as usize)
+            .sum();
+        assert!(index_bytes > 0);
+        assert_eq!(with_index, data_only + index_bytes);
+
+        // Commit an overlay file supplying new `foo` values for the first three
+        // rows of fragment 0. There is no high-level API that writes overlays
+        // yet, so write the overlay's data file and commit the `DataOverlay`
+        // operation by hand.
+        let read_version = dataset.version().version;
+        let fragment_id = dataset.get_fragments()[0].id() as u64;
+        let foo_field_id = dataset.schema().field("foo").unwrap().id;
+        let overlay_schema = dataset.schema().project_by_ids(&[foo_field_id], true);
+        let file_version = ConcreteFileVersion::from(LanceFileVersion::Stable);
+
+        let filename = "overlay.lance".to_string();
+        let store = dataset.object_store(None).await.unwrap();
+        let path = dataset.data_dir().child(filename.clone());
+        let obj_writer = store.create(&path).await.unwrap();
+        let mut writer = lance_file::versions::create_writer(
+            file_version,
+            obj_writer,
+            overlay_schema,
+            FileWriterOptions::default(),
+        )
+        .unwrap();
+        writer
+            .write_column(0, Arc::new(Int32Array::from(vec![1000, 1001, 1002])) as _)
+            .await
+            .unwrap();
+        let summary = writer.finish().await.unwrap();
+        let overlay_bytes = summary.size_bytes as usize;
+        assert!(overlay_bytes > 0);
+
+        let mut data_file = DataFile::new_unstarted(filename, file_version);
+        data_file.fields = writer
+            .field_id_to_column_indices()
+            .iter()
+            .map(|(field_id, _)| *field_id as i32)
+            .collect::<Vec<_>>()
+            .into();
+        data_file.column_indices = writer
+            .field_id_to_column_indices()
+            .iter()
+            .map(|(_, column_index)| *column_index as i32)
+            .collect::<Vec<_>>()
+            .into();
+        data_file.file_size_bytes = CachedFileSize::new(summary.size_bytes);
+
+        let overlay = DataOverlayFile {
+            data_file,
+            coverage: OverlayCoverage::dense(RoaringBitmap::from_iter(0..3)),
+            committed_version: 0,
+        };
+        Dataset::commit(
+            WriteDestination::Dataset(Arc::new(dataset)),
+            Operation::DataOverlay {
+                groups: vec![DataOverlayGroup {
+                    fragment_id,
+                    overlays: vec![overlay],
+                }],
+            },
+            Some(read_version),
+            None,
+            None,
+            Arc::new(Default::default()),
+            false,
+        )
+        .await
+        .unwrap();
+
+        table.checkout_latest().await.unwrap();
+        let with_overlay = table.stats().await.unwrap().total_bytes;
+        assert_eq!(with_overlay, with_index + overlay_bytes);
+    }
+
+    /// `stats()` must stay manifest-only. Summing per-field `bytes_on_disk`
+    /// instead opens every data file, so cost would grow with fragment count.
+    #[tokio::test]
+    pub async fn test_stats_does_not_read_data_files() {
+        let tmp_dir = tempdir().unwrap();
+        let uri = tmp_dir.path().to_str().unwrap();
+
+        let conn = ConnectBuilder::new(uri).execute().await.unwrap();
+
+        let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
+        let batch = RecordBatch::try_new(
+            schema.clone(),
+            vec![Arc::new(Int32Array::from_iter_values(0..10))],
+        )
+        .unwrap();
+
+        conn.create_table("test_stats_io", batch.clone())
+            .execute()
+            .await
+            .unwrap();
+        let table = conn.open_table("test_stats_io").execute().await.unwrap();
+        const NUM_APPENDS: usize = 20;
+        for _ in 0..NUM_APPENDS {
+            table.add(batch.clone()).execute().await.unwrap();
+        }
+
+        // Reopen through a tracking store so the counters cover `stats()` alone and
+        // not the writes above.
+        let (wrapper, io_stats) = IoTrackingStore::new_wrapper();
+        let table = conn
+            .open_table("test_stats_io")
+            .lance_read_params(ReadParams {
+                store_options: Some(ObjectStoreParams {
+                    object_store_wrapper: Some(wrapper),
+                    ..Default::default()
+                }),
+                ..Default::default()
+            })
+            .execute()
+            .await
+            .unwrap();
+        io_stats.lock().unwrap().read_iops = 0;
+
+        let stats = table.stats().await.unwrap();
+        let read_iops = io_stats.lock().unwrap().read_iops;
+
+        assert_eq!(stats.fragment_stats.num_fragments, NUM_APPENDS + 1);
+        assert!(stats.total_bytes > 0);
+        // Reading the fragments' data files would take at least one IOP each.
+        assert!(
+            read_iops < stats.fragment_stats.num_fragments as u64,
+            "stats() issued {} read IOPs across {} fragments",
+            read_iops,
+            stats.fragment_stats.num_fragments
+        );
+    }
 }

rust/lancedb/src/table/merge.rs

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

rust/lancedb/src/table/merge/lsm.rs

@@ -29,6 +29,7 @@ use arrow_schema::{DataType, Schema as ArrowSchema, SchemaRef};
 use lance::Dataset;
 use lance::dataset::mem_wal::{
     DatasetMemWalExt, ShardWriter, ShardWriterConfig, evaluate_sharding_spec,
+    validate_maintained_indexes,
 };
 use lance::index::DatasetIndexExt;
 use lance_core::datatypes::Schema as LanceSchema;
@@ -37,8 +38,9 @@ use tokio::sync::RwLock;
 use uuid::Uuid;
 
 use crate::error::{Error, Result};
+use crate::index::IndexConfig;
 use crate::table::merge::{MergeInsertBuilder, MergeResult};
-use crate::table::{LsmWriteSpec, NativeTable};
+use crate::table::{BaseTable, LsmWriteSpec, NativeTable};
 
 /// Spec id of the sole sharding spec installed by [`set_lsm_write_spec`].
 /// Must match Lance's `InitializeMemWalBuilder` (`SHARDING_SPEC_ID`).
@@ -80,32 +82,44 @@ pub(crate) async fn set_lsm_write_spec(table: &NativeTable, spec: LsmWriteSpec)
         }
     }
 
+    // Before the builder borrows the dataset clone. `list_indices` merges an
+    // index's segments into one entry, so the result needs no dedup.
+    let maintained_indexes = {
+        let dataset = table.dataset.get().await?;
+        resolve_maintained_indexes(
+            &dataset,
+            &table.list_indices().await?,
+            spec.maintained_indexes(),
+        )
+        .await?
+    };
+
     let mut dataset = (*table.dataset.get().await?).clone();
     let mut builder = dataset.initialize_mem_wal();
-    let (maintained_indexes, writer_config_defaults) = match spec {
+    let writer_config_defaults = match spec {
         LsmWriteSpec::Bucket {
             column,
             num_buckets,
-            maintained_indexes,
             writer_config_defaults,
+            ..
         } => {
             builder = builder.bucket_sharding(column, num_buckets);
-            (maintained_indexes, writer_config_defaults)
+            writer_config_defaults
         }
         LsmWriteSpec::Identity {
             column,
-            maintained_indexes,
             writer_config_defaults,
+            ..
         } => {
             builder = builder.identity_sharding(column);
-            (maintained_indexes, writer_config_defaults)
+            writer_config_defaults
         }
         LsmWriteSpec::Unsharded {
-            maintained_indexes,
             writer_config_defaults,
+            ..
         } => {
             builder = builder.unsharded();
-            (maintained_indexes, writer_config_defaults)
+            writer_config_defaults
         }
     };
     builder = builder.maintained_indexes(maintained_indexes);
@@ -117,6 +131,58 @@ pub(crate) async fn set_lsm_write_spec(table: &NativeTable, spec: LsmWriteSpec)
     Ok(())
 }
 
+/// Resolve a spec's maintained-index selection against `indices`, as reported
+/// by [`Table::list_indices`](crate::Table::list_indices).
+///
+/// `None` means every index on the table, snapshotted now. Lance validates
+/// either selection against its shard-writer rules, so a spec that installs is
+/// one the MemWAL can open.
+///
+/// An unmaintainable index fails an inferred set rather than being dropped from
+/// it — dropping would leave the caller believing it is maintained.
+async fn resolve_maintained_indexes(
+    dataset: &Dataset,
+    indices: &[IndexConfig],
+    requested: Option<&[String]>,
+) -> Result<Vec<String>> {
+    let Some(requested) = requested else {
+        let all: Vec<String> = indices.iter().map(|index| index.name.clone()).collect();
+        validate_maintained_indexes(dataset, &all)
+            .await
+            .map_err(|source| Error::InvalidInput {
+                message: format!(
+                    "cannot maintain every index on this table: {source}. Set \
+                     maintained_indexes explicitly to choose from {}",
+                    index_name_list(indices),
+                ),
+            })?;
+        return Ok(all);
+    };
+    for name in requested {
+        if !indices.iter().any(|index| &index.name == name) {
+            return Err(Error::InvalidInput {
+                message: format!(
+                    "maintained index '{}' does not exist on this table; it has {}",
+                    name,
+                    index_name_list(indices),
+                ),
+            });
+        }
+    }
+    validate_maintained_indexes(dataset, requested).await?;
+    Ok(requested.to_vec())
+}
+
+/// Index names for an error message.
+fn index_name_list(indices: &[IndexConfig]) -> String {
+    if indices.is_empty() {
+        return "no indexes".to_string();
+    }
+    let mut names: Vec<&str> = indices.iter().map(|index| index.name.as_str()).collect();
+    names.sort_unstable();
+    format!("[{}]", names.join(", "))
+}
+
 // =============================================================================
 // unset_lsm_write_spec
 // =============================================================================