fix: forward commit outcome capabilities

Merge remote-tracking branch 'origin/main' into gatekeeper/fix-2623-1
chore: update lance dependency to v11.0.0-beta.8 (#3928 )
2026-08-26 16:08:43 +00:00 · 2026-08-13 16:52:23 +00:00 · 2026-08-13 16:03:12 +00:00 · 2026-08-14 00:00:18 +08:00 · 2026-08-13 15:53:12 +00:00 · 2026-08-13 21:22:42 +08:00
127 changed files with 15450 additions and 2300 deletions
@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "0.37.1-beta.0"
+current_version = "0.37.1-beta.1"
 parse = """(?x)
    (?P<major>0|[1-9]\\d*)\\.
    (?P<minor>0|[1-9]\\d*)\\.
@@ -0,0 +1,243 @@
+name: Check doc links
+
+# Checking external links is inherently noisy: third-party sites rate-limit
+# automated clients, reject non-browser user agents, and go down temporarily.
+# Blocking pull requests on that trades a lot of false failures for very little
+# signal, so this runs on a schedule and reports findings in a single tracking
+# issue instead of failing anyone's build.
+on:
+  schedule:
+    - cron: "0 7 * * *"
+  workflow_dispatch:
+
+# The report lives in one repository-global issue, so runs must not overlap: a
+# lookup racing a create produces duplicate issues, and a healthy run closing
+# the issue while a failing run only rewrites its body would leave a broken
+# report closed. The group is deliberately ref-independent so that a manual
+# dispatch serializes against the scheduled run.
+concurrency:
+  group: docs-link-check
+  cancel-in-progress: false
+
+permissions: {}
+
+env:
+  REPORT_TITLE: "Docs link checker report"
+
+jobs:
+  scan:
+    name: Scan links
+    runs-on: ubuntu-24.04
+    # lychee-action is pinned by SHA, but its wrapper downloads the lychee
+    # release tarball at run time without verifying a digest, and hands the
+    # resulting binary a GitHub token. Release assets remain replaceable, so
+    # that binary is confined to a job whose token can only read public
+    # content; everything that writes runs in the report job below.
+    permissions:
+      contents: read
+    outputs:
+      checker_outcome: ${{ steps.lychee.outcome }}
+      exit_code: ${{ steps.lychee.outputs.exit_code }}
+      status: ${{ steps.validate.outputs.status }}
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v6
+        with:
+          # workflow_dispatch can run from any ref, but the report is
+          # repository-global. Always measure the default branch so a manual
+          # run from a topic branch cannot close a report that main warrants,
+          # or overwrite it with branch-only findings.
+          ref: ${{ github.event.repository.default_branch }}
+          persist-credentials: false
+
+      - name: Check links
+        id: lychee
+        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
+          # API reference (the js/ tree comes from `npm run docs` in nodejs)
+          # and the hand-written pages use mkdocstrings cross-references and
+          # nav-relative paths that only resolve in the site mkdocs builds,
+          # not in this checkout, so relative links would be reported as
+          # broken on every run.
+          args: >-
+            --scheme https
+            --scheme http
+            --no-progress
+            --max-retries 3
+            --timeout 20
+            'docs/src/**/*.md'
+          format: json
+          output: ./lychee/out.json
+          jobSummary: false
+          # The report 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 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: |
+          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.validate.outputs.status == 'findings'
+        uses: actions/upload-artifact@v7
+        with:
+          name: link-report
+          path: ./lychee/out.json
+          retention-days: 7
+
+  report:
+    name: Update report issue
+    needs: scan
+    runs-on: ubuntu-24.04
+    # Deliberately no checkout: this job needs the report artifact and the
+    # issues API, not the repository contents.
+    permissions:
+      issues: write
+    env:
+      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: Find existing report issue
+        id: report
+        # Matched on title alone, and through search rather than a listing:
+        # the issue action applies labels in a separate call after creating the
+        # issue, so a label filter misses a half-created report, and this
+        # repository has far more open issues than one listing page holds.
+        # Closed issues are included because a healthy run closes the report:
+        # an open-only lookup would forget that identity and the next failing
+        # run would open a duplicate. The oldest match stays the canonical
+        # report and is reopened below when a problem recurs.
+        run: |
+          match=$(gh issue list --repo "$GITHUB_REPOSITORY" --state all \
+            --search "in:title \"$REPORT_TITLE\" author:app/github-actions" \
+            --limit 50 --json number,title,state \
+            --jq "[.[] | select(.title == \"$REPORT_TITLE\")] | sort_by(.number) | first // empty")
+          echo "number=$(jq -r '.number // empty' <<<"$match")" >> "$GITHUB_OUTPUT"
+          echo "state=$(jq -r '.state // empty' <<<"$match")" >> "$GITHUB_OUTPUT"
+
+      - name: Download report
+        if: env.STATUS == 'findings'
+        uses: actions/download-artifact@v8
+        with:
+          name: link-report
+          path: ./lychee
+
+      - name: Compose report
+        if: env.STATUS == 'findings'
+        run: |
+          run_url="$GITHUB_SERVER_URL/$GITHUB_REPOSITORY/actions/runs/$GITHUB_RUN_ID"
+          {
+            echo "Broken documentation links found by [\`$GITHUB_WORKFLOW\`]($run_url)."
+            echo
+            echo "This issue is rewritten by every scheduled run and closed automatically once all links resolve."
+            echo
+            echo "Entries can be false positives: some sites rate-limit or block automated clients while working fine in a browser. Confirm before editing the docs, and add persistent offenders to \`--exclude\` in \`.github/workflows/docs-link-check.yml\`."
+            echo
+            # Timeouts are reported alongside errors: entries land in
+            # timeout_map with a status text instead of an HTTP code.
+            jq -r '
+              "\(.errors) of \(.total) links failed, \(.timeouts) timed out.",
+              "",
+              ([(.error_map | to_entries[]), (.timeout_map | to_entries[])]
+               | group_by(.key)[] |
+                "### Errors in \(.[0].key)",
+                "",
+                (map(.value[])[] | "* [\(.status.code // .status.text // "ERR")] <\(.url)> — \(.status.details // .status.text // "unknown error")"),
+                "")
+            ' ./lychee/out.json
+          } > ./lychee/issue.md
+
+      - name: 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, 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 "The documentation link checker reported a problem again in [the latest run]($run_url)."
+
+      - 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
+          # the same issue is updated in place.
+          issue-number: ${{ steps.report.outputs.number }}
+          title: ${{ env.REPORT_TITLE }}
+          content-filepath: ./lychee/issue.md
+          labels: documentation
+
+      - name: Close report issue once links are healthy
+        # An OPEN state implies the lookup found a canonical issue; a report
+        # that is already closed needs nothing.
+        if: >-
+          env.STATUS == 'healthy' &&
+          steps.report.outputs.state == 'OPEN'
+        env:
+          ISSUE_NUMBER: ${{ steps.report.outputs.number }}
+        run: |
+          run_url="$GITHUB_SERVER_URL/$GITHUB_REPOSITORY/actions/runs/$GITHUB_RUN_ID"
+          gh issue close "$ISSUE_NUMBER" --repo "$GITHUB_REPOSITORY" \
+            --comment "All documentation links resolved in [the latest run]($run_url)."
@@ -69,6 +69,16 @@ jobs:
        uses: actions/setup-python@v6
        with:
          python-version: "3.10"
+      - name: Add swap for Arm fat LTO
+        if: matrix.config.platform == 'aarch64'
+        shell: bash
+        run: |
+          swap_file="$RUNNER_TEMP/lancedb-swap"
+          sudo fallocate --length 16G "$swap_file"
+          sudo chmod 600 "$swap_file"
+          sudo mkswap "$swap_file"
+          sudo swapon "$swap_file"
+          free -h
      - uses: ./.github/workflows/build_linux_wheel
        with:
          python-minor-version: 10
@@ -296,16 +296,18 @@ jobs:
          cargo update -p aws-types --precise 1.3.9
          cargo update -p aws-sigv4 --precise 1.3.5
          cargo update -p aws-credential-types --precise 1.2.8
-          cargo update -p aws-smithy-checksums --precise 0.63.9
+          # aws-smithy-checksums must stay at or above 0.63.13: OpenDAL's S3
+          # service needs crc-fast ~1.9, and older releases pin it to ~1.3.
+          cargo update -p aws-smithy-checksums --precise 0.63.13
          cargo update -p aws-smithy-runtime --precise 1.9.3
-          cargo update -p aws-smithy-http --precise 0.62.4
-          cargo update -p aws-smithy-eventstream --precise 0.60.12
+          cargo update -p aws-smithy-http --precise 0.62.6
+          cargo update -p aws-smithy-eventstream --precise 0.60.14
          cargo update -p aws-smithy-http-client --precise 1.1.3
          cargo update -p aws-smithy-observability --precise 0.1.4
          cargo update -p aws-smithy-query --precise 0.60.8
-          cargo update -p aws-smithy-runtime-api --precise 1.9.1
-          cargo update -p aws-smithy-async --precise 1.2.6
-          cargo update -p aws-smithy-types --precise 1.3.5
+          cargo update -p aws-smithy-runtime-api --precise 1.9.3
+          cargo update -p aws-smithy-async --precise 1.2.7
+          cargo update -p aws-smithy-types --precise 1.3.6
          cargo update -p aws-smithy-xml --precise 0.60.11
          cargo update -p home --precise 0.5.9
      - name: cargo +${{ matrix.msrv }} check
@@ -13,20 +13,20 @@ categories = ["database-implementations"]
 rust-version = "1.91.0"

 [workspace.dependencies]
-lance = { "version" = "=10.0.0-beta.5", default-features = false, "tag" = "v10.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
-lance-core = { "version" = "=10.0.0-beta.5", "tag" = "v10.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
-lance-datagen = { "version" = "=10.0.0-beta.5", "tag" = "v10.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
-lance-file = { "version" = "=10.0.0-beta.5", "tag" = "v10.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
-lance-io = { "version" = "=10.0.0-beta.5", default-features = false, "tag" = "v10.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
-lance-index = { "version" = "=10.0.0-beta.5", "tag" = "v10.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
-lance-linalg = { "version" = "=10.0.0-beta.5", "tag" = "v10.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace = { "version" = "=10.0.0-beta.5", "tag" = "v10.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace-impls = { "version" = "=10.0.0-beta.5", default-features = false, "tag" = "v10.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
-lance-table = { "version" = "=10.0.0-beta.5", "tag" = "v10.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
-lance-testing = { "version" = "=10.0.0-beta.5", "tag" = "v10.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
-lance-datafusion = { "version" = "=10.0.0-beta.5", "tag" = "v10.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
-lance-encoding = { "version" = "=10.0.0-beta.5", "tag" = "v10.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
-lance-arrow = { "version" = "=10.0.0-beta.5", "tag" = "v10.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
+lance = { "version" = "=11.0.0-beta.8", default-features = false, "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-core = { "version" = "=11.0.0-beta.8", "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-datagen = { "version" = "=11.0.0-beta.8", "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-file = { "version" = "=11.0.0-beta.8", "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-io = { "version" = "=11.0.0-beta.8", default-features = false, "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-index = { "version" = "=11.0.0-beta.8", "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-linalg = { "version" = "=11.0.0-beta.8", "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace = { "version" = "=11.0.0-beta.8", "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace-impls = { "version" = "=11.0.0-beta.8", default-features = false, "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-table = { "version" = "=11.0.0-beta.8", "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-testing = { "version" = "=11.0.0-beta.8", "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-datafusion = { "version" = "=11.0.0-beta.8", "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-encoding = { "version" = "=11.0.0-beta.8", "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-arrow = { "version" = "=11.0.0-beta.8", "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
 ahash = "0.8"
 # Note that this one does not include pyarrow
 arrow = { version = "58.0.0", optional = false }
@@ -52,7 +52,7 @@ env_logger = "0.11"
 half = { "version" = "2.7.1", default-features = false, features = [
    "num-traits",
 ] }
-futures = "0"
+futures = "0.3"
 log = "0.4"
 metrics = "0.24"
 metrics-util = "0.19"
@@ -101,6 +101,13 @@ ignore = [
    # https://rustsec.org/advisories/RUSTSEC-2026-0195
    { id = "RUSTSEC-2026-0194", reason = "transitive via inferno/lance/opendal; XML from trusted cloud endpoints, not attacker-controlled" },
    { id = "RUSTSEC-2026-0195", reason = "transitive via inferno/lance/opendal; XML from trusted cloud endpoints, not attacker-controlled" },
+    # smartstring: unmaintained — the repository was archived by its author on
+    # 2026-05-03. Not a vulnerability. Reached only transitively through polars
+    # (polars-core/-io/-ops/-time/-utils); nothing in LanceDB depends on it directly.
+    # The advisory states no safe upgrade is available: upstream recommends
+    # compact_str/smol_str, so clearing this requires polars to migrate.
+    # https://rustsec.org/advisories/RUSTSEC-2026-0249
+    { id = "RUSTSEC-2026-0249", reason = "smartstring unmaintained via polars; no fixed upstream release" },
 ]

 # ---------------------------------------------------------------------------
@@ -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>
 ```

@@ -25,6 +25,27 @@ the underlying connection has been closed.

 ## Methods

+### cancelJob()
+
+```ts
+abstract cancelJob(jobId): Promise<boolean>
+```
+
+Request cancellation of a server-side job by id.
+
+Resolves to true if the server accepted the cancellation, false if no
+such job exists. Cancelling an already-terminal job is a no-op success.
+
+#### Parameters
+
+* **jobId**: `string`
+
+#### Returns
+
+`Promise`&lt;`boolean`&gt;
+
+***
+
 ### cloneTable()

 ```ts
@@ -365,6 +386,26 @@ Drop an existing table.

 ***

+### getJob()
+
+```ts
+abstract getJob(jobId): Promise<null | JobDescription>
+```
+
+Describe a single server-side job by id.
+
+Resolves to `null` when the server has no such job.
+
+#### Parameters
+
+* **jobId**: `string`
+
+#### Returns
+
+`Promise`&lt;`null` \| [`JobDescription`](../interfaces/JobDescription.md)&gt;
+
+***
+
 ### isOpen()

 ```ts
@@ -379,6 +420,62 @@ Return true if the connection has not been closed

 ***

+### job()
+
+```ts
+abstract job(jobId): Job
+```
+
+A [Job](Job.md) handle for a server-side job by id.
+
+The handle is constructed without a server round trip; an unknown id
+surfaces when the handle is used. Dropping the handle has no effect on
+the job itself.
+
+#### Parameters
+
+* **jobId**: `string`
+
+#### Returns
+
+[`Job`](Job.md)
+
+***
+
+### jobHistory()
+
+```ts
+abstract jobHistory(jobId?): Promise<Table<any>>
+```
+
+The lifecycle event history of a server-side job, as an Arrow table.
+
+Lists history across all jobs when `jobId` is omitted.
+
+#### Parameters
+
+* **jobId?**: `string`
+
+#### Returns
+
+`Promise`&lt;`Table`&lt;`any`&gt;&gt;
+
+***
+
+### listJobs()
+
+```ts
+abstract listJobs(): Promise<JobInfo[]>
+```
+
+List server-side jobs across the database's tables.
+
+#### Returns
+
+`Promise`&lt;[`JobInfo`](../interfaces/JobInfo.md)[]&gt;
+
+***
+
 ### listNamespaces()

 ```ts
@@ -0,0 +1,83 @@
+[**@lancedb/lancedb**](../README.md) • **Docs**
+
+***
+
+[@lancedb/lancedb](../globals.md) / Job
+
+# Class: Job
+
+A handle to an operation that may still be running.
+
+## Constructors
+
+### new Job()
+
+```ts
+new Job(): Job
+```
+
+#### Returns
+
+[`Job`](Job.md)
+
+## Accessors
+
+### id
+
+```ts
+get id(): null | string
+```
+
+Identifies the operation on the server that is running it. Operations
+that run in this process have no server id. The value is opaque.
+
+#### Returns
+
+`null` \| `string`
+
+## Methods
+
+### cancel()
+
+```ts
+cancel(): Promise<void>
+```
+
+Request cancellation. Cancelling a finished operation is a no-op.
+
+#### Returns
+
+`Promise`&lt;`void`&gt;
+
+***
+
+### status()
+
+```ts
+status(): Promise<string>
+```
+
+The operation's current lifecycle state: "running", "finished",
+"failed", or "cancelled".
+
+A point snapshot; unlike [Job.wait](Job.md#wait) it does not block or reject
+on a terminal failure state. States a newer server reports that this
+client version does not know pass through as-is.
+
+#### Returns
+
+`Promise`&lt;`string`&gt;
+
+***
+
+### wait()
+
+```ts
+wait(): Promise<void>
+```
+
+Wait until the operation reaches a terminal state.
+
+#### Returns
+
+`Promise`&lt;`void`&gt;
@@ -295,6 +295,29 @@ await table.createIndex("my_float_col");

 ***

+### createIndexAsync()
+
+```ts
+abstract createIndexAsync(column, options?): Promise<Job>
+```
+
+Create an index, returning a handle to the indexing job.
+
+The job may already be complete when returned; callers must not assume
+the index exists until [Job.wait](Job.md#wait) resolves.
+
+#### Parameters
+
+* **column**: `string`
+
+* **options?**: `Partial`&lt;[`IndexOptions`](../interfaces/IndexOptions.md)&gt;
+
+#### Returns
+
+`Promise`&lt;[`Job`](Job.md)&gt;
+
+***
+
 ### currentBranch()

 ```ts
@@ -408,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

@@ -783,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)
@@ -25,6 +25,7 @@
 - [Connection](classes/Connection.md)
 - [HeaderProvider](classes/HeaderProvider.md)
 - [Index](classes/Index.md)
+- [Job](classes/Job.md)
 - [MakeArrowTableOptions](classes/MakeArrowTableOptions.md)
 - [MatchQuery](classes/MatchQuery.md)
 - [MergeInsertBuilder](classes/MergeInsertBuilder.md)
@@ -88,6 +89,9 @@
 - [IvfFlatOptions](interfaces/IvfFlatOptions.md)
 - [IvfPqOptions](interfaces/IvfPqOptions.md)
 - [IvfRqOptions](interfaces/IvfRqOptions.md)
+- [JobDescription](interfaces/JobDescription.md)
+- [JobFailureInfo](interfaces/JobFailureInfo.md)
+- [JobInfo](interfaces/JobInfo.md)
 - [ListNamespacesOptions](interfaces/ListNamespacesOptions.md)
 - [ListNamespacesResponse](interfaces/ListNamespacesResponse.md)
 - [LsmWriteSpec](interfaces/LsmWriteSpec.md)
@@ -0,0 +1,66 @@
+[**@lancedb/lancedb**](../README.md) • **Docs**
+
+***
+
+[@lancedb/lancedb](../globals.md) / JobDescription
+
+# Interface: JobDescription
+
+A described job from `Connection.getJob`.
+
+## Properties
+
+### creationMs
+
+```ts
+creationMs: number;
+```
+
+When the job was created, in milliseconds since the epoch.
+
+***
+
+### failure?
+
+```ts
+optional failure: JobFailureInfo;
+```
+
+Why the job failed, when the job is failed and the server reports a
+reason.
+
+***
+
+### jobId
+
+```ts
+jobId: string;
+```
+
+***
+
+### jobType
+
+```ts
+jobType: string;
+```
+
+***
+
+### specJson?
+
+```ts
+optional specJson: string;
+```
+
+The job-type-specific specification as a JSON string, when present.
+
+***
+
+### state
+
+```ts
+state: string;
+```
+
+Lifecycle state: "running", "finished", "failed", or "cancelled".
@@ -0,0 +1,33 @@
+[**@lancedb/lancedb**](../README.md) • **Docs**
+
+***
+
+[@lancedb/lancedb](../globals.md) / JobFailureInfo
+
+# Interface: JobFailureInfo
+
+The server's account of why a job failed.
+
+## Properties
+
+### message?
+
+```ts
+optional message: string;
+```
+
+***
+
+### phase?
+
+```ts
+optional phase: string;
+```
+
+***
+
+### retryable?
+
+```ts
+optional retryable: boolean;
+```
@@ -0,0 +1,58 @@
+[**@lancedb/lancedb**](../README.md) • **Docs**
+
+***
+
+[@lancedb/lancedb](../globals.md) / JobInfo
+
+# Interface: JobInfo
+
+A row from `Connection.listJobs`: one server-side job.
+
+## Properties
+
+### createdAtMillis
+
+```ts
+createdAtMillis: number;
+```
+
+When the job was created, in milliseconds since the epoch.
+
+***
+
+### jobId
+
+```ts
+jobId: string;
+```
+
+The job id -- what `Connection.getJob` and `Connection.cancelJob`
+accept.
+
+***
+
+### jobType
+
+```ts
+jobType: string;
+```
+
+***
+
+### state
+
+```ts
+state: string;
+```
+
+Lifecycle state: "running", "finished", "failed", or "cancelled".
+
+***
+
+### table
+
+```ts
+table: string;
+```
+
+The table the job runs against, without URI or namespace.
@@ -34,7 +34,9 @@ Bucket and identity variants: the sharding column.
 optional maintainedIndexes: string[];
 ```

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

 ***

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

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

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

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

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

    <groupId>com.lancedb</groupId>
    <artifactId>lancedb-parent</artifactId>
-    <version>0.37.1-beta.0</version>
+    <version>0.37.1-beta.1</version>
    <packaging>pom</packaging>
    <name>${project.artifactId}</name>
    <description>LanceDB Java SDK Parent POM</description>
@@ -28,7 +28,7 @@
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <arrow.version>15.0.0</arrow.version>
-        <lance-core.version>10.0.0-beta.5</lance-core.version>
+        <lance-core.version>11.0.0-beta.8</lance-core.version>
        <spotless.skip>false</spotless.skip>
        <spotless.version>2.30.0</spotless.version>
        <spotless.java.googlejavaformat.version>1.7</spotless.java.googlejavaformat.version>
@@ -1,7 +1,7 @@
 [package]
 name = "lancedb-nodejs"
 edition.workspace = true
-version = "0.37.1-beta.0"
+version = "0.37.1-beta.1"
 publish = false
 license.workspace = true
 description.workspace = true
@@ -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"];
@@ -197,6 +204,35 @@ describe.each([arrow15, arrow16, arrow17, arrow18])(
        expect(table.getChild("d")?.toJSON()).toEqual([9n, 10n, null]);
      });

+      it("will use a provided FixedSizeList schema with typed array values", function () {
+        const schema = new Schema([
+          new Field("text", new Utf8(), false),
+          new Field(
+            "vector",
+            new FixedSizeList(3, new Field("item", new Float32(), false)),
+            false,
+          ),
+        ]);
+
+        const table = makeArrowTable(
+          [
+            {
+              text: "foo",
+              vector: new Float32Array([1, 2, 3]),
+            },
+          ],
+          { schema },
+        );
+
+        expect(table.getChild("text")?.toJSON()).toEqual(["foo"]);
+        expect(
+          table
+            .getChild("vector")
+            ?.toJSON()
+            .map((value) => value.toJSON()),
+        ).toEqual([[1, 2, 3]]);
+      });
+
      it("will assume the column `vector` is FixedSizeList<Float32> by default", async function () {
        const schema = new Schema([
          new Field("a", new Float(Precision.DOUBLE), true),
@@ -1025,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()),
@@ -11,8 +11,11 @@ import {
  Float16,
  Float32,
  Float64,
+  Int32,
  Schema,
  Utf8,
+  fromDataToBuffer,
+  tableFromIPC,
 } from "../lancedb/arrow";
 import { EmbeddingFunction, LanceSchema } from "../lancedb/embedding";
 import { getRegistry, register } from "../lancedb/embedding/registry";
@@ -184,6 +187,63 @@ describe("embedding functions", () => {
    const vector0 = JSON.parse(JSON.stringify(arr[0].vector));
    expect(vector0).toEqual([1, 2, 3]);
  });
+
+  it("should append generated vectors to a non-nullable schema", async () => {
+    @register("non_nullable_schema_test")
+    class MockEmbeddingFunction extends EmbeddingFunction<string> {
+      ndims() {
+        return 3;
+      }
+      embeddingDataType(): Float {
+        return new Float64();
+      }
+      async computeSourceEmbeddings(data: string[]) {
+        return data.map(() => [1, 2, 3]);
+      }
+    }
+
+    const schema = new Schema([
+      new Field("id", new Int32()),
+      new Field("text", new Utf8()),
+      new Field("type", new Utf8()),
+      new Field(
+        "vector",
+        new FixedSizeList(3, new Field("item", new Float64())),
+      ),
+    ]);
+    const func = new MockEmbeddingFunction();
+    const db = await connect(tmpDir.name);
+    const table = await db.createEmptyTable("test_non_nullable", schema, {
+      embeddingFunction: {
+        function: func,
+        sourceColumn: "text",
+      },
+    });
+
+    const data = [
+      { id: 1, text: "Carrot", type: "vegetable" },
+      { id: 2, text: "Apple", type: "fruit" },
+    ];
+    const buffer = await fromDataToBuffer(
+      data,
+      undefined,
+      await table.schema(),
+    );
+    const generatedTable = tableFromIPC(buffer);
+    const vectorField = generatedTable.schema.fields.find(
+      (field) => field.name === "vector",
+    );
+    expect(vectorField?.nullable).toBe(false);
+
+    await table.add(data);
+
+    const rows = await table.query().toArray();
+    expect(rows).toHaveLength(2);
+    for (const row of rows) {
+      expect([...row.vector]).toEqual([1, 2, 3]);
+    }
+  });
+
  it("should error when appending to a table with an unregistered embedding function", async () => {
    @register("mock")
    class MockEmbeddingFunction extends EmbeddingFunction<string> {
@@ -0,0 +1,14 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+import packageJson = require("../package.json");
+
+describe("package metadata", () => {
+  it("requires Node.js type declarations compatible with the runtime", () => {
+    expect(packageJson.engines.node).toBe(">= 18");
+    expect(packageJson.peerDependencies["@types/node"]).toBe(">=18");
+    expect(packageJson.peerDependenciesMeta["@types/node"]).toEqual({
+      optional: true,
+    });
+  });
+});
@@ -110,6 +110,81 @@ describe("Query outputSchema", () => {
  });
 });

+describe("Search pagination", () => {
+  let tmpDir: tmp.DirResult;
+  let table: Table;
+
+  beforeEach(async () => {
+    tmpDir = tmp.dirSync({ unsafeCleanup: true });
+    const db = await connect(tmpDir.name);
+    const schema = new Schema([
+      new Field("id", new Int64(), false),
+      new Field("text", new Utf8(), false),
+      new Field(
+        "vector",
+        new FixedSizeList(2, new Field("item", new Float32())),
+        false,
+      ),
+    ]);
+    const data = makeArrowTable(
+      [
+        { id: 1n, text: "common", vector: [0, 0] },
+        { id: 2n, text: "common common", vector: [1, 1] },
+        { id: 3n, text: "common common common", vector: [2, 2] },
+        { id: 4n, text: "common common common common", vector: [3, 3] },
+      ],
+      { schema },
+    );
+    table = await db.createTable("test", data);
+  });
+
+  afterEach(() => {
+    tmpDir.removeCallback();
+  });
+
+  it("applies offset after the vector search limit", async () => {
+    const allResults = await table
+      .vectorSearch([0, 0])
+      .select(["id"])
+      .limit(4)
+      .toArray();
+    const secondPage = await table
+      .vectorSearch([0, 0])
+      .select(["id"])
+      .limit(2)
+      .offset(2)
+      .toArray();
+
+    expect(allResults).toHaveLength(4);
+    expect(secondPage).toHaveLength(2);
+    expect(secondPage.map((row) => row.id)).toEqual(
+      allResults.slice(2, 4).map((row) => row.id),
+    );
+  });
+
+  it("applies offset after the full-text search limit", async () => {
+    await table.createIndex("text", { config: Index.fts() });
+
+    const allResults = await table
+      .search("common", "fts")
+      .select(["id"])
+      .limit(4)
+      .toArray();
+    const secondPage = await table
+      .search("common", "fts")
+      .select(["id"])
+      .limit(2)
+      .offset(2)
+      .toArray();
+
+    expect(allResults).toHaveLength(4);
+    expect(secondPage).toHaveLength(2);
+    expect(secondPage.map((row) => row.id)).toEqual(
+      allResults.slice(2, 4).map((row) => row.id),
+    );
+  });
+});
+
 describe("Query orderBy", () => {
  let tmpDir: tmp.DirResult;
  let table: Table;
@@ -170,6 +170,38 @@ describe("remote connection", () => {
    );
  });

+  it("surfaces JSON server errors from remote table operations", async () => {
+    await withMockDatabase(
+      (req, res) => {
+        const path = req.url ?? "";
+        if (path.endsWith("/describe/")) {
+          res.writeHead(200, { "Content-Type": "application/json" }).end(
+            JSON.stringify({
+              name: "broken_table",
+              version: 1,
+              schema: { fields: [] },
+            }),
+          );
+          return;
+        }
+
+        if (path.endsWith("/count_rows/")) {
+          res
+            .writeHead(400, { "Content-Type": "application/json" })
+            .end(JSON.stringify({ error: "count rows failed" }));
+          return;
+        }
+
+        res.writeHead(404).end();
+      },
+      async (db) => {
+        const table = await db.openTable("broken_table");
+
+        await expect(table.countRows()).rejects.toThrow("count rows failed");
+      },
+    );
+  });
+
  it("should pass on requested extra headers", async () => {
    await withMockDatabase(
      (req, res) => {
@@ -877,3 +909,96 @@ describe("remote connection", () => {
    });
  });
 });
+
+describe("remote connection jobs surface", () => {
+  it("lists, describes, cancels, and reads history", async () => {
+    const { tableFromArrays, tableToIPC } = await import("apache-arrow");
+    const eventsTable = tableFromArrays({ state: ["created", "succeeded"] });
+    const eventsBody = Buffer.from(tableToIPC(eventsTable, "stream"));
+
+    await withMockDatabase(
+      (req, res) => {
+        let body = "";
+        req.on("data", (chunk) => {
+          body += chunk;
+        });
+        req.on("end", () => {
+          const payload = body.length > 0 ? JSON.parse(body) : {};
+          if (req.url === "/v1/jobs/list") {
+            if (payload["page_token"] === undefined) {
+              res
+                .writeHead(200, { "Content-Type": "application/json" })
+                .end(
+                  '{"jobs": [{"job_id": "job-1", "table": "t1", ' +
+                    '"job_type": "create_index", "state": "in_progress", ' +
+                    '"created_at_millis": 1000}], "page_token": "next"}',
+                );
+            } else {
+              res
+                .writeHead(200, { "Content-Type": "application/json" })
+                .end(
+                  '{"jobs": [{"job_id": "job-2", "table": "t2", ' +
+                    '"job_type": "create_index", "state": "succeeded", ' +
+                    '"created_at_millis": 2000}]}',
+                );
+            }
+          } else if (req.url === "/v1/jobs/describe") {
+            if (payload["job_id"] !== "job-1") {
+              res.writeHead(404).end("no such job");
+              return;
+            }
+            res
+              .writeHead(200, { "Content-Type": "application/json" })
+              .end(
+                '{"job_id": "job-1", "job_type": "create_index", ' +
+                  '"job_state": "FAILED", "creation_ms": 1000, ' +
+                  '"spec": {"column": "vec"}, "failure": {"phase": "execute", ' +
+                  '"message": "worker died", "retryable": true}}',
+              );
+          } else if (req.url === "/v1/jobs/cancel") {
+            if (payload["job_id"] !== "job-1") {
+              res.writeHead(404).end("no such job");
+              return;
+            }
+            res
+              .writeHead(200, { "Content-Type": "application/json" })
+              .end('{"job_id": "job-1"}');
+          } else if (req.url === "/v1/jobs/query_events") {
+            res
+              .writeHead(200, {
+                "Content-Type": "application/vnd.apache.arrow.stream",
+              })
+              .end(eventsBody);
+          } else {
+            res.writeHead(404).end();
+          }
+        });
+      },
+      async (db) => {
+        const jobs = await db.listJobs();
+        expect(jobs.map((job) => job.jobId)).toEqual(["job-1", "job-2"]);
+        expect(jobs[0].state).toEqual("running");
+        expect(jobs[1].state).toEqual("finished");
+
+        const description = await db.getJob("job-1");
+        expect(description?.state).toEqual("failed");
+        expect(JSON.parse(description?.specJson ?? "")).toEqual({
+          column: "vec",
+        });
+        expect(description?.failure?.message).toEqual("worker died");
+        expect(await db.getJob("missing")).toBeNull();
+
+        expect(await db.cancelJob("job-1")).toBe(true);
+        expect(await db.cancelJob("missing")).toBe(false);
+
+        const history = await db.jobHistory("job-1");
+        expect(history.numRows).toEqual(2);
+
+        const job = db.job("job-1");
+        expect(job.id).toEqual("job-1");
+        expect(await job.status()).toEqual("failed");
+        await expect(job.wait()).rejects.toThrow("worker died");
+      },
+    );
+  });
+});
@@ -86,6 +86,44 @@ describe.each([arrow15, arrow16, arrow17, arrow18])(
      await expect(table.countRows()).resolves.toBe(3);
    });

+    it("should support a foreign Float64 vector schema end to end", async () => {
+      const conn = await connect(tmpDir.name);
+      const schema = new arrow.Schema([
+        new arrow.Field("resource_id", new arrow.Int32(), false),
+        new arrow.Field(
+          "vector",
+          new arrow.FixedSizeList(
+            3,
+            new arrow.Field("value", new arrow.Float64(), true),
+          ),
+          false,
+        ),
+      ]);
+      const data = [
+        {
+          // biome-ignore lint/style/useNamingConvention: matches the reported schema
+          resource_id: 0,
+          vector: [0.1, 0.1, 0.1],
+        },
+      ];
+
+      const resources = await conn.createTable("resources", data, { schema });
+
+      const existing = await resources
+        .query()
+        .where("resource_id = 0")
+        .limit(1)
+        .toArray();
+      expect(existing).toHaveLength(1);
+
+      const matched = await resources
+        .search(Float64Array.from(data[0].vector))
+        .limit(1)
+        .toArray();
+      expect(matched).toHaveLength(1);
+      expect(matched[0]["resource_id"]).toBe(0);
+    });
+
    it("should support branches", async () => {
      await table.add([{ id: 1 }]);
      expect(await table.countRows()).toBe(1);
@@ -239,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 () => {
@@ -851,7 +897,11 @@ describe("When creating an index", () => {
  afterEach(() => tmpDir.removeCallback());

  it("should create a vector index on vector columns", async () => {
-    await tbl.createIndex("vec");
+    const job = await tbl.createIndexAsync("vec");
+    expect(job.id).toBeNull();
+    await job.wait();
+    // Cancelling a job that already finished succeeds and does nothing.
+    await job.cancel();

    // check index directory
    const indexDir = path.join(tmpDir.name, "test.lance", "_indices");
@@ -1,6 +1,7 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright The LanceDB Authors

+import { tableFromIPC } from "apache-arrow";
 import {
  Data,
  SchemaLike,
@@ -20,6 +21,9 @@ import type {
  CreateNamespaceResponse,
  DescribeNamespaceResponse,
  DropNamespaceResponse,
+  Job,
+  JobDescription,
+  JobInfo,
  ListNamespacesResponse,
 } from "./native";
 export type {
@@ -436,6 +440,40 @@ export abstract class Connection {
    newName: string,
    options?: RenameTableOptions,
  ): Promise<void>;
+
+  /**
+   * A {@link Job} handle for a server-side job by id.
+   *
+   * The handle is constructed without a server round trip; an unknown id
+   * surfaces when the handle is used. Dropping the handle has no effect on
+   * the job itself.
+   */
+  abstract job(jobId: string): Job;
+
+  /** List server-side jobs across the database's tables. */
+  abstract listJobs(): Promise<JobInfo[]>;
+
+  /**
+   * Describe a single server-side job by id.
+   *
+   * Resolves to `null` when the server has no such job.
+   */
+  abstract getJob(jobId: string): Promise<JobDescription | null>;
+
+  /**
+   * Request cancellation of a server-side job by id.
+   *
+   * Resolves to true if the server accepted the cancellation, false if no
+   * such job exists. Cancelling an already-terminal job is a no-op success.
+   */
+  abstract cancelJob(jobId: string): Promise<boolean>;
+
+  /**
+   * The lifecycle event history of a server-side job, as an Arrow table.
+   *
+   * Lists history across all jobs when `jobId` is omitted.
+   */
+  abstract jobHistory(jobId?: string): Promise<ArrowTable>;
 }

 /** @hideconstructor */
@@ -722,6 +760,30 @@ export class LocalConnection extends Connection {
      options?.newNamespacePath,
    );
  }
+
+  job(jobId: string): Job {
+    return this.inner.job(jobId);
+  }
+
+  async listJobs(): Promise<JobInfo[]> {
+    return this.inner.listJobs();
+  }
+
+  async getJob(jobId: string): Promise<JobDescription | null> {
+    return this.inner.getJob(jobId);
+  }
+
+  async cancelJob(jobId: string): Promise<boolean> {
+    return this.inner.cancelJob(jobId);
+  }
+
+  async jobHistory(jobId?: string): Promise<ArrowTable> {
+    const buf = await this.inner.jobHistory(jobId);
+    if (buf.length === 0) {
+      return new ArrowTable();
+    }
+    return tableFromIPC(buf);
+  }
 }

 /**
@@ -85,7 +85,13 @@ export {
  RenameTableOptions,
 } from "./connection";

-export { Session } from "./native.js";
+export {
+  Job,
+  JobDescription,
+  JobFailureInfo,
+  JobInfo,
+  Session,
+} from "./native.js";

 export {
  ExecutableQuery,
@@ -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 = {
@@ -30,6 +30,7 @@ import {
  DropColumnsResult,
  IndexConfig,
  IndexStatistics,
+  Job,
  Branches as NativeBranches,
  OptimizeStats,
  TableStatistics,
@@ -196,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>;
@@ -358,6 +363,17 @@ export abstract class Table {
    options?: Partial<IndexOptions>,
  ): Promise<void>;

+  /**
+   * Create an index, returning a handle to the indexing job.
+   *
+   * The job may already be complete when returned; callers must not assume
+   * the index exists until {@link Job.wait} resolves.
+   */
+  abstract createIndexAsync(
+    column: string,
+    options?: Partial<IndexOptions>,
+  ): Promise<Job>;
+
  /**
   * Drop an index from the table.
   *
@@ -583,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
@@ -610,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>;
@@ -940,6 +962,22 @@ export class LocalTable extends Table {
    );
  }

+  async createIndexAsync(
+    column: string,
+    options?: Partial<IndexOptions>,
+  ): Promise<Job> {
+    // biome-ignore lint/suspicious/noExplicitAny: skip
+    const nativeIndex = (options?.config as any)?.inner;
+    return await this.inner.createIndexAsync(
+      nativeIndex,
+      column,
+      options?.replace,
+      options?.waitTimeoutSeconds,
+      options?.name,
+      options?.train,
+    );
+  }
+
  async dropIndex(name: string): Promise<void> {
    await this.inner.dropIndex(name);
  }
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-darwin-arm64",
-	"version": "0.37.1-beta.0",
+	"version": "0.37.1-beta.1",
 	"os": ["darwin"],
 	"cpu": ["arm64"],
 	"main": "lancedb.darwin-arm64.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-gnu",
-	"version": "0.37.1-beta.0",
+	"version": "0.37.1-beta.1",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-gnu.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-musl",
-	"version": "0.37.1-beta.0",
+	"version": "0.37.1-beta.1",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-musl.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-gnu",
-	"version": "0.37.1-beta.0",
+	"version": "0.37.1-beta.1",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-gnu.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-musl",
-	"version": "0.37.1-beta.0",
+	"version": "0.37.1-beta.1",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-musl.node",
@@ -1,6 +1,6 @@
 {
  "name": "@lancedb/lancedb-win32-arm64-msvc",
-  "version": "0.37.1-beta.0",
+  "version": "0.37.1-beta.1",
  "os": [
    "win32"
  ],
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-win32-x64-msvc",
-	"version": "0.37.1-beta.0",
+	"version": "0.37.1-beta.1",
 	"os": ["win32"],
 	"cpu": ["x64"],
 	"main": "lancedb.win32-x64-msvc.node",
@@ -1,12 +1,12 @@
 {
  "name": "@lancedb/lancedb",
-  "version": "0.37.1-beta.0",
+  "version": "0.37.1-beta.1",
  "lockfileVersion": 3,
  "requires": true,
  "packages": {
    "": {
      "name": "@lancedb/lancedb",
-      "version": "0.37.1-beta.0",
+      "version": "0.37.1-beta.1",
      "cpu": [
        "x64",
        "arm64"
@@ -55,7 +55,13 @@
        "openai": "4.29.2"
      },
      "peerDependencies": {
+        "@types/node": ">=18",
        "apache-arrow": ">=15.0.0 <=18.1.0"
+      },
+      "peerDependenciesMeta": {
+        "@types/node": {
+          "optional": true
+        }
      }
    },
    "node_modules/@aws-crypto/crc32": {
@@ -11,7 +11,7 @@
    "ann"
  ],
  "private": false,
-  "version": "0.37.1-beta.0",
+  "version": "0.37.1-beta.1",
  "main": "dist/index.js",
  "exports": {
    ".": "./dist/index.js",
@@ -101,6 +101,12 @@
    "openai": "4.29.2"
  },
  "peerDependencies": {
+    "@types/node": ">=18",
    "apache-arrow": ">=15.0.0 <=18.1.0"
+  },
+  "peerDependenciesMeta": {
+    "@types/node": {
+      "optional": true
+    }
  }
 }
@@ -340,6 +340,69 @@ impl Connection {
        self.get_inner()?.drop_all_tables(&ns).await.default_error()
    }

+    /// A `Job` handle for a server-side job by id.
+    ///
+    /// The handle is constructed without a server round trip; an unknown id
+    /// surfaces when the handle is used.
+    #[napi]
+    pub fn job(&self, job_id: String) -> napi::Result<crate::job::Job> {
+        let job = self.get_inner()?.job(job_id).default_error()?;
+        Ok(crate::job::Job::new(job))
+    }
+
+    /// List server-side jobs across the database's tables.
+    #[napi(catch_unwind)]
+    pub async fn list_jobs(&self) -> napi::Result<Vec<crate::job::JobInfo>> {
+        let jobs = self.get_inner()?.list_jobs().await.default_error()?;
+        Ok(jobs.into_iter().map(Into::into).collect())
+    }
+
+    /// Describe a single server-side job by id. `null` when the server has
+    /// no such job.
+    #[napi(catch_unwind)]
+    pub async fn get_job(
+        &self,
+        job_id: String,
+    ) -> napi::Result<Option<crate::job::JobDescription>> {
+        let description = self.get_inner()?.get_job(&job_id).await.default_error()?;
+        Ok(description.map(Into::into))
+    }
+
+    /// Request cancellation of a server-side job by id. Returns true if the
+    /// server accepted the cancellation, false if no such job exists.
+    #[napi(catch_unwind)]
+    pub async fn cancel_job(&self, job_id: String) -> napi::Result<bool> {
+        self.get_inner()?.cancel_job(&job_id).await.default_error()
+    }
+
+    /// The lifecycle event history of a server-side job (all jobs when
+    /// `job_id` is null), as an Arrow IPC stream buffer. Empty when there is
+    /// no history.
+    #[napi(catch_unwind)]
+    pub async fn job_history(&self, job_id: Option<String>) -> napi::Result<Buffer> {
+        let batches = self
+            .get_inner()?
+            .job_history(job_id.as_deref())
+            .await
+            .default_error()?;
+        let Some(first) = batches.first() else {
+            return Ok(Buffer::from(Vec::<u8>::new()));
+        };
+        let mut out = Vec::new();
+        let mut writer = arrow_ipc::writer::StreamWriter::try_new(&mut out, &first.schema())
+            .map_err(|e| napi::Error::from_reason(e.to_string()))?;
+        for batch in &batches {
+            writer
+                .write(batch)
+                .map_err(|e| napi::Error::from_reason(e.to_string()))?;
+        }
+        writer
+            .finish()
+            .map_err(|e| napi::Error::from_reason(e.to_string()))?;
+        drop(writer);
+        Ok(Buffer::from(out))
+    }
+
    #[napi(catch_unwind)]
    /// Describe a namespace and return its properties.
    pub async fn describe_namespace(
@@ -0,0 +1,123 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+use std::sync::Arc;
+
+use napi_derive::napi;
+
+use crate::error::NapiErrorExt;
+
+/// A handle to an operation that may still be running.
+#[napi]
+pub struct Job {
+    inner: Arc<lancedb::Job>,
+}
+
+impl Job {
+    pub(crate) fn new(inner: lancedb::Job) -> Self {
+        Self {
+            inner: Arc::new(inner),
+        }
+    }
+}
+
+#[napi]
+impl Job {
+    /// Identifies the operation on the server that is running it. Operations
+    /// that run in this process have no server id. The value is opaque.
+    #[napi(getter)]
+    pub fn id(&self) -> Option<String> {
+        self.inner.id().map(str::to_string)
+    }
+
+    /// The operation's current lifecycle state: "running", "finished",
+    /// "failed", or "cancelled".
+    ///
+    /// A point snapshot; unlike {@link Job.wait} it does not block or reject
+    /// on a terminal failure state. States a newer server reports that this
+    /// client version does not know pass through as-is.
+    #[napi(catch_unwind)]
+    pub async fn status(&self) -> napi::Result<String> {
+        self.inner.status().await.default_error()
+    }
+
+    /// Wait until the operation reaches a terminal state.
+    #[napi(catch_unwind)]
+    pub async fn wait(&self) -> napi::Result<()> {
+        self.inner.wait().await.default_error()
+    }
+
+    /// Request cancellation. Cancelling a finished operation is a no-op.
+    #[napi(catch_unwind)]
+    pub async fn cancel(&self) -> napi::Result<()> {
+        self.inner.cancel().await.default_error()
+    }
+}
+
+/// A row from `Connection.listJobs`: one server-side job.
+#[napi(object)]
+pub struct JobInfo {
+    /// The job id -- what `Connection.getJob` and `Connection.cancelJob`
+    /// accept.
+    pub job_id: String,
+    /// The table the job runs against, without URI or namespace.
+    pub table: String,
+    pub job_type: String,
+    /// Lifecycle state: "running", "finished", "failed", or "cancelled".
+    pub state: String,
+    /// When the job was created, in milliseconds since the epoch.
+    pub created_at_millis: i64,
+}
+
+impl From<lancedb::database::JobInfo> for JobInfo {
+    fn from(info: lancedb::database::JobInfo) -> Self {
+        Self {
+            job_id: info.job_id,
+            table: info.table,
+            job_type: info.job_type,
+            state: info.state,
+            created_at_millis: info.created_at_millis,
+        }
+    }
+}
+
+/// The server's account of why a job failed.
+#[napi(object)]
+pub struct JobFailureInfo {
+    pub phase: Option<String>,
+    pub message: Option<String>,
+    pub retryable: Option<bool>,
+}
+
+/// A described job from `Connection.getJob`.
+#[napi(object)]
+pub struct JobDescription {
+    pub job_id: String,
+    pub job_type: String,
+    /// Lifecycle state: "running", "finished", "failed", or "cancelled".
+    pub state: String,
+    /// When the job was created, in milliseconds since the epoch.
+    pub creation_ms: i64,
+    /// The job-type-specific specification as a JSON string, when present.
+    pub spec_json: Option<String>,
+    /// Why the job failed, when the job is failed and the server reports a
+    /// reason.
+    pub failure: Option<JobFailureInfo>,
+}
+
+impl From<lancedb::database::JobDescription> for JobDescription {
+    fn from(description: lancedb::database::JobDescription) -> Self {
+        Self {
+            job_id: description.job_id,
+            job_type: description.job_type,
+            state: description.state,
+            creation_ms: description.creation_ms,
+            spec_json: (!description.spec.is_null()).then(|| description.spec.to_string()),
+            failure: description.failure.map(|failure| JobFailureInfo {
+                phase: failure.phase,
+                message: failure.message,
+                retryable: failure.retryable,
+            }),
+        }
+    }
+}
@@ -11,6 +11,7 @@ mod error;
 mod header;
 mod index;
 mod iterator;
+mod job;
 pub mod merge;
 pub mod otel;
 pub mod permutation;
@@ -168,6 +168,39 @@ impl Table {
        builder.execute().await.default_error()
    }

+    #[napi(catch_unwind)]
+    pub async fn create_index_async(
+        &self,
+        index: Option<&Index>,
+        column: String,
+        replace: Option<bool>,
+        wait_timeout_s: Option<i64>,
+        name: Option<String>,
+        train: Option<bool>,
+    ) -> napi::Result<crate::job::Job> {
+        let lancedb_index = if let Some(index) = index {
+            index.consume()?
+        } else {
+            lancedb::index::Index::Auto
+        };
+        let mut builder = self.inner_ref()?.create_index(&[column], lancedb_index);
+        if let Some(replace) = replace {
+            builder = builder.replace(replace);
+        }
+        if let Some(timeout) = wait_timeout_s {
+            builder =
+                builder.wait_timeout(std::time::Duration::from_secs(timeout.try_into().unwrap()));
+        }
+        if let Some(name) = name {
+            builder = builder.name(name);
+        }
+        if let Some(train) = train {
+            builder = builder.train(train);
+        }
+        let job = builder.execute_async().await.default_error()?;
+        Ok(crate::job::Job::new(job))
+    }
+
    #[napi(catch_unwind)]
    pub async fn drop_index(&self, index_name: String) -> napi::Result<()> {
        self.inner_ref()?
@@ -306,7 +339,9 @@ impl Table {
        let transforms = NewColumnTransform::SqlExpressions(transforms);
        let res = self
            .inner_ref()?
-            .add_columns(transforms, None)
+            .add_columns()
+            .transform(transforms)
+            .execute()
            .await
            .default_error()?;
        Ok(res.into())
@@ -323,7 +358,9 @@ impl Table {
        let transforms = NewColumnTransform::AllNulls(schema);
        let res = self
            .inner_ref()?
-            .add_columns(transforms, None)
+            .add_columns()
+            .transform(transforms)
+            .execute()
            .await
            .default_error()?;
        Ok(res.into())
@@ -735,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>>,
@@ -745,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" => {
@@ -772,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))
    }
 }
@@ -790,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 {
@@ -801,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 {
@@ -811,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),
            },
        }
@@ -1006,7 +1043,10 @@ impl From<lancedb::index::IndexStatistics> for IndexStatistics {

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

    /// The number of rows in the table
@@ -1,6 +1,6 @@
 [package]
 name = "lancedb-python"
-version = "0.37.1-beta.0"
+version = "0.37.1-beta.1"
 publish = false
 edition.workspace = true
 description = "Python bindings for LanceDB"
@@ -26,7 +26,7 @@ lance-namespace-impls.workspace = true
 lance-io.workspace = true
 env_logger.workspace = true
 log.workspace = true
-pyo3 = { version = "0.28", features = ["extension-module", "abi3-py39", "chrono"] }
+pyo3 = { version = "0.28", features = ["extension-module", "abi3-py310", "chrono"] }
 chrono = { version = "0.4", default-features = false, features = ["clock"] }
 pyo3-async-runtimes = { version = "0.28", features = [
    "attributes",
@@ -43,7 +43,7 @@ libc = "0.2"
 [build-dependencies]
 pyo3-build-config = { version = "0.28", features = [
    "extension-module",
-    "abi3-py39",
+    "abi3-py310",
 ] }

 [features]
@@ -60,7 +60,7 @@ tests = [
    "pytest-asyncio>=0.21",
    "duckdb>=0.9.0",
    "pytz>=2023.3",
-    "polars>=0.19, <=1.3.0",
+    "polars>=0.19, <=1.32.3",
    "pyarrow<25",
    "pyarrow-stubs>=16.0",
    "pylance==9.0.0rc1",
@@ -140,6 +140,7 @@ include = [
    "python/lancedb/remote/errors.py",
    "python/lancedb/embeddings/__init__.py",
    "python/lancedb/_lancedb.pyi",
+    "python/type_tests/connect.py",
 ]
 exclude = ["python/tests/"]
 pythonVersion = "3.13"
@@ -20,6 +20,7 @@ from .remote import ClientConfig
 from .remote.db import RemoteDBConnection
 from .expr import Expr, col, lit, func
 from .schema import blob, vector, BlobType
+from .job import AsyncJob, Job
 from .table import AsyncTable, Table
 from .types import BaseTokenizerType
 from ._lancedb import Session
@@ -500,6 +501,7 @@ __all__ = [
    "connect_namespace",
    "connect_namespace_async",
    "AsyncConnection",
+    "AsyncJob",
    "AsyncLanceNamespaceDBConnection",
    "AsyncTable",
    "FtsToken",
@@ -513,6 +515,7 @@ __all__ = [
    "BlobType",
    "vector",
    "DBConnection",
+    "Job",
    "LanceDBConnection",
    "LanceNamespaceDBConnection",
    "RemoteDBConnection",
@@ -14,14 +14,10 @@ import pyarrow as pa
 from .expr import Expr
 from .schema import blob_v2_column_paths
 from .types import BlobMode, QueryProjection, QueryProjectionSpec
-from .util import get_uri_scheme

 if TYPE_CHECKING:
    from _typeshed import WriteableBuffer

-    from .remote.table import RemoteTable
-    from .table import AsyncTable, Table
-
 BLOB_MODE_TO_HANDLING = {
    "lazy": "blobs_descriptions",
    "bytes": "all_binary",
@@ -104,22 +100,6 @@ def validate_blob_mode(blob_mode: BlobMode) -> None:
        raise ValueError(f"blob_mode must be one of {modes}, got {blob_mode!r}")


-def supports_blob_auto_row_id(table: Table | AsyncTable | RemoteTable) -> bool:
-    """Blob auto row-id applies to native tables, not LanceDB Cloud."""
-    from .remote.table import RemoteTable
-
-    if isinstance(table, RemoteTable):
-        return False
-
-    inner = getattr(table, "_inner", None)
-    if inner is not None:
-        uri = inner.database().uri
-        if isinstance(uri, str) and get_uri_scheme(uri) == "db":
-            return False
-
-    return True
-
-
 def projection_includes_blob_column(
    projection: QueryProjection,
    blob_columns: Iterable[str],
@@ -164,16 +144,14 @@ def v2_projection_needs_row_id(


 def blob_auto_row_id_for_scan(
-    table: Table | AsyncTable | RemoteTable,
    schema: pa.Schema,
    projection: QueryProjection,
    *,
    with_row_id: bool | None,
 ) -> bool:
+    """Auto row-id only applies when the caller said nothing about row ids."""
    if with_row_id is not None:
        return False
-    if not supports_blob_auto_row_id(table):
-        return False
    return v2_projection_needs_row_id(schema, projection, with_row_id=False)


@@ -186,6 +164,11 @@ def finalize_blob_query_table(
 ) -> pa.Table:
    if user_requested_row_id or not blob_auto_row_id:
        return tbl
+    if "_rowid" not in tbl.column_names:
+        # A backend that ignores the row-id request leaves nothing to stash. Hand
+        # back the projection as-is so fetch_blobs raises the error that names the
+        # ways to supply row ids, rather than failing here about a hidden column.
+        return tbl
    return stash_auto_row_ids(tbl, blob_paths)


@@ -146,6 +146,13 @@ class Connection(object):
        start_after: Optional[str],
        limit: Optional[int],
    ) -> list[str]: ...  # Deprecated: Use list_tables instead
+    def job(self, job_id: str) -> Job: ...
+    async def list_jobs(self) -> List[JobInfo]: ...
+    async def get_job(self, job_id: str) -> Optional[JobDescription]: ...
+    async def cancel_job(self, job_id: str) -> bool: ...
+    async def job_history(
+        self, job_id: Optional[str] = None
+    ) -> List[pa.RecordBatch]: ...
    async def create_table(
        self,
        name: str,
@@ -209,6 +216,47 @@ class BlobFile:
    def read_range(self, offset: int, length: int) -> bytes: ...
    def read_up_to(self, length: int) -> bytes: ...

+class Job:
+    @property
+    def id(self) -> Optional[str]: ...
+    async def status(self) -> str: ...
+    async def wait(self) -> None: ...
+    async def cancel(self) -> None: ...
+
+class JobInfo:
+    @property
+    def job_id(self) -> str: ...
+    @property
+    def table(self) -> str: ...
+    @property
+    def job_type(self) -> str: ...
+    @property
+    def state(self) -> str: ...
+    @property
+    def created_at_millis(self) -> int: ...
+
+class JobFailureInfo:
+    @property
+    def phase(self) -> Optional[str]: ...
+    @property
+    def message(self) -> Optional[str]: ...
+    @property
+    def retryable(self) -> Optional[bool]: ...
+
+class JobDescription:
+    @property
+    def job_id(self) -> str: ...
+    @property
+    def job_type(self) -> str: ...
+    @property
+    def state(self) -> str: ...
+    @property
+    def creation_ms(self) -> int: ...
+    @property
+    def spec_json(self) -> Optional[str]: ...
+    @property
+    def failure(self) -> Optional[JobFailureInfo]: ...
+
 class Table:
    def name(self) -> str: ...
    def __repr__(self) -> str: ...
@@ -248,6 +296,28 @@ class Table:
        name: Optional[str],
        train: Optional[bool],
    ): ...
+    async def create_index_async(
+        self,
+        column: str,
+        index: Union[
+            IvfFlat,
+            IvfSq,
+            IvfPq,
+            HnswPq,
+            HnswSq,
+            HnswFlat,
+            BTree,
+            Bitmap,
+            LabelList,
+            Fm,
+            FTS,
+        ],
+        replace: Optional[bool],
+        wait_timeout: Optional[object],
+        *,
+        name: Optional[str],
+        train: Optional[bool],
+    ) -> Job: ...
    async def list_versions(self) -> List[Dict[str, Any]]: ...
    async def version(self) -> int: ...
    async def checkout(self, version: Union[int, str]): ...
@@ -285,6 +355,10 @@ class Table:
    async def set_lsm_write_spec(self, spec: LsmWriteSpec) -> None: ...
    async def unset_lsm_write_spec(self) -> None: ...
    async def get_lsm_write_spec(self) -> Optional[LsmWriteSpec]: ...
+    async def checkpoint_lsm(self) -> None: ...
+    async def flush_lsm(self) -> None: ...
+    async def compact_lsm(self) -> None: ...
+    async def get_lsm_stats(self, include_generation_rows: bool) -> Optional[dict]: ...
    async def close_lsm_writers(self) -> None: ...
    @property
    def tags(self) -> Tags: ...
@@ -579,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
@@ -596,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]: ...

@@ -45,6 +45,7 @@ from lance_namespace.errors import NamespaceNotEmptyError, TableNotFoundError

 from . import __version__
 from ._lancedb import connect as lancedb_connect  # type: ignore
+from .job import AsyncJob, Job
 from .table import (
    AsyncTable,
    LanceTable,
@@ -63,6 +64,7 @@ if TYPE_CHECKING:
    from .pydantic import LanceModel

    from ._lancedb import Connection as LanceDbConnection
+    from ._lancedb import JobDescription, JobInfo
    from .common import DATA, URI
    from .embeddings import EmbeddingFunctionConfig
    from ._lancedb import Session
@@ -178,6 +180,51 @@ class DBConnection(EnforceOverrides):
            "Namespace operations are not supported for this connection type"
        )

+    def namespace_exists(self, namespace_id: List[str]) -> bool:
+        """Check if a namespace exists.
+
+        Parameters
+        ----------
+        namespace_id: List[str]
+            The namespace identifier to check.
+
+        Returns
+        -------
+        bool
+            True if the namespace exists, False otherwise.
+
+        Raises
+        ------
+        NotImplementedError
+            If the connection type does not support namespace operations.
+        """
+        raise NotImplementedError(
+            "Namespace operations are not supported for this connection type"
+        )
+
+    def table_exists(self, table_id: List[str]) -> bool:
+        """Check if a table exists.
+
+        Parameters
+        ----------
+        table_id: List[str]
+            The table identifier to check (full path including namespace
+            segments and table name).
+
+        Returns
+        -------
+        bool
+            True if the table exists, False otherwise.
+
+        Raises
+        ------
+        NotImplementedError
+            If the connection type does not support namespace operations.
+        """
+        raise NotImplementedError(
+            "Namespace operations are not supported for this connection type"
+        )
+
    def list_tables(
        self,
        namespace_path: Optional[List[str]] = None,
@@ -563,6 +610,46 @@ class DBConnection(EnforceOverrides):
        """
        raise NotImplementedError("serialize is not supported for this connection type")

+    def job(self, job_id: str) -> Job:
+        """A [Job][lancedb.job.Job] handle for a server-side job by id.
+
+        The handle is constructed without a server round trip; an unknown id
+        surfaces when the handle is used. Dropping the handle has no effect
+        on the job itself.
+        """
+        raise NotImplementedError("job is not supported for this connection type")
+
+    def list_jobs(self) -> List[JobInfo]:
+        """List server-side jobs across the database's tables."""
+        raise NotImplementedError("list_jobs is not supported for this connection type")
+
+    def get_job(self, job_id: str) -> Optional[JobDescription]:
+        """Describe a single server-side job by id.
+
+        Returns None when the server has no such job.
+        """
+        raise NotImplementedError("get_job is not supported for this connection type")
+
+    def cancel_job(self, job_id: str) -> bool:
+        """Request cancellation of a server-side job by id.
+
+        Returns True if the server accepted the cancellation, False if no
+        such job exists. Cancelling an already-terminal job is a no-op
+        success.
+        """
+        raise NotImplementedError(
+            "cancel_job is not supported for this connection type"
+        )
+
+    def job_history(self, job_id: Optional[str] = None) -> List[pa.RecordBatch]:
+        """The lifecycle event history of a server-side job, as Arrow batches.
+
+        Lists history across all jobs when `job_id` is None.
+        """
+        raise NotImplementedError(
+            "job_history is not supported for this connection type"
+        )
+

 class LanceDBConnection(DBConnection):
    """
@@ -620,26 +707,18 @@ class LanceDBConnection(DBConnection):
        self._namespace_client_properties = namespace_client_properties
        if _inner is not None:
            self._conn = _inner
+            # Native-derived wrappers resolve this in their async reconstruction
+            # path so construction never synchronously re-enters LOOP.
+            self._read_consistency_interval = read_consistency_interval
            self._cached_namespace_client = None
            return

        if not isinstance(uri, Path):
            scheme = get_uri_scheme(uri)
        is_local = isinstance(uri, Path) or scheme == "file"
-        if is_local:
+        is_file_uri = isinstance(uri, str) and uri.lower().startswith("file:")
+        if is_local and not is_file_uri:
            if isinstance(uri, str):
-                # Strip file:// or file:/ scheme if present
-                # file:///path becomes file:/path after URL normalization
-                if uri.startswith("file://"):
-                    uri = uri[7:]  # Remove "file://"
-                elif uri.startswith("file:/"):
-                    uri = uri[5:]  # Remove "file:"
-
-                if sys.platform == "win32":
-                    # On Windows, a path like /C:/path should become C:/path
-                    if len(uri) >= 3 and uri[0] == "/" and uri[2] == ":":
-                        uri = uri[1:]
-
                uri = Path(uri)
            uri = uri.expanduser().absolute()
            Path(uri).mkdir(parents=True, exist_ok=True)
@@ -669,11 +748,14 @@ class LanceDBConnection(DBConnection):
        # storage_options.  Also, this class really shouldn't be holding any state
        # beyond _conn.
        self._conn = AsyncConnection(LOOP.run(do_connect()))
+        # Keep property access synchronous so debugger introspection cannot wait on
+        # the background loop while that thread is suspended at a breakpoint.
+        self._read_consistency_interval = read_consistency_interval
        self._cached_namespace_client: Optional[LanceNamespace] = None

    @property
    def read_consistency_interval(self) -> Optional[timedelta]:
-        return LOOP.run(self._conn.get_read_consistency_interval())
+        return self._read_consistency_interval

    @property
    def session(self) -> Optional[Session]:
@@ -684,15 +766,19 @@ class LanceDBConnection(DBConnection):
        return self._conn.uri

    @classmethod
-    def from_inner(cls, inner: LanceDbConnection):
-        return cls(None, _inner=inner)
+    def from_inner(
+        cls,
+        inner: LanceDbConnection,
+        read_consistency_interval: Optional[timedelta],
+    ):
+        return cls(
+            None,
+            read_consistency_interval=read_consistency_interval,
+            _inner=inner,
+        )

    def __repr__(self) -> str:
-        val = f"{self.__class__.__name__}(uri={self._conn.uri!r}"
-        if self.read_consistency_interval is not None:
-            val += f", read_consistency_interval={repr(self.read_consistency_interval)}"
-        val += ")"
-        return val
+        return f"{self.__class__.__name__}(uri={self._conn.uri!r})"

    @override
    def serialize(self) -> str:
@@ -1129,6 +1215,47 @@ class LanceDBConnection(DBConnection):
            )
        )

+    @override
+    def job(self, job_id: str) -> Job:
+        """A [Job][lancedb.job.Job] handle for a server-side job by id.
+
+        The handle is constructed without a server round trip; an unknown id
+        surfaces when the handle is used. Dropping the handle has no effect
+        on the job itself.
+        """
+        return Job(self._conn.job(job_id))
+
+    @override
+    def list_jobs(self) -> List[JobInfo]:
+        """List server-side jobs across the database's tables."""
+        return LOOP.run(self._conn.list_jobs())
+
+    @override
+    def get_job(self, job_id: str) -> Optional[JobDescription]:
+        """Describe a single server-side job by id.
+
+        Returns None when the server has no such job.
+        """
+        return LOOP.run(self._conn.get_job(job_id))
+
+    @override
+    def cancel_job(self, job_id: str) -> bool:
+        """Request cancellation of a server-side job by id.
+
+        Returns True if the server accepted the cancellation, False if no
+        such job exists. Cancelling an already-terminal job is a no-op
+        success.
+        """
+        return LOOP.run(self._conn.cancel_job(job_id))
+
+    @override
+    def job_history(self, job_id: Optional[str] = None) -> List[pa.RecordBatch]:
+        """The lifecycle event history of a server-side job, as Arrow batches.
+
+        Lists history across all jobs when `job_id` is None.
+        """
+        return LOOP.run(self._conn.job_history(job_id))
+
    @override
    def namespace_client(self) -> LanceNamespace:
        """Get the equivalent namespace client for this connection.
@@ -1838,6 +1965,43 @@ class AsyncConnection(object):
            namespace_path = []
        await self._inner.drop_all_tables(namespace_path=namespace_path)

+    def job(self, job_id: str) -> AsyncJob:
+        """An [AsyncJob][lancedb.job.AsyncJob] handle for a server-side job
+        by id.
+
+        The handle is constructed without a server round trip; an unknown id
+        surfaces when the handle is used. Dropping the handle has no effect
+        on the job itself.
+        """
+        return AsyncJob(self._inner.job(job_id))
+
+    async def list_jobs(self) -> List[JobInfo]:
+        """List server-side jobs across the database's tables."""
+        return await self._inner.list_jobs()
+
+    async def get_job(self, job_id: str) -> Optional[JobDescription]:
+        """Describe a single server-side job by id.
+
+        Returns None when the server has no such job.
+        """
+        return await self._inner.get_job(job_id)
+
+    async def cancel_job(self, job_id: str) -> bool:
+        """Request cancellation of a server-side job by id.
+
+        Returns True if the server accepted the cancellation, False if no
+        such job exists. Cancelling an already-terminal job is a no-op
+        success.
+        """
+        return await self._inner.cancel_job(job_id)
+
+    async def job_history(self, job_id: Optional[str] = None) -> List[pa.RecordBatch]:
+        """The lifecycle event history of a server-side job, as Arrow batches.
+
+        Lists history across all jobs when `job_id` is None.
+        """
+        return await self._inner.job_history(job_id)
+
    async def namespace_client(self) -> LanceNamespace:
        """Get the equivalent namespace client for this connection.

@@ -101,8 +101,7 @@ class InstructorEmbeddingFunction(TextEmbeddingFunction):

    @weak_lru(maxsize=1)
    def ndims(self):
-        model = self.get_model()
-        return model.encode("foo").shape[0]
+        return len(self.generate_embeddings([[self.source_instruction, "foo"]])[0])

    def compute_query_embeddings(self, query: str, *args, **kwargs) -> List[np.array]:
        return self.generate_embeddings([[self.query_instruction, query]])
@@ -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)))
@@ -23,3 +23,15 @@ class MissingColumnError(KeyError):
        return (
            f"Error: Column '{self.column_name}' does not exist in the DataFrame object"
        )
+
+
+class JobFailedError(RuntimeError):
+    """Exception raised when an asynchronous job reaches the failed state."""
+
+    pass
+
+
+class JobCancelledError(RuntimeError):
+    """Exception raised when an asynchronous job was cancelled."""
+
+    pass
@@ -0,0 +1,105 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+"""Handles to operations a server may run asynchronously."""
+
+import asyncio
+from datetime import timedelta
+from typing import Optional
+
+from lancedb.background_loop import LOOP
+
+from . import _lancedb
+
+
+class AsyncJob:
+    """A handle to an operation that may still be running.
+
+    The operation may already be complete when the handle is created.
+    """
+
+    def __init__(self, inner: Optional["_lancedb.Job"]):
+        self._inner = inner
+
+    @property
+    def id(self) -> Optional[str]:
+        """Identifies the operation on the server that is running it.
+
+        Returned for correlating with server logs or the jobs API. Operations
+        that run in this process have no server id and return `None`. The value
+        is opaque: parsing it or storing it to resume the job later is not
+        supported.
+        """
+        return self._inner.id if self._inner is not None else None
+
+    async def status(self) -> str:
+        """The operation's current lifecycle state: "running", "finished",
+        "failed", or "cancelled".
+
+        A point snapshot; unlike `wait` it does not block or raise on a
+        terminal failure state. States a newer server reports that this
+        client version does not know pass through as-is.
+        """
+        if self._inner is None:
+            return "finished"
+        return await self._inner.status()
+
+    async def wait(self, timeout: Optional[timedelta] = None):
+        """Wait until the operation reaches a terminal state.
+
+        Raises `JobFailedError` if the operation failed, `JobCancelledError`
+        if it was cancelled, and `TimeoutError` if `timeout` elapses first.
+        """
+        if self._inner is None:
+            return
+        if timeout is None:
+            await self._inner.wait()
+        else:
+            await asyncio.wait_for(self._inner.wait(), timeout.total_seconds())
+
+    async def cancel(self):
+        """Request cancellation. Cancelling a finished operation is a no-op."""
+        if self._inner is None:
+            return
+        await self._inner.cancel()
+
+
+class Job:
+    """Synchronous counterpart of `AsyncJob`."""
+
+    def __init__(self, inner: Optional[AsyncJob]):
+        self._inner = inner
+
+    @property
+    def id(self) -> Optional[str]:
+        """Identifies the operation on the server that is running it.
+
+        See :attr:`AsyncJob.id`.
+        """
+        return self._inner.id if self._inner is not None else None
+
+    def status(self) -> str:
+        """The operation's current lifecycle state: "running", "finished",
+        "failed", or "cancelled".
+
+        See :meth:`AsyncJob.status`.
+        """
+        if self._inner is None:
+            return "finished"
+        return LOOP.run(self._inner.status())
+
+    def wait(self, timeout: Optional[timedelta] = None):
+        """Block until the operation reaches a terminal state.
+
+        Raises `JobFailedError` if the operation failed, `JobCancelledError`
+        if it was cancelled, and `TimeoutError` if `timeout` elapses first.
+        """
+        if self._inner is None:
+            return
+        LOOP.run(self._inner.wait(timeout))
+
+    def cancel(self):
+        """Request cancellation. Cancelling a finished operation is a no-op."""
+        if self._inner is None:
+            return
+        LOOP.run(self._inner.cancel())
@@ -92,8 +92,10 @@ class LanceMergeInsertBuilder(object):
        self._when_not_matched_by_source_delete = True
        if isinstance(condition, Expr):
            self._when_not_matched_by_source_condition_expr = condition._inner
-        elif condition is not None:
+            self._when_not_matched_by_source_condition = None
+        else:
            self._when_not_matched_by_source_condition = condition
+            self._when_not_matched_by_source_condition_expr = None
        return self

    def use_index(self, use_index: bool) -> LanceMergeInsertBuilder:
@@ -38,7 +38,11 @@ from lance_namespace_urllib3_client.models.query_table_request_vector import (
    QueryTableRequestVector,
 )
 from lance_namespace_urllib3_client.models.string_fts_query import StringFtsQuery
-from lance_namespace.errors import NamespaceNotEmptyError, TableNotFoundError
+from lance_namespace.errors import (
+    NamespaceNotEmptyError,
+    NamespaceNotFoundError,
+    TableNotFoundError,
+)
 from lancedb._lancedb import (
    connect_namespace as _connect_namespace,
    connect_namespace_client as _connect_namespace_client,
@@ -53,6 +57,8 @@ from lance_namespace import (
    DropNamespaceResponse,
    ListNamespacesResponse,
    ListTablesResponse,
+    NamespaceExistsRequest,
+    TableExistsRequest,
 )
 from lancedb.table import AsyncTable, LanceTable, Table
 from lancedb.util import validate_table_name
@@ -780,6 +786,51 @@ class LanceNamespaceDBConnection(DBConnection):
        """
        return LOOP.run(self._inner.describe_namespace(namespace_path))

+    @override
+    def namespace_exists(self, namespace_id: List[str]) -> bool:
+        """
+        Check if a namespace exists.
+
+        Parameters
+        ----------
+        namespace_id : List[str]
+            The namespace identifier to check.
+
+        Returns
+        -------
+        bool
+            True if the namespace exists, False otherwise.
+        """
+        request = NamespaceExistsRequest(id=namespace_id)
+        try:
+            self._namespace_client.namespace_exists(request)
+            return True
+        except NamespaceNotFoundError:
+            return False
+
+    @override
+    def table_exists(self, table_id: List[str]) -> bool:
+        """
+        Check if a table exists.
+
+        Parameters
+        ----------
+        table_id : List[str]
+            The table identifier to check (full path including namespace
+            segments and table name).
+
+        Returns
+        -------
+        bool
+            True if the table exists, False otherwise.
+        """
+        request = TableExistsRequest(id=table_id)
+        try:
+            self._namespace_client.table_exists(request)
+            return True
+        except TableNotFoundError:
+            return False
+
    @override
    def list_tables(
        self,
@@ -868,17 +919,7 @@ class LanceNamespaceDBConnection(DBConnection):
            The namespace client for this connection.
        """
        if self._namespace_client is None:
-            if (
-                self._namespace_client_impl is None
-                or self._namespace_client_properties is None
-            ):
-                raise ValueError(
-                    "Cannot construct a Python namespace client without "
-                    "namespace implementation properties"
-                )
-            self._namespace_client = namespace_connect(
-                self._namespace_client_impl, self._namespace_client_properties
-            )
+            self._namespace_client = LOOP.run(self._inner.namespace_client())
        return self._namespace_client


@@ -1233,6 +1274,49 @@ class AsyncLanceNamespaceDBConnection:
        """
        return await self._inner.describe_namespace(namespace_path)

+    async def namespace_exists(self, namespace_id: List[str]) -> bool:
+        """
+        Check if a namespace exists.
+
+        Parameters
+        ----------
+        namespace_id : List[str]
+            The namespace identifier to check.
+
+        Returns
+        -------
+        bool
+            True if the namespace exists, False otherwise.
+        """
+        request = NamespaceExistsRequest(id=namespace_id)
+        try:
+            self._namespace_client.namespace_exists(request)
+            return True
+        except NamespaceNotFoundError:
+            return False
+
+    async def table_exists(self, table_id: List[str]) -> bool:
+        """
+        Check if a table exists.
+
+        Parameters
+        ----------
+        table_id : List[str]
+            The table identifier to check (full path including namespace
+            segments and table name).
+
+        Returns
+        -------
+        bool
+            True if the table exists, False otherwise.
+        """
+        request = TableExistsRequest(id=table_id)
+        try:
+            self._namespace_client.table_exists(request)
+            return True
+        except TableNotFoundError:
+            return False
+
    async def list_tables(
        self,
        namespace_path: Optional[List[str]] = None,
@@ -1276,17 +1360,7 @@ class AsyncLanceNamespaceDBConnection:
            The namespace client for this connection.
        """
        if self._namespace_client is None:
-            if (
-                self._namespace_client_impl is None
-                or self._namespace_client_properties is None
-            ):
-                raise ValueError(
-                    "Cannot construct a Python namespace client without "
-                    "namespace implementation properties"
-                )
-            self._namespace_client = namespace_connect(
-                self._namespace_client_impl, self._namespace_client_properties
-            )
+            self._namespace_client = await self._inner.namespace_client()
        return self._namespace_client


@@ -226,7 +226,7 @@ class PermutationBuilder:

        async def do_execute():
            inner_tbl = await self._async.execute()
-            return LanceTable.from_inner(inner_tbl)
+            return await LanceTable.from_inner(inner_tbl)

        return LOOP.run(do_execute())

@@ -0,0 +1 @@
+
@@ -153,6 +153,16 @@ def Vector(
    return FixedSizeList


+def _raise_bare_vector_error(*_args):
+    raise TypeError("Vector must be parameterized with a dimension, e.g. Vector(128).")
+
+
+# Pydantic v1 and v2 otherwise treat the bare Vector factory as a field validator
+# and inspect its signature, which produces misleading errors about internal types.
+setattr(Vector, "__get_validators__", _raise_bare_vector_error)
+setattr(Vector, "__get_pydantic_core_schema__", _raise_bare_vector_error)
+
+
 def MultiVector(
    dim: int, value_type: pa.DataType = pa.float32(), nullable: bool = True
 ) -> Type:
@@ -52,7 +52,6 @@ from ._blob import (
    finalize_blob_query_table,
    replace_v2_blob_columns_with_bytes,
    replace_v2_blob_columns_with_bytes_sync,
-    supports_blob_auto_row_id,
    validate_blob_mode,
 )
 from .types import BlobMode, QueryProjection
@@ -1280,10 +1279,7 @@ class LanceQueryBuilder(ABC):
        return self._with_row_id is True

    def _blob_auto_row_id_enabled(self) -> bool:
-        if not supports_blob_auto_row_id(self._table):
-            return False
        return blob_auto_row_id_for_scan(
-            self._table,
            self._table.schema,
            self._columns,
            with_row_id=self._with_row_id,
@@ -2701,7 +2697,7 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
            self._fts_query.phrase_query(True)
        if self._distance_type:
            self._vector_query.metric(self._distance_type)
-        if self._minimum_nprobes:
+        if self._minimum_nprobes is not None:
            self._vector_query.minimum_nprobes(self._minimum_nprobes)
        if self._maximum_nprobes is not None:
            self._vector_query.maximum_nprobes(self._maximum_nprobes)
@@ -2774,7 +2770,7 @@ class AsyncQueryBase(object):
        )

    async def _maybe_add_blob_row_id(self) -> None:
-        if self._table is None or not supports_blob_auto_row_id(self._table):
+        if self._table is None:
            self._blob_auto_row_id = False
            self._blob_paths = ()
            return
@@ -2782,7 +2778,6 @@ class AsyncQueryBase(object):
        req = self._inner.to_query_request()
        schema = await self._table.schema()
        self._blob_auto_row_id = blob_auto_row_id_for_scan(
-            self._table,
            schema,
            req.select,
            with_row_id=self._with_row_id,
@@ -3034,7 +3029,6 @@ class AsyncQueryBase(object):

        schema = await self._table.schema()
        blob_auto_row_id = blob_auto_row_id_for_scan(
-            self._table,
            schema,
            query.columns,
            with_row_id=self._with_row_id,
@@ -3880,10 +3874,9 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
        req = fts_query._inner.to_query_request()
        blob_auto_row_id = False
        blob_paths: tuple[str, ...] = ()
-        if self._table is not None and supports_blob_auto_row_id(self._table):
+        if self._table is not None:
            schema = await self._table.schema()
            blob_auto_row_id = blob_auto_row_id_for_scan(
-                self._table,
                schema,
                req.select,
                with_row_id=self._with_row_id,
@@ -7,7 +7,7 @@ import json
 import logging
 from concurrent.futures import ThreadPoolExecutor
 import sys
-from typing import Any, Dict, Iterable, List, Optional, Union
+from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Optional, Union
 from urllib.parse import urlparse
 import warnings

@@ -23,6 +23,10 @@ import pyarrow as pa

 from ..common import DATA
 from ..db import DBConnection, LOOP
+from ..job import Job
+
+if TYPE_CHECKING:
+    from .._lancedb import JobDescription, JobInfo
 from ..embeddings import EmbeddingFunctionConfig
 from lance_namespace import (
    LanceNamespace,
@@ -415,6 +419,11 @@ class RemoteDBConnection(DBConnection):

        if namespace_path is None:
            namespace_path = []
+        if storage_options is not None:
+            logging.info(
+                "storage_options is ignored in LanceDb Cloud"
+                " (storage is managed; set storage_options on connect() instead)"
+            )
        if index_cache_size is not None:
            logging.info(
                "index_cache_size is ignored in LanceDb Cloud"
@@ -684,6 +693,47 @@ class RemoteDBConnection(DBConnection):
            )
        )

+    @override
+    def job(self, job_id: str) -> Job:
+        """A [Job][lancedb.job.Job] handle for a server-side job by id.
+
+        The handle is constructed without a server round trip; an unknown id
+        surfaces when the handle is used. Dropping the handle has no effect
+        on the job itself.
+        """
+        return Job(self._conn.job(job_id))
+
+    @override
+    def list_jobs(self) -> List["JobInfo"]:
+        """List server-side jobs across the database's tables."""
+        return LOOP.run(self._conn.list_jobs())
+
+    @override
+    def get_job(self, job_id: str) -> Optional["JobDescription"]:
+        """Describe a single server-side job by id.
+
+        Returns None when the server has no such job.
+        """
+        return LOOP.run(self._conn.get_job(job_id))
+
+    @override
+    def cancel_job(self, job_id: str) -> bool:
+        """Request cancellation of a server-side job by id.
+
+        Returns True if the server accepted the cancellation, False if no
+        such job exists. Cancelling an already-terminal job is a no-op
+        success.
+        """
+        return LOOP.run(self._conn.cancel_job(job_id))
+
+    @override
+    def job_history(self, job_id: Optional[str] = None) -> List[pa.RecordBatch]:
+        """The lifecycle event history of a server-side job, as Arrow batches.
+
+        Lists history across all jobs when `job_id` is None.
+        """
+        return LOOP.run(self._conn.job_history(job_id))
+
    @override
    def namespace_client(self) -> LanceNamespace:
        """Get the equivalent namespace client for this connection.
@@ -20,6 +20,7 @@ from typing import (
 import warnings

 from lancedb import __version__
+from lancedb._blob import BlobFile

 from lancedb._lancedb import (
    AddColumnsResult,
@@ -47,6 +48,7 @@ from lancedb.index import (
    IvfSq,
    LabelList,
 )
+from lancedb.job import Job
 from lancedb.remote.db import LOOP
 from lancedb.table import IndexConfigType, KNOWN_METRICS
 import pyarrow as pa
@@ -540,6 +542,34 @@ class RemoteTable(Table):
            )
        )

+    def create_index_async(
+        self,
+        column: str,
+        *,
+        config: IndexConfigType,
+        replace: Optional[bool] = None,
+        wait_timeout: Optional[timedelta] = None,
+        name: Optional[str] = None,
+        train: bool = True,
+    ) -> Job:
+        """Create an index, returning a handle to the indexing job.
+
+        The job may already be complete when returned; callers must not assume
+        the index exists until :meth:`Job.wait` returns.
+        """
+        return Job(
+            LOOP.run(
+                self._table.create_index_async(
+                    column,
+                    replace=replace,
+                    config=config,
+                    wait_timeout=wait_timeout,
+                    name=name,
+                    train=train,
+                )
+            )
+        )
+
    def _is_legacy_create_index_call(
        self,
        first_arg: str,
@@ -1039,22 +1069,22 @@ class RemoteTable(Table):
        )

    def blob_columns(self) -> list[str]:
-        raise NotImplementedError(
-            "blob_columns() is not yet supported on the LanceDB Cloud"
-        )
+        return LOOP.run(self._table.blob_columns())

-    def fetch_blobs(self, column: str, row_ids) -> pa.LargeBinaryArray:
-        raise NotImplementedError("fetch_blobs() is not supported on LanceDB Cloud")
+    def fetch_blobs(
+        self, column: str, row_ids: Union[list[int], pa.Table]
+    ) -> pa.LargeBinaryArray:
+        return LOOP.run(self._table.fetch_blobs(column, row_ids))

    def fetch_blob_ranges(self, column: str, requests) -> pa.LargeBinaryArray:
        raise NotImplementedError(
            "fetch_blob_ranges() is not supported on LanceDB Cloud"
        )

-    def fetch_blob_files(self, column: str, row_ids):
-        raise NotImplementedError(
-            "fetch_blob_files() is not supported on LanceDB Cloud"
-        )
+    def fetch_blob_files(
+        self, column: str, row_ids: Union[list[int], pa.Table]
+    ) -> "list[Optional[BlobFile]]":
+        return LOOP.run(self._table.fetch_blob_files(column, row_ids))

    def head(self, n=5) -> pa.Table:
        """
@@ -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
@@ -40,6 +40,7 @@ from ._blob import (
 from .types import BlobMode
 from lancedb.arrow import peek_reader
 from lancedb.background_loop import LOOP, embedding_executor
+from lancedb.job import AsyncJob, Job
 from .dependencies import (
    _check_for_hugging_face,
    _check_for_lance,
@@ -107,6 +108,11 @@ def _should_push_down_query_table(
    return namespace_client is not None and "QueryTable" in pushdown_operations


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

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

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

@@ -977,6 +989,24 @@ class Table(ABC):
        """
        raise NotImplementedError

+    def create_index_async(
+        self,
+        column: str,
+        *,
+        config: IndexConfigType,
+        replace: Optional[bool] = None,
+        wait_timeout: Optional[timedelta] = None,
+        name: Optional[str] = None,
+        train: bool = True,
+    ) -> Job:
+        """Create an index, returning a handle to the indexing job.
+
+        Takes the same arguments as :meth:`create_index`. The job may already
+        be complete when returned; callers must not assume the index exists
+        until :meth:`Job.wait` returns.
+        """
+        raise NotImplementedError
+
    def drop_index(self, name: str) -> None:
        """
        Drop an index from the table.
@@ -1574,8 +1604,10 @@ class Table(ABC):
        """Open lazy, seekable :class:`~lancedb._blob.BlobFile` handles.

        Prefer this over :meth:`fetch_blobs` for large payloads. ``row_ids`` is
-        a ``list[int]`` or query ``pyarrow.Table`` with ``_rowid`` (or stashed
-        row-id metadata). Null rows are ``None``. Local tables only.
+        a ``list[int]`` or a query ``pyarrow.Table`` carrying row identity via
+        ``_rowid`` or a ``_lance_row_id`` field on the blob descriptor. Null
+        rows are ``None``. Remote tables require LanceDB Cloud server 0.5.0 or
+        newer.
        """

    @abstractmethod
@@ -2161,11 +2193,15 @@ class LanceTable(Table):
        return self.name

    @classmethod
-    def from_inner(cls, tbl: LanceDBTable):
-        from .db import LanceDBConnection
+    async def from_inner(cls, tbl: LanceDBTable):
+        from .db import AsyncConnection, LanceDBConnection

        async_tbl = AsyncTable(tbl)
-        conn = LanceDBConnection.from_inner(tbl.database())
+        inner_conn = tbl.database()
+        read_consistency_interval = await AsyncConnection(
+            inner_conn
+        ).get_read_consistency_interval()
+        conn = LanceDBConnection.from_inner(inner_conn, read_consistency_interval)
        return cls(
            conn,
            async_tbl.name,
@@ -2469,13 +2505,7 @@ class LanceTable(Table):
        return LOOP.run(self._table.count_rows(filter))

    def __repr__(self) -> str:
-        val = f"{self.__class__.__name__}(name={self.name!r}"
-        if self._conn.read_consistency_interval is not None:
-            val += ", read_consistency_interval={!r}".format(
-                self._conn.read_consistency_interval
-            )
-        val += f", _conn={self._conn!r})"
-        return val
+        return f"{self.__class__.__name__}(name={self.name!r}, _conn={self._conn!r})"

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

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

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

    # New unified API overload
@@ -2784,6 +2821,34 @@ class LanceTable(Table):
            )
        )

+    def create_index_async(
+        self,
+        column: str,
+        *,
+        config: IndexConfigType,
+        replace: Optional[bool] = None,
+        wait_timeout: Optional[timedelta] = None,
+        name: Optional[str] = None,
+        train: bool = True,
+    ) -> Job:
+        """Create an index, returning a handle to the indexing job.
+
+        The job may already be complete when returned; callers must not assume
+        the index exists until :meth:`Job.wait` returns.
+        """
+        return Job(
+            LOOP.run(
+                self._table.create_index_async(
+                    column,
+                    replace=replace,
+                    config=config,
+                    wait_timeout=wait_timeout,
+                    name=name,
+                    train=train,
+                )
+            )
+        )
+
    def _is_legacy_create_index_call(
        self,
        first_arg: str,
@@ -3911,6 +3976,28 @@ class LanceTable(Table):
        [`AsyncTable.get_lsm_write_spec`][lancedb.AsyncTable.get_lsm_write_spec]."""
        return LOOP.run(self._table.get_lsm_write_spec())

+    def checkpoint_lsm(self) -> None:
+        """Synchronous version of
+        [`AsyncTable.checkpoint_lsm`][lancedb.AsyncTable.checkpoint_lsm]."""
+        return LOOP.run(self._table.checkpoint_lsm())
+
+    def flush_lsm(self) -> None:
+        """Synchronous version of
+        [`AsyncTable.flush_lsm`][lancedb.AsyncTable.flush_lsm]."""
+        return LOOP.run(self._table.flush_lsm())
+
+    def compact_lsm(self) -> None:
+        """Synchronous version of
+        [`AsyncTable.compact_lsm`][lancedb.AsyncTable.compact_lsm]."""
+        return LOOP.run(self._table.compact_lsm())
+
+    def get_lsm_stats(self, *, include_generation_rows: bool = False) -> Optional[dict]:
+        """Synchronous version of
+        [`AsyncTable.get_lsm_stats`][lancedb.AsyncTable.get_lsm_stats]."""
+        return LOOP.run(
+            self._table.get_lsm_stats(include_generation_rows=include_generation_rows)
+        )
+
    def close_lsm_writers(self) -> None:
        """Close cached MemWAL shard writers. See
        [`AsyncTable.close_lsm_writers`][lancedb.AsyncTable.close_lsm_writers]."""
@@ -4589,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
@@ -4615,12 +4709,73 @@ class AsyncTable:

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

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

@@ -4871,6 +5026,46 @@ class AsyncTable:
                )
            raise e

+    async def create_index_async(
+        self,
+        column: str,
+        *,
+        replace: Optional[bool] = None,
+        config: Optional[
+            Union[
+                IvfFlat,
+                IvfPq,
+                IvfRq,
+                HnswPq,
+                HnswSq,
+                HnswFlat,
+                BTree,
+                Bitmap,
+                LabelList,
+                Fm,
+                FTS,
+            ]
+        ] = None,
+        wait_timeout: Optional[timedelta] = None,
+        name: Optional[str] = None,
+        train: bool = True,
+    ) -> AsyncJob:
+        """Create an index, returning a handle to the indexing job.
+
+        Takes the same arguments as :meth:`create_index`. The job may already
+        be complete when returned; callers must not assume the index exists
+        until :meth:`AsyncJob.wait` resolves.
+        """
+        job = await self._inner.create_index_async(
+            column,
+            index=config,
+            replace=replace,
+            wait_timeout=wait_timeout,
+            name=name,
+            train=train,
+        )
+        return AsyncJob(job)
+
    async def drop_index(self, name: str) -> None:
        """
        Drop an index from the table.
@@ -6146,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
@@ -395,6 +395,11 @@ def _(value: dict):
    )


+@value_to_sql.register(pa.Scalar)
+def _(value: pa.Scalar):
+    return value_to_sql(value.as_py())
+
+
@value_to_sql.register(np.ndarray)
 def _(value: np.ndarray):
    return value_to_sql(value.tolist())
@@ -226,13 +226,13 @@ def test_fetch_blob_ranges_validates_requests():
    table = _blob_table("range_validation", [{"id": 1, "image": b"abc"}])
    row_id = _row_ids_by_id(table)[1]

-    with pytest.raises(RuntimeError, match="exceeds blob size"):
+    with pytest.raises(ValueError, match="exceeds blob size"):
        table.fetch_blob_ranges("image", [(row_id, 2, 2)])

-    with pytest.raises(RuntimeError, match="offset \\+ length overflowed"):
+    with pytest.raises(ValueError, match="offset \\+ length overflowed"):
        table.fetch_blob_ranges("image", [(row_id, 2**64 - 1, 1)])

-    with pytest.raises(ValueError, match="row ids"):
+    with pytest.raises(ValueError, match="row IDs"):
        table.fetch_blob_ranges("image", [(2**64 - 1, 0, 1)])


@@ -2,9 +2,11 @@
 # SPDX-FileCopyrightText: Copyright The LanceDB Authors


+import inspect
 import re
 import sys
 from datetime import timedelta
+from importlib import resources
 import os
 from types import SimpleNamespace

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


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

@@ -62,6 +68,70 @@ def test_basic(tmp_path):
    assert db.open_table("test").name == db["test"].name


+def test_sync_debugger_inspection_does_not_use_background_loop(tmp_path, monkeypatch):
+    from lancedb.background_loop import LOOP
+
+    db = lancedb.connect(tmp_path)
+    table = db.create_table("test", data=[{"id": 1}])
+
+    def fail_run(*args, **kwargs):
+        raise AssertionError("debugger inspection should not use the background loop")
+
+    monkeypatch.setattr(LOOP, "run", fail_run)
+
+    # Debuggers enumerate and evaluate every exposed attribute when expanding a
+    # variable. This must remain safe while their breakpoint suspends LOOP's thread.
+    members = dict(inspect.getmembers(db))
+
+    assert members["uri"] == str(tmp_path)
+    assert members["read_consistency_interval"] is None
+    assert repr(db) == f"LanceDBConnection(uri={str(tmp_path)!r})"
+    assert repr(table) == f"LanceTable(name='test', _conn={db!r})"
+
+
+def test_connect_preserves_file_uri_authority(monkeypatch):
+    uri = "file://server/share/database"
+    received = []
+
+    async def fake_connect(passed_uri, *_args):
+        received.append(passed_uri)
+        return SimpleNamespace(uri=passed_uri)
+
+    monkeypatch.setattr("lancedb.db.lancedb_connect", fake_connect)
+    db = lancedb.connect(uri)
+
+    assert received == [uri]
+    assert db.uri == uri
+
+
+def test_connect_file_uri_lifecycle(tmp_path):
+    uri = (tmp_path / "sync").as_uri()
+    db = lancedb.connect(uri)
+
+    db.create_table("test", data=[{"id": 1}])
+    assert db.table_names() == ["test"]
+    assert db.open_table("test").count_rows() == 1
+    db.drop_table("test")
+    assert db.table_names() == []
+
+
+def test_read_consistency_interval_does_not_use_background_loop(tmp_path, monkeypatch):
+    from lancedb.background_loop import LOOP
+    from lancedb.db import LanceDBConnection
+
+    consistency_interval = timedelta(seconds=5)
+    db = lancedb.connect(tmp_path, read_consistency_interval=consistency_interval)
+    db_from_inner = LanceDBConnection.from_inner(db._inner, consistency_interval)
+
+    def fail_run(*args, **kwargs):
+        raise AssertionError("properties should not use the Python background loop")
+
+    monkeypatch.setattr(LOOP, "run", fail_run)
+
+    assert db.read_consistency_interval == consistency_interval
+    assert db_from_inner.read_consistency_interval == consistency_interval
+
+
 def test_ingest_pd(tmp_path):
    db = lancedb.connect(tmp_path)

@@ -361,6 +431,35 @@ async def test_connect(tmp_path):
    assert str(db) == f"ListingDatabase(uri={tmp_path}, read_consistency_interval=5s)"


+@pytest.mark.asyncio
+async def test_connect_async_preserves_file_uri_authority(monkeypatch):
+    uri = "file://server/share/database"
+    received = []
+
+    async def fake_connect(passed_uri, *_args):
+        received.append(passed_uri)
+        return SimpleNamespace(uri=passed_uri)
+
+    monkeypatch.setattr(lancedb, "lancedb_connect", fake_connect)
+    db = await lancedb.connect_async(uri)
+
+    assert received == [uri]
+    assert db.uri == uri
+
+
+@pytest.mark.asyncio
+async def test_connect_async_file_uri_lifecycle(tmp_path):
+    uri = (tmp_path / "async").as_uri()
+    db = await lancedb.connect_async(uri)
+
+    await db.create_table("test", data=[{"id": 1}])
+    assert await db.table_names() == ["test"]
+    table = await db.open_table("test")
+    assert await table.count_rows() == 1
+    await db.drop_table("test")
+    assert await db.table_names() == []
+
+
@pytest.mark.asyncio
 async def test_close(mem_db_async: lancedb.AsyncConnection):
    assert mem_db_async.is_open()
@@ -1136,6 +1235,40 @@ def test_clone_table_deep_clone_fails(tmp_path):
        db.clone_table("cloned", source_uri, is_shallow=False)


+class _UnsupportedNamespaceConfig:
+    async def namespace_client_config(self):
+        raise RuntimeError("UNC namespace client export is not supported")
+
+
+def test_sync_namespace_client_propagates_export_guard(monkeypatch):
+    from lancedb.db import AsyncConnection, LanceDBConnection
+
+    monkeypatch.setattr(
+        "lancedb.db.namespace_connect",
+        lambda *_args, **_kwargs: pytest.fail("guarded config was reconstructed"),
+    )
+    db = LanceDBConnection.__new__(LanceDBConnection)
+    db._conn = AsyncConnection(_UnsupportedNamespaceConfig())
+    db._cached_namespace_client = None
+
+    with pytest.raises(RuntimeError, match="UNC namespace client export"):
+        db.namespace_client()
+
+
+@pytest.mark.asyncio
+async def test_async_namespace_client_propagates_export_guard(monkeypatch):
+    from lancedb.db import AsyncConnection
+
+    monkeypatch.setattr(
+        "lancedb.db.namespace_connect",
+        lambda *_args, **_kwargs: pytest.fail("guarded config was reconstructed"),
+    )
+    db = AsyncConnection(_UnsupportedNamespaceConfig())
+
+    with pytest.raises(RuntimeError, match="UNC namespace client export"):
+        await db.namespace_client()
+
+
@pytest.mark.skipif(sys.platform == "win32", reason="Namespace client issues")
 def test_namespace_client_native_storage(tmp_path):
    """Test namespace_client() returns DirectoryNamespace for native storage."""
@@ -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(
@@ -64,6 +64,23 @@ def test_embedding_function(tmp_path):
    assert np.allclose(actual, expected)


+def test_instructor_ndims_uses_instruction():
+    instructor = get_registry().get("instructor").create()
+    model = MagicMock()
+    model.encode.return_value = np.zeros((1, 384))
+
+    with patch.object(type(instructor), "get_model", return_value=model):
+        assert instructor.ndims() == 384
+
+    model.encode.assert_called_once_with(
+        [[instructor.source_instruction, "foo"]],
+        batch_size=instructor.batch_size,
+        show_progress_bar=instructor.show_progress_bar,
+        normalize_embeddings=instructor.normalize_embeddings,
+        device=instructor.device,
+    )
+
+
 def test_embedding_function_variables():
    @register("variable-testing")
    class VariableTestingFunction(TextEmbeddingFunction):
@@ -115,34 +132,16 @@ def test_embedding_function_variables():
    assert func.safe_model_dump()["secret_key"] == "$var:secret"


-def test_parse_functions_with_variables():
-    @register("variable-parsing-test")
-    class VariableParsingFunction(TextEmbeddingFunction):
-        api_key: str
-        base_url: Optional[str] = None
-
-        @staticmethod
-        def sensitive_keys():
-            return ["api_key"]
-
-        def ndims(self):
-            return 10
-
-        def generate_embeddings(self, texts):
-            # Mock implementation that just returns random embeddings
-            # In real usage, this would use the api_key to call an API
-            return [np.random.rand(self.ndims()).tolist() for _ in texts]
-
+def test_openai_variables_survive_metadata_round_trip():
    registry = EmbeddingFunctionRegistry.get_instance()

    registry.set_var("test_api_key", "sk-test-key-12345")
-    registry.set_var("test_base_url", "https://api.example.com")

    conf = EmbeddingFunctionConfig(
        source_column="text",
        vector_column="vector",
-        function=registry.get("variable-parsing-test").create(
-            api_key="$var:test_api_key", base_url="$var:test_base_url"
+        function=registry.get("openai").create(
+            api_key="$var:test_api_key", base_url="https://api.example.com"
        ),
    )

@@ -150,7 +149,10 @@ def test_parse_functions_with_variables():

    # Create a mock arrow table with the metadata
    schema = pa.schema(
-        [pa.field("text", pa.string()), pa.field("vector", pa.list_(pa.float32(), 10))]
+        [
+            pa.field("text", pa.string()),
+            pa.field("vector", pa.list_(pa.float32(), 1536)),
+        ]
    )
    table = pa.table({"text": [], "vector": []}, schema=schema)
    table = table.replace_schema_metadata(metadata)
@@ -164,13 +166,15 @@ def test_parse_functions_with_variables():

    assert parsed_func.api_key == "sk-test-key-12345"
    assert parsed_func.base_url == "https://api.example.com"
-
-    embeddings = parsed_func.generate_embeddings(["test text"])
-    assert len(embeddings) == 1
-    assert len(embeddings[0]) == 10
-
    assert parsed_func.safe_model_dump()["api_key"] == "$var:test_api_key"

+    with patch("lancedb.embeddings.openai.attempt_import_or_raise") as import_openai:
+        parsed_func._openai_client
+
+    import_openai.return_value.OpenAI.assert_called_once_with(
+        api_key="sk-test-key-12345", base_url="https://api.example.com"
+    )
+

 def test_embedding_with_bad_results(tmp_path):
    @register("null-embedding")
@@ -627,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
@@ -12,7 +12,7 @@ import pyarrow.compute as pc
 import pytest
 import pytest_asyncio

-from lancedb.index import FTS
+from lancedb.index import BTree, FTS, IvfPq
 from lancedb.table import AsyncTable, Table


@@ -99,6 +99,86 @@ async def test_async_hybrid_query_filters(table: AsyncTable):
    assert result["text"].to_pylist() == ["cat", "b"]


+@pytest.mark.asyncio
+async def test_hybrid_query_with_stale_fixed_size_binary_prefilter(
+    tmpdir_factory,
+):
+    tmp_path = str(tmpdir_factory.mktemp("stale_scalar_prefilter"))
+    db = await lancedb.connect_async(tmp_path)
+
+    def fixed_size_binary(value: int) -> bytes:
+        return value.to_bytes(16, byteorder="big")
+
+    num_rows = 1000
+    data = pa.table(
+        {
+            "space_id": pa.array(
+                [fixed_size_binary(i) for i in range(num_rows)],
+                type=pa.binary(16),
+            ),
+            "text": ["book"] * num_rows,
+            "vector": pa.array(
+                [[float(i), float(i)] for i in range(num_rows)],
+                type=pa.list_(pa.float32(), 2),
+            ),
+        }
+    )
+    table = await db.create_table("test", data)
+    await table.create_index(
+        "vector", config=IvfPq(num_partitions=4, num_sub_vectors=2)
+    )
+    await table.create_index("space_id", config=BTree())
+    await table.create_index("text", config=FTS(with_position=False))
+
+    # Advance the search indices without advancing the scalar index. This is the
+    # state that previously let hybrid search use an incomplete scalar prefilter.
+    await table.add(data)
+    lance_dataset = await table.to_lance()
+    lance_dataset.optimize.optimize_indices(index_names=["vector_idx", "text_idx"])
+    await table.checkout_latest()
+
+    scalar_stats = await table.index_stats("space_id_idx")
+    assert scalar_stats is not None
+    assert scalar_stats.num_indexed_rows == num_rows
+    assert scalar_stats.num_unindexed_rows == num_rows
+
+    for index_name in ["vector_idx", "text_idx"]:
+        search_stats = await table.index_stats(index_name)
+        assert search_stats is not None
+        assert search_stats.num_indexed_rows == num_rows * 2
+        assert search_stats.num_unindexed_rows == 0
+
+    matching_ids = [5, 10, 15, 20, 25, 30]
+    literals = [
+        f"arrow_cast(0x{fixed_size_binary(i).hex()}, 'FixedSizeBinary(16)')"
+        for i in matching_ids
+    ]
+    predicate = f"space_id IN ({', '.join(literals)})"
+    expected_ids = sorted(fixed_size_binary(i) for i in matching_ids for _ in range(2))
+
+    vector_query = (
+        table.query().where(predicate).nearest_to([5.0, 5.0]).limit(num_rows * 2)
+    )
+    vector_results = await vector_query.to_arrow()
+    assert sorted(vector_results["space_id"].to_pylist()) == expected_ids
+
+    fts_query = (
+        table.query().where(predicate).nearest_to_text("book").limit(num_rows * 2)
+    )
+    fts_results = await fts_query.to_arrow()
+    assert sorted(fts_results["space_id"].to_pylist()) == expected_ids
+
+    hybrid_results = await (
+        table.query()
+        .where(predicate)
+        .nearest_to([5.0, 5.0])
+        .nearest_to_text("book")
+        .limit(num_rows * 2)
+        .to_arrow()
+    )
+    assert sorted(hybrid_results["space_id"].to_pylist()) == expected_ids
+
+
@pytest.mark.asyncio
 async def test_async_hybrid_query_default_limit(table: AsyncTable):
    # add 10 new rows
@@ -123,6 +203,19 @@ async def test_async_hybrid_query_default_limit(table: AsyncTable):
    assert texts.count("a") == 1


+def test_hybrid_query_minimum_nprobes_zero_raises(sync_table: Table):
+    # minimum_nprobes(0) must raise the same validation error a plain vector
+    # query raises, not silently no-op because 0 is falsy.
+    with pytest.raises(ValueError, match="minimum_nprobes must be greater than 0"):
+        (
+            sync_table.search(query_type="hybrid")
+            .vector([0.0, 0.4])
+            .text("dog")
+            .minimum_nprobes(0)
+            .to_arrow()
+        )
+
+
 def test_hybrid_query_distance_range(sync_table: Table):
    reranker = RRFReranker(return_score="all")
    result = (
@@ -0,0 +1,33 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+import re
+import shutil
+import subprocess
+import sys
+
+import lancedb._lancedb as _lancedb
+import pytest
+
+
+@pytest.mark.skipif(sys.platform != "linux", reason="ldd is Linux-specific")
+def test_native_extension_does_not_link_openssl():
+    """OpenSSL-linked wheels abort when imported on RHEL hosts in FIPS mode."""
+    ldd = shutil.which("ldd")
+    if ldd is None:
+        pytest.skip("ldd is not installed")
+
+    result = subprocess.run(
+        [ldd, _lancedb.__file__],
+        check=True,
+        capture_output=True,
+        text=True,
+    )
+    openssl_libraries = re.findall(
+        r"^\s*(lib(?:crypto|ssl)\S*)\s+=>", result.stdout, flags=re.MULTILINE
+    )
+
+    assert not openssl_libraries, (
+        "the LanceDB native extension must use rustls instead of linking OpenSSL: "
+        f"{openssl_libraries}"
+    )
@@ -84,6 +84,15 @@ async def binary_table(db_async):
    )


+@pytest.mark.asyncio
+async def test_create_index_async_returns_done_job(some_table: AsyncTable):
+    job = await some_table.create_index_async("id", config=BTree())
+    assert job.id is None
+    await job.wait()
+    assert len(await some_table.list_indices()) == 1
+    await job.cancel()
+
+
@pytest.mark.asyncio
 async def test_create_scalar_index(some_table: AsyncTable):
    # Can create
@@ -363,6 +372,31 @@ async def test_create_vector_index(some_table: AsyncTable):
    assert stats.num_indices == 1


+@pytest.mark.asyncio
+async def test_create_ivf_index_reports_unsplittable_partitions(db_async):
+    dim = 8
+    num_partitions = 300  # More than 256 selects hierarchical k-means.
+    base_vectors = [[float(row == column) for column in range(dim)] for row in range(5)]
+    vectors = pa.array(base_vectors * 200, pa.list_(pa.float32(), dim))
+    table = await db_async.create_table(
+        "unsplittable_partitions",
+        pa.table({"vector": vectors}),
+    )
+
+    error_pattern = (
+        rf"Cannot create {num_partitions} IVF partitions: k-means could only form"
+    )
+    with pytest.raises(RuntimeError, match=error_pattern):
+        await table.create_index(
+            "vector",
+            config=IvfFlat(
+                distance_type="dot",
+                num_partitions=num_partitions,
+                max_iterations=10,
+            ),
+        )
+
+
@pytest.mark.asyncio
 async def test_create_4bit_ivfpq_index(some_table: AsyncTable):
    # Can create
@@ -83,7 +83,9 @@ def test_lsm_write_spec_repr():
    assert s.spec_type == "bucket"
    assert s.column == "id"
    assert s.num_buckets == 4
-    assert s.maintained_indexes == []
+    # A fresh spec defers its maintained set to install time.
+    assert s.maintained_indexes is None
+    assert s.with_maintained_indexes([]).maintained_indexes == []
    assert "bucket" in repr(s)
    assert "id" in repr(s)
    assert "4" in repr(s)
@@ -169,18 +171,23 @@ def test_get_lsm_write_spec(tmp_path):
    table.unset_lsm_write_spec()
    assert table.get_lsm_write_spec() is None

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

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


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

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

@@ -60,6 +60,11 @@ class _NamespaceClient:
        return _ipc_file()


+class _UnsupportedNamespaceConfig:
+    async def namespace_client_config(self):
+        raise RuntimeError("UNC namespace client export is not supported")
+
+
 def _namespace_lance_table(namespace_client: _NamespaceClient) -> LanceTable:
    table = LanceTable.__new__(LanceTable)
    table._table = _FailingSyncInner()
@@ -138,6 +143,24 @@ class TestNamespaceConnection:
        db.drop_namespace(["test_ns"])
        assert "test_ns" not in db.list_namespaces().namespaces

+    def test_sync_namespace_client_propagates_export_guard(self, monkeypatch):
+        from lancedb.db import AsyncConnection
+
+        monkeypatch.setattr(
+            "lancedb.namespace.namespace_connect",
+            lambda *_args, **_kwargs: pytest.fail("guarded config was reconstructed"),
+        )
+        db = lancedb.LanceNamespaceDBConnection.__new__(
+            lancedb.LanceNamespaceDBConnection
+        )
+        db._namespace_client = None
+        db._namespace_client_impl = "dir"
+        db._namespace_client_properties = {"root": "file://server/share/database"}
+        db._inner = AsyncConnection(_UnsupportedNamespaceConfig())
+
+        with pytest.raises(RuntimeError, match="UNC namespace client export"):
+            db.namespace_client()
+
    def test_create_table_through_namespace(self):
        """Test creating a table through namespace."""
        db = lancedb.connect_namespace("dir", {"root": self.temp_dir})
@@ -639,6 +662,24 @@ class TestAsyncNamespaceConnection:
        await db.drop_namespace(["test_ns"])
        assert "test_ns" not in (await db.list_namespaces()).namespaces

+    async def test_async_namespace_client_propagates_export_guard(self, monkeypatch):
+        from lancedb.db import AsyncConnection
+
+        monkeypatch.setattr(
+            "lancedb.namespace.namespace_connect",
+            lambda *_args, **_kwargs: pytest.fail("guarded config was reconstructed"),
+        )
+        db = lancedb.AsyncLanceNamespaceDBConnection.__new__(
+            lancedb.AsyncLanceNamespaceDBConnection
+        )
+        db._namespace_client = None
+        db._namespace_client_impl = "dir"
+        db._namespace_client_properties = {"root": "file://server/share/database"}
+        db._inner = AsyncConnection(_UnsupportedNamespaceConfig())
+
+        with pytest.raises(RuntimeError, match="UNC namespace client export"):
+            await db.namespace_client()
+
    async def test_async_namespace_client_is_lazy(self):
        """namespace_client() should still return the backing client on demand."""
        pytest.importorskip("lance")
@@ -18,6 +18,7 @@ Tests verify:
 """

 import copy
+import os
 import shutil
 import sys
 import tempfile
@@ -239,7 +240,7 @@ def create_tracking_namespace(

    dir_props = {f"storage.{k}": v for k, v in storage_options_with_refresh.items()}

-    if bucket_name.startswith("/") or bucket_name.startswith("file://"):
+    if os.path.isabs(bucket_name) or bucket_name.startswith("file://"):
        dir_props["root"] = f"{bucket_name}/namespace_root"
    else:
        dir_props["root"] = f"s3://{bucket_name}/namespace_root"
@@ -767,3 +768,70 @@ def test_namespace_with_schema_only(s3_bucket: str, use_custom: bool):

    # Verify data was added
    assert table.count_rows() == 2
+
+
+@pytest.mark.parametrize("use_custom", [False, True], ids=["DirectoryNS", "CustomNS"])
+def test_namespace_exists(use_custom: bool):
+    """
+    Test namespace_exists returns True for existing and False for non-existent.
+    """
+    temp_dir = tempfile.mkdtemp()
+    try:
+        ns_client, _ = create_tracking_namespace(
+            bucket_name=temp_dir,
+            storage_options={},
+            credential_expires_in_seconds=3600,
+            use_custom=use_custom,
+        )
+        db = LanceNamespaceDBConnection(ns_client)
+
+        namespace_name = f"test_ns_{uuid.uuid4().hex[:8]}"
+        db.create_namespace([namespace_name])
+
+        # Existing namespace should return True
+        assert db.namespace_exists(namespace_id=[namespace_name]) is True
+
+        # Non-existent namespace should return False
+        assert db.namespace_exists(namespace_id=["nonexistent_ns"]) is False
+    finally:
+        shutil.rmtree(temp_dir, ignore_errors=True)
+
+
+@pytest.mark.parametrize("use_custom", [False, True], ids=["DirectoryNS", "CustomNS"])
+def test_table_exists(use_custom: bool):
+    """
+    Test table_exists returns True for existing table and False for non-existent.
+    """
+    temp_dir = tempfile.mkdtemp()
+    try:
+        ns_client, _ = create_tracking_namespace(
+            bucket_name=temp_dir,
+            storage_options={},
+            credential_expires_in_seconds=3600,
+            use_custom=use_custom,
+        )
+        db = LanceNamespaceDBConnection(ns_client)
+
+        namespace_name = f"test_ns_{uuid.uuid4().hex[:8]}"
+        db.create_namespace([namespace_name])
+
+        table_name = f"test_table_{uuid.uuid4().hex}"
+        namespace_path = [namespace_name]
+        schema = pa.schema(
+            [
+                pa.field("id", pa.int64()),
+                pa.field("vector", pa.list_(pa.float32(), 2)),
+                pa.field("text", pa.string()),
+            ]
+        )
+
+        db.create_table(table_name, schema=schema, namespace_path=namespace_path)
+
+        # Existing table should return True
+        table_id = namespace_path + [table_name]
+        assert db.table_exists(table_id=table_id) is True
+
+        # Non-existent table should return False
+        assert db.table_exists(table_id=namespace_path + ["nonexistent_table"]) is False
+    finally:
+        shutil.rmtree(temp_dir, ignore_errors=True)
@@ -0,0 +1,42 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+import importlib
+import re
+import sys
+from pathlib import Path
+
+import pytest
+
+
+def test_pyo3_abi_matches_minimum_supported_python():
+    project_dir = Path(__file__).parents[2]
+    pyproject = (project_dir / "pyproject.toml").read_text()
+    cargo_manifest = (project_dir / "Cargo.toml").read_text()
+
+    minimum_python = re.search(
+        r'^requires-python\s*=\s*">=(\d+)\.(\d+)"$', pyproject, re.MULTILINE
+    )
+    assert minimum_python is not None
+
+    major, minor = minimum_python.groups()
+    expected_abi = f"abi3-py{major}{minor}"
+    configured_abis = re.findall(r'"(abi3-py\d+)"', cargo_manifest)
+
+    assert configured_abis == [expected_abi, expected_abi], (
+        "the pyo3 runtime and build ABI features must both match requires-python"
+    )
+
+
+@pytest.mark.skipif(sys.platform != "win32", reason="Windows wheel regression test")
+def test_windows_wheel_tag_and_native_import():
+    project_dir = Path(__file__).parents[2]
+    wheels = list((project_dir.parent / "target" / "wheels").glob("lancedb-*.whl"))
+    if not wheels:
+        pytest.skip("no wheel artifact is available in this development environment")
+
+    assert len(wheels) == 1
+    assert wheels[0].name.endswith("-cp310-abi3-win_amd64.whl")
+
+    native_module = importlib.import_module("lancedb._lancedb")
+    assert Path(native_module.__file__).suffix == ".pyd"
@@ -6,6 +6,7 @@ import math
 import pytest

 from lancedb import DBConnection, Table, connect
+from lancedb.background_loop import LOOP
 from lancedb.permutation import Permutation, Permutations, permutation_builder


@@ -31,6 +32,25 @@ def test_split_random_ratios(mem_db):
    assert 65 <= split_1_count <= 75  # ~70% ± tolerance


+def test_execute_does_not_reenter_background_loop(tmp_path, monkeypatch):
+    import threading
+
+    db = connect(tmp_path)
+    tbl = db.create_table("test_table", pa.table({"x": range(10)}))
+    original_run = LOOP.run
+
+    def fail_on_reentry(future):
+        assert threading.current_thread() is not LOOP.thread
+        return original_run(future)
+
+    monkeypatch.setattr(LOOP, "run", fail_on_reentry)
+
+    permutation_tbl = permutation_builder(tbl).execute()
+
+    assert permutation_tbl.count_rows() == 10
+    assert permutation_tbl._conn.read_consistency_interval is None
+
+
 def test_split_random_counts(mem_db):
    """Test random splitting with absolute counts."""
    tbl = mem_db.create_table(
@@ -415,6 +415,17 @@ def test_nullable_vector():
    assert schema == pa.schema([pa.field("vec", pa.list_(pa.float32(), 16), True)])


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


+def test_query_multiple_vectors(table):
+    results = table.search([np.array([1, 2]), np.array([4, 5])]).limit(1).to_list()
+
+    assert len(results) == 2
+    results_by_query = {result["query_index"]: result for result in results}
+    assert results_by_query[0]["id"] == 1
+    assert results_by_query[1]["id"] == 2
+
+
 def test_with_row_id(table: lancedb.table.Table):
    rs = table.search().with_row_id(True).to_arrow()
    assert "_rowid" in rs.column_names
@@ -35,6 +35,12 @@ def make_mock_http_handler(handler):
    return MockLanceDBHandler


+@pytest.mark.parametrize("db_name", ["a" * 64, "invalid..database"])
+def test_connect_rejects_invalid_cloud_dns_hostname(db_name):
+    with pytest.raises(ValueError, match="DNS labels must contain 1 to 63 bytes"):
+        lancedb.connect(f"db://{db_name}", api_key="fake")
+
+
@contextlib.contextmanager
 def mock_lancedb_connection(handler):
    with http.server.HTTPServer(
@@ -812,6 +818,121 @@ def test_table_create_indices():
        table.drop_index("custom_fts_idx")


+def test_remote_create_index_async_returns_job():
+    from lancedb.index import BTree
+
+    describe_calls = []
+
+    def handler(request):
+        content_len = int(request.headers.get("Content-Length", 0))
+        body = request.rfile.read(content_len) if content_len > 0 else b""
+        if request.path == "/v1/table/test/create_index/":
+            request.send_response(200)
+            request.send_header("Content-Type", "application/json")
+            request.end_headers()
+            request.wfile.write(b'{"job_id": "job-1"}')
+        elif request.path == "/v1/jobs/describe":
+            assert json.loads(body)["job_id"] == "job-1"
+            describe_calls.append(1)
+            state = "IN_PROGRESS" if len(describe_calls) == 1 else "DONE"
+            request.send_response(200)
+            request.send_header("Content-Type", "application/json")
+            request.end_headers()
+            request.wfile.write(
+                json.dumps(dict(job_id="job-1", job_state=state)).encode()
+            )
+        elif request.path == "/v1/jobs/cancel":
+            assert json.loads(body)["job_id"] == "job-1"
+            request.send_response(200)
+            request.send_header("Content-Type", "application/json")
+            request.end_headers()
+            request.wfile.write(b"{}")
+        elif request.path == "/v1/table/test/create/?mode=create":
+            request.send_response(200)
+            request.send_header("Content-Type", "application/json")
+            request.end_headers()
+            request.wfile.write(b"{}")
+        elif request.path == "/v1/table/test/describe/":
+            request.send_response(200)
+            request.send_header("Content-Type", "application/json")
+            request.end_headers()
+            request.wfile.write(
+                json.dumps(
+                    dict(
+                        version=1,
+                        schema=dict(
+                            fields=[
+                                dict(name="id", type={"type": "int64"}, nullable=False),
+                            ]
+                        ),
+                    )
+                ).encode()
+            )
+        else:
+            request.send_response(404)
+            request.end_headers()
+
+    with mock_lancedb_connection(handler) as db:
+        table = db.create_table("test", [{"id": 1}])
+        job = table.create_index_async("id", config=BTree())
+        assert job.id == "job-1"
+        job.wait(timeout=timedelta(seconds=30))
+        assert len(describe_calls) == 2
+        job.cancel()
+
+
+def test_remote_job_wait_raises_on_failure():
+    from lancedb.exceptions import JobFailedError
+    from lancedb.index import BTree
+
+    def handler(request):
+        content_len = int(request.headers.get("Content-Length", 0))
+        body = request.rfile.read(content_len) if content_len > 0 else b""
+        if request.path == "/v1/table/test/create_index/":
+            request.send_response(200)
+            request.send_header("Content-Type", "application/json")
+            request.end_headers()
+            request.wfile.write(b'{"job_id": "job-2"}')
+        elif request.path == "/v1/jobs/describe":
+            assert json.loads(body)["job_id"] == "job-2"
+            request.send_response(200)
+            request.send_header("Content-Type", "application/json")
+            request.end_headers()
+            request.wfile.write(
+                json.dumps(dict(job_id="job-2", job_state="FAILED")).encode()
+            )
+        elif request.path == "/v1/table/test/create/?mode=create":
+            request.send_response(200)
+            request.send_header("Content-Type", "application/json")
+            request.end_headers()
+            request.wfile.write(b"{}")
+        elif request.path == "/v1/table/test/describe/":
+            request.send_response(200)
+            request.send_header("Content-Type", "application/json")
+            request.end_headers()
+            request.wfile.write(
+                json.dumps(
+                    dict(
+                        version=1,
+                        schema=dict(
+                            fields=[
+                                dict(name="id", type={"type": "int64"}, nullable=False),
+                            ]
+                        ),
+                    )
+                ).encode()
+            )
+        else:
+            request.send_response(404)
+            request.end_headers()
+
+    with mock_lancedb_connection(handler) as db:
+        table = db.create_table("test", [{"id": 1}])
+        job = table.create_index_async("id", config=BTree())
+        with pytest.raises(JobFailedError, match="job-2"):
+            job.wait()
+
+
 def test_remote_create_index_new_api():
    received_requests = []

@@ -1020,7 +1141,7 @@ def query_test_table(query_handler, *, server_version=Version("0.1.0")):
            request.send_header("Content-Type", "application/json")
            request.send_header("phalanx-version", str(server_version))
            request.end_headers()
-            request.wfile.write(b"{}")
+            request.wfile.write(b'{"version": 1, "schema": {"fields": []}}')
        elif request.path == "/v1/table/test/query/":
            content_len = int(request.headers.get("Content-Length"))
            body = request.rfile.read(content_len)
@@ -1858,3 +1979,330 @@ def test_inherited_remote_table_reopens_after_fork():
    finally:
        server.shutdown()
        server_thread.join()
+
+
+BLOB_DESCRIBE_RESPONSE = {
+    "table": "test",
+    "version": 1,
+    "schema": {
+        "fields": [
+            {"name": "id", "type": {"type": "int64"}, "nullable": False},
+            {
+                "name": "image",
+                "type": {
+                    "type": "struct",
+                    "fields": [
+                        {
+                            "name": "data",
+                            "type": {"type": "large_binary"},
+                            "nullable": True,
+                        },
+                        {"name": "uri", "type": {"type": "string"}, "nullable": True},
+                    ],
+                },
+                "nullable": True,
+                "metadata": {
+                    "ARROW:extension:name": "lance.blob.v2",
+                    "ARROW:extension:metadata": "",
+                },
+            },
+        ]
+    },
+}
+
+
+def blob_query_response_table():
+    image_field = pa.field(
+        "image",
+        pa.struct(
+            [
+                pa.field("kind", pa.uint8(), nullable=False),
+                pa.field("position", pa.uint64(), nullable=False),
+                pa.field("size", pa.uint64(), nullable=False),
+                pa.field("blob_id", pa.uint32(), nullable=False),
+                pa.field("blob_uri", pa.string(), nullable=False),
+            ]
+        ),
+        metadata={"lance-encoding:blob": "true"},
+    )
+    images = pa.StructArray.from_arrays(
+        [
+            pa.array([1, 0, 0], type=pa.uint8()),
+            pa.array([0, 0, 0], type=pa.uint64()),
+            pa.array([5, 0, 5], type=pa.uint64()),
+            pa.array([1, 0, 2], type=pa.uint32()),
+            pa.array(["", "", ""], type=pa.string()),
+        ],
+        fields=image_field.type,
+        mask=pa.array([False, True, False]),
+    )
+    return pa.Table.from_arrays(
+        [
+            pa.array([1, 2, 3], type=pa.int64()),
+            images,
+            pa.array([10, 20, 30], type=pa.uint64()),
+        ],
+        schema=pa.schema(
+            [
+                pa.field("id", pa.int64(), nullable=False),
+                image_field,
+                pa.field("_rowid", pa.uint64()),
+            ]
+        ),
+    )
+
+
+@contextlib.contextmanager
+def blob_remote_table(*, server_version=Version("0.5.0")):
+    def handler(request):
+        if request.path == "/v1/table/test/describe/":
+            request.send_response(200)
+            request.send_header("Content-Type", "application/json")
+            request.send_header("phalanx-version", str(server_version))
+            request.end_headers()
+            request.wfile.write(json.dumps(BLOB_DESCRIBE_RESPONSE).encode())
+        elif request.path.startswith("/v1/table/test/blob/image/"):
+            path = request.path.partition("?")[0]
+            row_id = int(path.split("/")[-2])
+            payload = {10: b"alpha", 20: None, 30: b"gamma"}[row_id]
+            if payload is None:
+                request.send_response(204)
+                request.end_headers()
+                return
+            byte_range = request.headers["Range"].removeprefix("bytes=")
+            start_text, end_text = byte_range.split("-", maxsplit=1)
+            start = int(start_text)
+            end = int(end_text) if end_text else len(payload) - 1
+            chunk = payload[start : end + 1]
+            request.send_response(206)
+            request.send_header("Content-Range", f"bytes {start}-{end}/{len(payload)}")
+            request.send_header("Content-Length", str(len(chunk)))
+            request.end_headers()
+            request.wfile.write(chunk)
+        elif request.path == "/v1/table/test/query/":
+            content_len = int(request.headers.get("Content-Length", 0))
+            body = json.loads(request.rfile.read(content_len))
+            assert body["columns"] == ["id", "image"]
+            assert body["with_row_id"] is True
+            response_table = blob_query_response_table()
+            request.send_response(200)
+            request.send_header("Content-Type", "application/vnd.apache.arrow.file")
+            request.end_headers()
+            with pa.ipc.new_file(request.wfile, response_table.schema) as writer:
+                writer.write_table(response_table)
+        elif request.path == "/v1/table/test/fetch_blobs/":
+            content_len = int(request.headers.get("Content-Length", 0))
+            body = json.loads(request.rfile.read(content_len))
+            assert body["column"] == "image"
+            assert body["row_ids"] == [10, 20, 30]
+            response_table = pa.table(
+                {"image": pa.array([b"alpha", None, b"gamma"], type=pa.large_binary())}
+            )
+            request.send_response(200)
+            request.send_header("Content-Type", "application/vnd.apache.arrow.stream")
+            request.end_headers()
+            with pa.ipc.new_stream(request.wfile, response_table.schema) as writer:
+                writer.write_table(response_table)
+        else:
+            request.send_response(404)
+            request.end_headers()
+
+    with mock_lancedb_connection(handler) as db:
+        yield db.open_table("test")
+
+
+def test_remote_blob_columns_and_fetch():
+    with blob_remote_table() as table:
+        assert table.blob_columns() == ["image"]
+        blobs = table.fetch_blobs("image", [10, 20, 30])
+        assert blobs.to_pylist() == [b"alpha", None, b"gamma"]
+
+
+def test_remote_blob_files_are_lazy_seekable_handles():
+    with blob_remote_table() as table:
+        files = table.fetch_blob_files("image", [10, 20, 30])
+
+        assert len(files) == 3
+        alpha, null_row, gamma = files
+        assert null_row is None
+        assert alpha is not None
+        assert gamma is not None
+        assert alpha.size() == 5
+        assert alpha.read_range(1, 3) == b"lph"
+        gamma.seek(2)
+        assert gamma.read() == b"mma"
+
+
+def test_remote_blob_fetch_accepts_query_table():
+    hits = pa.table({"_rowid": pa.array([10, 20, 30], type=pa.uint64())})
+
+    with blob_remote_table() as table:
+        blobs = table.fetch_blobs("image", hits)
+
+    assert blobs.to_pylist() == [b"alpha", None, b"gamma"]
+
+
+def test_remote_blob_query_stashes_row_ids_for_fetch():
+    with blob_remote_table() as table:
+        hits = table.search().select(["id", "image"]).limit(3).to_arrow()
+        assert "_rowid" not in hits.column_names
+        assert "_lance_row_id" in hits.schema.field("image").type.names
+        blobs = table.fetch_blobs("image", hits)
+
+    assert blobs.to_pylist() == [b"alpha", None, b"gamma"]
+
+
+def test_remote_blob_query_survives_a_server_that_ignores_the_row_id_request():
+    def handler(request):
+        if request.path == "/v1/table/test/describe/":
+            request.send_response(200)
+            request.send_header("Content-Type", "application/json")
+            request.send_header("phalanx-version", "0.5.0")
+            request.end_headers()
+            request.wfile.write(json.dumps(BLOB_DESCRIBE_RESPONSE).encode())
+        elif request.path == "/v1/table/test/query/":
+            content_len = int(request.headers.get("Content-Length", 0))
+            assert json.loads(request.rfile.read(content_len))["with_row_id"] is True
+            response_table = blob_query_response_table().drop_columns(["_rowid"])
+            request.send_response(200)
+            request.send_header("Content-Type", "application/vnd.apache.arrow.file")
+            request.end_headers()
+            with pa.ipc.new_file(request.wfile, response_table.schema) as writer:
+                writer.write_table(response_table)
+        else:
+            request.send_response(404)
+            request.end_headers()
+
+    with mock_lancedb_connection(handler) as db:
+        table = db.open_table("test")
+        hits = table.search().select(["id", "image"]).limit(3).to_arrow()
+
+        assert hits.column_names == ["id", "image"]
+        assert "_lance_row_id" not in hits.schema.field("image").type.names
+        with pytest.raises(ValueError, match="pass a list of row ids"):
+            table.fetch_blobs("image", hits)
+
+
+def test_remote_blob_byte_apis_not_supported_on_old_server():
+    with blob_remote_table(server_version=Version("0.1.0")) as table:
+        assert table.blob_columns() == ["image"]
+        with pytest.raises(NotImplementedError, match="not supported"):
+            table.fetch_blobs("image", [1])
+        with pytest.raises(NotImplementedError, match="not supported"):
+            table.fetch_blob_files("image", [1])
+
+
+def test_remote_connection_jobs_surface():
+    from lancedb.exceptions import JobFailedError
+
+    schema = pa.schema([("state", pa.string())])
+    batch = pa.record_batch([pa.array(["created", "done"])], schema=schema)
+    sink = pa.BufferOutputStream()
+    with pa.ipc.new_stream(sink, schema) as writer:
+        writer.write_batch(batch)
+    events_body = sink.getvalue().to_pybytes()
+
+    def handler(request):
+        content_len = int(request.headers.get("Content-Length", 0))
+        body = request.rfile.read(content_len) if content_len > 0 else b""
+        payload = json.loads(body) if body else {}
+        if request.path == "/v1/jobs/list":
+            if payload.get("page_token") is None:
+                rsp = dict(
+                    jobs=[
+                        dict(
+                            job_id="job-1",
+                            table="t1",
+                            job_type="create_index",
+                            state="in_progress",
+                            created_at_millis=1000,
+                        )
+                    ],
+                    page_token="next",
+                )
+            else:
+                assert payload["page_token"] == "next"
+                rsp = dict(
+                    jobs=[
+                        dict(
+                            job_id="job-2",
+                            table="t2",
+                            job_type="create_index",
+                            state="succeeded",
+                            created_at_millis=2000,
+                        )
+                    ]
+                )
+            request.send_response(200)
+            request.send_header("Content-Type", "application/json")
+            request.end_headers()
+            request.wfile.write(json.dumps(rsp).encode())
+        elif request.path == "/v1/jobs/describe":
+            if payload["job_id"] != "job-1":
+                request.send_response(404)
+                request.end_headers()
+                return
+            request.send_response(200)
+            request.send_header("Content-Type", "application/json")
+            request.end_headers()
+            request.wfile.write(
+                json.dumps(
+                    dict(
+                        job_id="job-1",
+                        job_type="create_index",
+                        job_state="FAILED",
+                        creation_ms=1000,
+                        spec=dict(column="vec"),
+                        failure=dict(
+                            phase="execute", message="worker died", retryable=True
+                        ),
+                    )
+                ).encode()
+            )
+        elif request.path == "/v1/jobs/cancel":
+            if payload["job_id"] != "job-1":
+                request.send_response(404)
+                request.end_headers()
+                return
+            request.send_response(200)
+            request.send_header("Content-Type", "application/json")
+            request.end_headers()
+            request.wfile.write(b'{"job_id": "job-1"}')
+        elif request.path == "/v1/jobs/query_events":
+            assert payload["job_id"] == "job-1"
+            request.send_response(200)
+            request.send_header("Content-Type", "application/vnd.apache.arrow.stream")
+            request.end_headers()
+            request.wfile.write(events_body)
+        else:
+            request.send_response(404)
+            request.end_headers()
+
+    with mock_lancedb_connection(handler) as db:
+        jobs = db.list_jobs()
+        assert [job.job_id for job in jobs] == ["job-1", "job-2"]
+        assert jobs[0].state == "running"
+        assert jobs[0].table == "t1"
+        assert jobs[1].state == "finished"
+
+        description = db.get_job("job-1")
+        assert description.job_type == "create_index"
+        assert description.state == "failed"
+        assert json.loads(description.spec_json) == {"column": "vec"}
+        assert description.failure.message == "worker died"
+        assert description.failure.retryable is True
+        assert db.get_job("missing") is None
+
+        assert db.cancel_job("job-1") is True
+        assert db.cancel_job("missing") is False
+
+        batches = db.job_history("job-1")
+        assert len(batches) == 1
+        assert batches[0].num_rows == 2
+        assert batches[0].column("state").to_pylist() == ["created", "done"]
+
+        job = db.job("job-1")
+        assert job.id == "job-1"
+        assert job.status() == "failed"
+        with pytest.raises(JobFailedError, match="worker died"):
+            job.wait(timeout=timedelta(seconds=5))
@@ -2,10 +2,14 @@
 # SPDX-FileCopyrightText: Copyright The LanceDB Authors


+import ctypes
+import gc
 import os
 import sys
 import threading
 import warnings
+import weakref
+from concurrent.futures import ThreadPoolExecutor
 from datetime import date, datetime, timedelta
 from time import sleep
 from typing import List
@@ -98,6 +102,30 @@ def test_basic(mem_db: DBConnection):
    assert table.to_arrow() == expected_data


+def test_search_preserves_nulls_from_sliced_arrow_table(mem_db: DBConnection):
+    data = pa.table(
+        {
+            "id": [0, 1, 2, 3, 4],
+            "score_cn": [None, 22, None, 5, 8],
+            "score_mt": [None, 42, None, 5, 8],
+            "vector": [
+                [20, 19, -1, -1],
+                [41, 38, 22, 42],
+                [10, 10, -1, -1],
+                [5, 5, 5, 5],
+                [8, 8, 8, 8],
+            ],
+        }
+    ).slice(1)
+
+    table = mem_db.create_table("sliced_nullable", data=data)
+    result = table.search([41, 38, 22, 42]).limit(1).to_arrow()
+
+    assert result["id"].to_pylist() == [1]
+    assert result["score_cn"].to_pylist() == [22]
+    assert result["score_mt"].to_pylist() == [42]
+
+
 def test_table_to_pandas_default_matches_arrow(tmp_db: DBConnection):
    pd = pytest.importorskip("pandas")
    data = pa.table({"id": [1, 2], "text": ["one", "two"]})
@@ -434,6 +462,38 @@ def test_add(mem_db: DBConnection):
    _add(table, schema)


+def test_add_releases_arrow_buffers_without_gc(mem_db: DBConnection):
+    """Regression test for https://github.com/lancedb/lancedb/issues/2512."""
+    schema = pa.schema([pa.field("x", pa.int64())])
+    table = mem_db.create_table("test_add_releases_arrow_buffers", schema=schema)
+
+    class BufferOwner:
+        def __init__(self, size: int):
+            self.memory = ctypes.create_string_buffer(size)
+
+    owner_refs = []
+    gc_was_enabled = gc.isenabled()
+    gc.disable()
+    try:
+        for _ in range(3):
+            size = 8 * 1024
+            owner = BufferOwner(size)
+            arrow_buffer = pa.foreign_buffer(
+                ctypes.addressof(owner.memory), size, owner
+            )
+            array = pa.Array.from_buffers(pa.int64(), 1024, [None, arrow_buffer])
+            batch = pa.RecordBatch.from_arrays([array], schema=schema)
+            owner_refs.append(weakref.ref(owner))
+
+            table.add(batch)
+            del batch, array, arrow_buffer, owner
+
+        assert all(owner_ref() is None for owner_ref in owner_refs)
+    finally:
+        if gc_was_enabled:
+            gc.enable()
+
+
 def test_add_write_parallelism(mem_db: DBConnection):
    schema = pa.schema([pa.field("id", pa.int64())])
    table = mem_db.create_table("test", schema=schema)
@@ -869,6 +929,7 @@ def test_polars(mem_db: DBConnection):

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

    # make sure filtering isn't broken
@@ -1402,6 +1463,15 @@ async def test_async_open_table_with_branch_version(tmp_path):
    assert await pinned.count_rows() == 4  # writable again


+def test_create_index_async_returns_done_job(mem_db: DBConnection):
+    table = mem_db.create_table("job_test", [{"id": i} for i in range(10)])
+    job = table.create_index_async("id", config=BTree())
+    assert job.id is None
+    job.wait()
+    assert len(table.list_indices()) == 1
+    job.cancel()
+
+
@patch("lancedb.table.AsyncTable.create_index")
 def test_create_index_method(mock_create_index, mem_db: DBConnection):
    table = mem_db.create_table(
@@ -1776,6 +1846,27 @@ def test_add_with_empty_fixed_size_list_drops_bad_rows(mem_db: DBConnection):
    assert np.allclose(data["embedding"].to_pylist()[0], np.array([0.1] * 16))


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


+def test_read_mostly_null_list_v2_2_page_boundary(tmp_path):
+    # Regression test for #3194. This row/value count crosses a v2.2 structural
+    # encoding page boundary where Lance 3.0.0 sliced repetition/definition
+    # levels by row offset and decoded child arrays at different lengths.
+    num_rows = 64_885
+    num_values = 217
+    list_type = pa.list_(pa.float32())
+    source = pa.table(
+        {
+            "id": np.arange(num_rows, dtype=np.int64),
+            "coords": pa.array(
+                [[1.0, 2.0, 3.0, 4.0]] * num_values + [None] * (num_rows - num_values),
+                type=list_type,
+            ),
+        }
+    )
+    db = lancedb.connect(
+        tmp_path,
+        storage_options={"new_table_data_storage_version": "2.2"},
+    )
+    table = db.create_table("test_sparse_nullable_list", data=source)
+
+    result = table.search().select(["id", "coords"]).limit(num_rows).to_arrow()
+
+    assert result.equals(source)
+
+
 def test_add_with_integer_embeddings_preserves_casting(mem_db: DBConnection):
    class Schema(LanceModel):
        text: str
@@ -2100,6 +2218,45 @@ def test_merge(tmp_db: DBConnection, tmp_path):
    table.merge(other_dataset, left_on="id")


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


+def test_concurrent_deletes_are_thread_safe(mem_db: DBConnection):
+    num_workers = 8
+    table = mem_db.create_table(
+        "my_table", data=[{"id": row_id} for row_id in range(num_workers)]
+    )
+    barrier = threading.Barrier(num_workers)
+
+    def delete(row_id: int):
+        barrier.wait()
+        return table.delete(f"id = {row_id}")
+
+    with ThreadPoolExecutor(max_workers=num_workers) as pool:
+        results = list(pool.map(delete, range(num_workers)))
+
+    assert all(result.num_deleted_rows == 1 for result in results)
+    assert sorted(result.version for result in results) == list(
+        range(2, num_workers + 2)
+    )
+    assert table.count_rows() == 0
+
+
 def test_delete_expr(mem_db: DBConnection):
    table = mem_db.create_table(
        "my_table",
@@ -2165,6 +2343,20 @@ def test_update(mem_db: DBConnection):
    assert np.allclose(v, np.array([[1.2, 1.9], [1.1, 1.1]]))


+def test_update_with_arrow_scalar(mem_db: DBConnection):
+    schema = pa.schema({"id": pa.int64(), "vector": pa.list_(pa.float32(), 4)})
+    table = mem_db.create_table("my_table", schema=schema)
+    table.add([{"id": 1, "vector": [1.0, 2.0, 3.0, 4.0]}])
+
+    value = table.search().select(["vector"]).limit(1).to_arrow()["vector"][0]
+    assert isinstance(value, pa.FixedSizeListScalar)
+
+    result = table.update(where="id == 1", values={"vector": value})
+
+    assert result.rows_updated == 1
+    assert table.to_arrow()["vector"].to_pylist() == [[1.0, 2.0, 3.0, 4.0]]
+
+
 def test_update_types(mem_db: DBConnection):
    table = mem_db.create_table(
        "my_table",
@@ -2332,6 +2524,55 @@ def test_merge_insert(mem_db: DBConnection):
        )


+def test_merge_insert_nullable_pandas_into_pydantic_schema(mem_db: DBConnection):
+    # Regression test for https://github.com/lancedb/lancedb/issues/2366
+    pd = pytest.importorskip("pandas")
+
+    class Document(LanceModel):
+        id: int
+        title: str
+        content: str
+
+    table = mem_db.create_table("documents", schema=Document)
+    table.add(
+        pd.DataFrame(
+            {
+                "title": ["Old title", "Unchanged"],
+                "id": [2, 3],
+                "content": ["Old content", "Keep this"],
+            }
+        )
+    )
+
+    # Pandas produces nullable Arrow fields, in an order that differs from the
+    # non-nullable Pydantic schema. This is valid as long as the data has no nulls.
+    new_data = pd.DataFrame(
+        {
+            "title": ["Inserted", "Updated"],
+            "id": [1, 2],
+            "content": ["New row", "New content"],
+        }
+    )
+    result = (
+        table.merge_insert("id")
+        .when_matched_update_all()
+        .when_not_matched_insert_all()
+        .execute(new_data)
+    )
+
+    assert result.num_inserted_rows == 1
+    assert result.num_updated_rows == 1
+    expected = pa.Table.from_pylist(
+        [
+            {"id": 1, "title": "Inserted", "content": "New row"},
+            {"id": 2, "title": "Updated", "content": "New content"},
+            {"id": 3, "title": "Unchanged", "content": "Keep this"},
+        ],
+        schema=Document.to_arrow_schema(),
+    )
+    assert table.to_arrow().sort_by("id") == expected
+
+
 def test_merge_insert_by_source_delete_expr(mem_db: DBConnection):
    table = mem_db.create_table(
        "my_table",
@@ -2355,6 +2596,29 @@ def test_merge_insert_by_source_delete_expr(mem_db: DBConnection):
    assert table.to_arrow().sort_by("a") == expected


+def test_merge_insert_by_source_delete_reconfigure(mem_db: DBConnection):
+    # Calling when_not_matched_by_source_delete() again with no condition must
+    # widen the delete to unconditional, not keep the earlier condition around.
+    table = mem_db.create_table(
+        "my_table",
+        data=pa.table({"a": [1, 2, 3], "b": ["a", "b", "c"]}),
+    )
+    new_data = pa.table({"a": [2, 4], "b": ["x", "z"]})
+
+    merge_insert_res = (
+        table.merge_insert("a")
+        .when_matched_update_all()
+        .when_not_matched_insert_all()
+        .when_not_matched_by_source_delete("a > 2")
+        .when_not_matched_by_source_delete()
+        .execute(new_data)
+    )
+    assert merge_insert_res.num_deleted_rows == 2
+
+    expected = pa.table({"a": [2, 4], "b": ["x", "z"]})
+    assert table.to_arrow().sort_by("a") == expected
+
+
@pytest.mark.asyncio
 async def test_merge_insert_by_source_delete_expr_async(
    mem_db_async: AsyncConnection,
@@ -2409,6 +2673,36 @@ def test_merge_insert_subschema(mem_db: DBConnection, data_format):
    assert table.to_arrow().sort_by("id") == expected


+def test_repeated_partial_merge_insert_with_scalar_index(mem_db: DBConnection):
+    def make_batch(start: int) -> pa.Table:
+        return pa.table(
+            {
+                "id": [f"id-{i:04}" for i in range(start, start + 100)],
+                "category": ["A"] * 100,
+                "value_a": [float(i) for i in range(start, start + 100)],
+                "value_b": [float(i) / 10 for i in range(100)],
+            }
+        )
+
+    table = mem_db.create_table("my_table", data=make_batch(0))
+    table.add(make_batch(100))
+    table.add(make_batch(200))
+    table.create_index("id", config=BTree())
+
+    ids = [f"id-{i:04}" for i in range(100, 200)]
+    for value in (999.0, 888.0):
+        result = (
+            table.merge_insert("id")
+            .when_matched_update_all()
+            .execute(pa.table({"id": ids, "value_a": [value] * 100}))
+        )
+        assert result.num_updated_rows == 100
+
+    actual = table.to_arrow().sort_by("id")
+    assert actual.num_rows == 300
+    assert actual["value_a"].to_pylist()[100:200] == [888.0] * 100
+
+
@pytest.mark.asyncio
 async def test_merge_insert_async(mem_db_async: AsyncConnection):
    data = pa.table({"a": [1, 2, 3], "b": ["a", "b", "c"]})
@@ -2505,15 +2799,40 @@ def test_create_with_embedding_function(mem_db: DBConnection):
    assert actual == expected


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

+    rng = np.random.default_rng(42)
    df = pa.table(
        {
            "text": [f"s-{i}" for i in range(512)],
-            "vector": [np.random.randn(32).astype(np.float16) for _ in range(512)],
+            "vector": [rng.standard_normal(32).astype(np.float16) for _ in range(512)],
        }
    )
    table = mem_db.create_table(
@@ -3087,9 +3406,6 @@ def test_consistency(tmp_path, consistency_interval):

    db2 = lancedb.connect(tmp_path, read_consistency_interval=consistency_interval)
    table2 = db2.open_table("my_table")
-    if consistency_interval is not None:
-        assert "read_consistency_interval=datetime.timedelta(" in repr(db2)
-        assert "read_consistency_interval=datetime.timedelta(" in repr(table2)
    assert table2.version == table.version

    table.add([{"id": 1}])
@@ -3397,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": {
@@ -3415,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
@@ -3438,8 +3762,8 @@ def test_create_table_empty_list_no_schema_error(mem_db: DBConnection):
        mem_db.create_table("test_empty_no_schema", data=[])


-def test_add_table_with_empty_embeddings(tmp_path):
-    """Test exact scenario from issue #1968
+def test_create_table_without_data_with_vector_schema(tmp_path):
+    """Test exact scenario from issue #1968.

    Regression test for issue #1968:
    https://github.com/lancedb/lancedb/issues/1968
@@ -3451,6 +3775,9 @@ def test_add_table_with_empty_embeddings(tmp_path):
        embedding: Vector(16)

    table = db.create_table("test", schema=MySchema)
+    assert table.count_rows() == 0
+    assert table.schema == MySchema.to_arrow_schema()
+
    table.add(
        [{"text": "bar", "embedding": [0.1] * 16}],
        on_bad_vectors="drop",
@@ -75,6 +75,22 @@ class TestVoyageAIModelRegistration:
        with pytest.raises(ValueError, match="not supported"):
            func.ndims()

+    def test_voyage3_source_embeddings_use_text_api(self, mock_voyageai_client):
+        """Regression test for text table data being sent to the multimodal API."""
+        mock_voyageai_client.tokenize.return_value = [["hello", "world"]]
+        mock_voyageai_client.embed.return_value.embeddings = [[0.1] * 1024]
+
+        registry = get_registry()
+        func = registry.get("voyageai").create(name="voyage-3")
+
+        embeddings = func.compute_source_embeddings("hello world")
+
+        assert embeddings == [[0.1] * 1024]
+        mock_voyageai_client.embed.assert_called_once_with(
+            texts=["hello world"], model="voyage-3", input_type="document"
+        )
+        mock_voyageai_client.multimodal_embed.assert_not_called()
+
    @pytest.mark.parametrize(
        "model_name",
        [
@@ -0,0 +1,15 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+from typing import assert_type
+
+import lancedb
+from lancedb import AsyncConnection, DBConnection
+
+
+def check_connect_type() -> None:
+    assert_type(lancedb.connect("memory://"), DBConnection)
+
+
+async def check_connect_async_type() -> None:
+    assert_type(await lancedb.connect_async("memory://"), AsyncConnection)
@@ -13,7 +13,11 @@ use crate::{
    runtime::future_into_py,
    table::Table,
 };
-use arrow::{datatypes::Schema, ffi_stream::ArrowArrayStreamReader, pyarrow::FromPyArrow};
+use arrow::{
+    datatypes::Schema,
+    ffi_stream::ArrowArrayStreamReader,
+    pyarrow::{FromPyArrow, ToPyArrow},
+};
 use lancedb::{
    connection::Connection as LanceConnection,
    connection::NamespaceClientPushdownOperation,
@@ -24,7 +28,7 @@ use pyo3::{
    Bound, FromPyObject, Py, PyAny, PyRef, PyResult, Python,
    exceptions::{PyRuntimeError, PyValueError},
    pyclass, pyfunction, pymethods,
-    types::{PyDict, PyDictMethods},
+    types::{PyDict, PyDictMethods, PyList, PyListMethods},
 };

 #[pyclass]
@@ -536,6 +540,55 @@ impl Connection {
            })
        })
    }
+
+    pub fn job(&self, job_id: String) -> PyResult<crate::job::Job> {
+        let inner = self.get_inner()?.clone();
+        Ok(crate::job::Job::new(inner.job(job_id).infer_error()?))
+    }
+
+    pub fn list_jobs(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.get_inner()?.clone();
+        future_into_py(self_.py(), async move {
+            let jobs = inner.list_jobs().await.infer_error()?;
+            Ok(jobs
+                .into_iter()
+                .map(crate::job::JobInfo::from)
+                .collect::<Vec<_>>())
+        })
+    }
+
+    pub fn get_job(self_: PyRef<'_, Self>, job_id: String) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.get_inner()?.clone();
+        future_into_py(self_.py(), async move {
+            let description = inner.get_job(&job_id).await.infer_error()?;
+            Ok(description.map(crate::job::JobDescription::from))
+        })
+    }
+
+    pub fn cancel_job(self_: PyRef<'_, Self>, job_id: String) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.get_inner()?.clone();
+        future_into_py(self_.py(), async move {
+            inner.cancel_job(&job_id).await.infer_error()
+        })
+    }
+
+    #[pyo3(signature = (job_id=None))]
+    pub fn job_history(
+        self_: PyRef<'_, Self>,
+        job_id: Option<String>,
+    ) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.get_inner()?.clone();
+        future_into_py(self_.py(), async move {
+            let batches = inner.job_history(job_id.as_deref()).await.infer_error()?;
+            Python::attach(|py| {
+                let list = PyList::empty(py);
+                for batch in batches {
+                    list.append(batch.to_pyarrow(py)?)?;
+                }
+                Ok(list.unbind())
+            })
+        })
+    }
 }

 #[pyfunction]
@@ -102,6 +102,18 @@ impl<T> PythonErrorExt<T> for std::result::Result<T, LanceError> {
                    err.setattr(intern!(py, "__cause__"), cause_err)?;
                    Err(PyErr::from_value(err))
                }),
+                LanceError::JobFailed { .. } => Python::attach(|py| {
+                    let cls = py
+                        .import(intern!(py, "lancedb.exceptions"))?
+                        .getattr(intern!(py, "JobFailedError"))?;
+                    Err(PyErr::from_value(cls.call1((err.to_string(),))?))
+                }),
+                LanceError::JobCancelled { .. } => Python::attach(|py| {
+                    let cls = py
+                        .import(intern!(py, "lancedb.exceptions"))?
+                        .getattr(intern!(py, "JobCancelledError"))?;
+                    Err(PyErr::from_value(cls.call1((err.to_string(),))?))
+                }),
                _ => self.runtime_error(),
            },
        }
@@ -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
@@ -0,0 +1,145 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+use std::sync::Arc;
+
+use crate::runtime::future_into_py;
+use pyo3::{Bound, PyAny, PyRef, PyResult, pyclass, pymethods};
+
+use crate::error::PythonErrorExt;
+
+#[pyclass]
+pub struct Job {
+    inner: Arc<lancedb::Job>,
+}
+
+impl Job {
+    pub(crate) fn new(inner: lancedb::Job) -> Self {
+        Self {
+            inner: Arc::new(inner),
+        }
+    }
+}
+
+#[pymethods]
+impl Job {
+    #[getter]
+    pub fn id(&self) -> Option<String> {
+        self.inner.id().map(str::to_string)
+    }
+
+    pub fn status(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.inner.clone();
+        future_into_py(
+            self_.py(),
+            async move { inner.status().await.infer_error() },
+        )
+    }
+
+    pub fn wait(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.inner.clone();
+        future_into_py(self_.py(), async move {
+            inner.wait().await.infer_error()?;
+            Ok(())
+        })
+    }
+
+    pub fn cancel(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.inner.clone();
+        future_into_py(self_.py(), async move {
+            inner.cancel().await.infer_error()?;
+            Ok(())
+        })
+    }
+}
+
+/// A row from `Connection.list_jobs`: one server-side job.
+#[pyclass(get_all, skip_from_py_object)]
+#[derive(Clone)]
+pub struct JobInfo {
+    job_id: String,
+    table: String,
+    job_type: String,
+    state: String,
+    created_at_millis: i64,
+}
+
+#[pymethods]
+impl JobInfo {
+    fn __repr__(&self) -> String {
+        format!(
+            "JobInfo(job_id={:?}, table={:?}, job_type={:?}, state={:?}, created_at_millis={})",
+            self.job_id, self.table, self.job_type, self.state, self.created_at_millis
+        )
+    }
+}
+
+impl From<lancedb::database::JobInfo> for JobInfo {
+    fn from(info: lancedb::database::JobInfo) -> Self {
+        Self {
+            job_id: info.job_id,
+            table: info.table,
+            job_type: info.job_type,
+            state: info.state,
+            created_at_millis: info.created_at_millis,
+        }
+    }
+}
+
+/// The server's account of why a job failed.
+#[pyclass(get_all, skip_from_py_object)]
+#[derive(Clone)]
+pub struct JobFailureInfo {
+    phase: Option<String>,
+    message: Option<String>,
+    retryable: Option<bool>,
+}
+
+#[pymethods]
+impl JobFailureInfo {
+    fn __repr__(&self) -> String {
+        format!(
+            "JobFailureInfo(phase={:?}, message={:?}, retryable={:?})",
+            self.phase, self.message, self.retryable
+        )
+    }
+}
+
+/// A described job from `Connection.get_job`.
+#[pyclass(get_all, skip_from_py_object)]
+#[derive(Clone)]
+pub struct JobDescription {
+    job_id: String,
+    job_type: String,
+    state: String,
+    creation_ms: i64,
+    spec_json: Option<String>,
+    failure: Option<JobFailureInfo>,
+}
+
+#[pymethods]
+impl JobDescription {
+    fn __repr__(&self) -> String {
+        format!(
+            "JobDescription(job_id={:?}, job_type={:?}, state={:?}, creation_ms={})",
+            self.job_id, self.job_type, self.state, self.creation_ms
+        )
+    }
+}
+
+impl From<lancedb::database::JobDescription> for JobDescription {
+    fn from(description: lancedb::database::JobDescription) -> Self {
+        Self {
+            job_id: description.job_id,
+            job_type: description.job_type,
+            state: description.state,
+            creation_ms: description.creation_ms,
+            spec_json: (!description.spec.is_null()).then(|| description.spec.to_string()),
+            failure: description.failure.map(|failure| JobFailureInfo {
+                phase: failure.phase,
+                message: failure.message,
+                retryable: failure.retryable,
+            }),
+        }
+    }
+}
@@ -25,6 +25,7 @@ pub mod error;
 pub mod expr;
 pub mod header;
 pub mod index;
+pub mod job;
 pub mod namespace;
 pub mod oauth;
 pub mod otel;
@@ -44,6 +45,10 @@ pub fn _lancedb(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
    m.add_class::<Connection>()?;
    m.add_class::<Session>()?;
    m.add_class::<Table>()?;
+    m.add_class::<crate::job::Job>()?;
+    m.add_class::<crate::job::JobInfo>()?;
+    m.add_class::<crate::job::JobDescription>()?;
+    m.add_class::<crate::job::JobFailureInfo>()?;
    m.add_class::<PyBlobFile>()?;
    m.add_class::<IndexConfig>()?;
    m.add_class::<Query>()?;
@@ -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>,
@@ -28,11 +28,72 @@ use pyo3::{
    Bound, FromPyObject, Py, PyAny, PyRef, PyResult, Python,
    exceptions::{PyRuntimeError, PyValueError},
    pyclass, pyfunction, pymethods,
-    types::{IntoPyDict, PyAnyMethods, PyBytes, PyDict, PyDictMethods},
+    types::{IntoPyDict, PyAnyMethods, PyBytes, PyDict, PyDictMethods, PyList, PyListMethods},
 };

 mod scannable;

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

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

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

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

    /// Default `ShardWriter` configuration recorded by this spec.
@@ -426,9 +503,11 @@ pub struct PyBlobFile {
 impl PyBlobFile {
    fn read_bytes(self_: PyRef<'_, Self>) -> PyResult<Py<PyBytes>> {
        let inner = self_.inner.clone();
-        let bytes = block_on(async move { inner.read().await })
+        let py = self_.py();
+        let bytes = py
+            .detach(move || block_on(async move { inner.read().await }))
            .map_err(|e| PyRuntimeError::new_err(format!("blob read failed: {e}")))?;
-        Ok(PyBytes::new(self_.py(), bytes.as_ref()).unbind())
+        Ok(PyBytes::new(py, bytes.as_ref()).unbind())
    }

    pub fn read(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
@@ -444,24 +523,32 @@ impl PyBlobFile {

    fn close(self_: PyRef<'_, Self>) -> PyResult<()> {
        let inner = self_.inner.clone();
-        block_on(async move { inner.close().await })
+        self_
+            .py()
+            .detach(move || block_on(async move { inner.close().await }))
            .map_err(|e| PyRuntimeError::new_err(format!("blob close failed: {e}")))
    }

    fn is_closed(self_: PyRef<'_, Self>) -> bool {
        let inner = self_.inner.clone();
-        block_on(async move { inner.is_closed().await })
+        self_
+            .py()
+            .detach(move || block_on(async move { inner.is_closed().await }))
    }

    fn seek(self_: PyRef<'_, Self>, position: u64) -> PyResult<()> {
        let inner = self_.inner.clone();
-        block_on(async move { inner.seek(position).await })
+        self_
+            .py()
+            .detach(move || block_on(async move { inner.seek(position).await }))
            .map_err(|e| PyRuntimeError::new_err(format!("blob seek failed: {e}")))
    }

    fn tell(self_: PyRef<'_, Self>) -> PyResult<u64> {
        let inner = self_.inner.clone();
-        block_on(async move { inner.tell().await })
+        self_
+            .py()
+            .detach(move || block_on(async move { inner.tell().await }))
            .map_err(|e| PyRuntimeError::new_err(format!("blob tell failed: {e}")))
    }

@@ -475,20 +562,24 @@ impl PyBlobFile {
            .checked_add(length as u64)
            .ok_or_else(|| PyValueError::new_err("offset + length overflowed"))?;
        let inner = self_.inner.clone();
-        let bytes = block_on(async move { inner.read_range(offset..end).await })
+        let py = self_.py();
+        let bytes = py
+            .detach(move || block_on(async move { inner.read_range(offset..end).await }))
            .map_err(|e| PyRuntimeError::new_err(format!("blob read_range failed: {e}")))?;
-        Ok(PyBytes::new(self_.py(), bytes.as_ref()).unbind())
+        Ok(PyBytes::new(py, bytes.as_ref()).unbind())
    }

    fn read_up_to(self_: PyRef<'_, Self>, length: usize) -> PyResult<Py<PyBytes>> {
        let inner = self_.inner.clone();
-        let bytes = block_on(async move { inner.read_up_to(length).await })
-            .map_err(|e| PyRuntimeError::new_err(format!("blob read failed: {e}")))?;
-        Ok(PyBytes::new(self_.py(), bytes.as_ref()).unbind())
+        let py = self_.py();
+        let bytes = py
+            .detach(move || block_on(async move { inner.read_up_to(length).await }))
+            .map_err(|e| PyRuntimeError::new_err(format!("blob read_up_to failed: {e}")))?;
+        Ok(PyBytes::new(py, bytes.as_ref()).unbind())
    }
 }

-#[pyclass(get_all, from_py_object)]
+#[pyclass(module = "lancedb._lancedb", get_all, from_py_object)]
 #[derive(Clone, Debug)]
 pub struct FtsToken {
    pub text: String,
@@ -731,6 +822,9 @@ impl Table {

    #[allow(private_interfaces)]
    pub fn delete(self_: PyRef<'_, Self>, condition: PredicateArg) -> PyResult<Bound<'_, PyAny>> {
+        // Do not hold the Python borrow across the await.  The cloned Rust table
+        // handle is thread-safe and allows deletes on the same Python table to
+        // run concurrently without PyO3 reporting "Already borrowed".
        let inner = self_.inner_ref()?.clone();
        future_into_py(self_.py(), async move {
            let result = match &condition {
@@ -805,6 +899,37 @@ impl Table {
        })
    }

+    #[pyo3(signature = (column, index=None, replace=None, wait_timeout=None, *, name=None, train=None))]
+    pub fn create_index_async<'a>(
+        self_: PyRef<'a, Self>,
+        column: String,
+        index: Option<Bound<'_, PyAny>>,
+        replace: Option<bool>,
+        wait_timeout: Option<Bound<'_, PyAny>>,
+        name: Option<String>,
+        train: Option<bool>,
+    ) -> PyResult<Bound<'a, PyAny>> {
+        let index = extract_index_params(&index)?;
+        let timeout = wait_timeout.map(|t| t.extract::<std::time::Duration>().unwrap());
+        let mut op = self_
+            .inner_ref()?
+            .create_index_with_timeout(&[column], index, timeout);
+        if let Some(replace) = replace {
+            op = op.replace(replace);
+        }
+        if let Some(name) = name {
+            op = op.name(name);
+        }
+        if let Some(train) = train {
+            op = op.train(train);
+        }
+
+        future_into_py(self_.py(), async move {
+            let job = op.execute_async().await.infer_error()?;
+            Ok(crate::job::Job::new(job))
+        })
+    }
+
    pub fn drop_index(self_: PyRef<'_, Self>, index_name: String) -> PyResult<Bound<'_, PyAny>> {
        let inner = self_.inner_ref()?.clone();
        future_into_py(self_.py(), async move {
@@ -1291,6 +1416,51 @@ impl Table {
        })
    }

+    /// Converge the table's LSM write path into its base table.
+    ///
+    /// Best-effort: with writes flowing, new rows may land after the last
+    /// pass. Errors if the table stops making progress.
+    pub fn checkpoint_lsm(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.inner_ref()?.clone();
+        future_into_py(self_.py(), async move {
+            inner.checkpoint_lsm().await.infer_error()
+        })
+    }
+
+    /// Seal every bucket's active memtable into L0.
+    pub fn flush_lsm(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.inner_ref()?.clone();
+        future_into_py(
+            self_.py(),
+            async move { inner.flush_lsm().await.infer_error() },
+        )
+    }
+
+    /// Trigger a background L0 → base pass per bucket. Returns once the
+    /// passes are dispatched, not once they finish — watch `get_lsm_stats`.
+    pub fn compact_lsm(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.inner_ref()?.clone();
+        future_into_py(self_.py(), async move {
+            inner.compact_lsm().await.infer_error()
+        })
+    }
+
+    /// Live LSM state, or `None` when the LSM write path is not enabled.
+    #[pyo3(signature = (include_generation_rows=false))]
+    pub fn get_lsm_stats(
+        self_: PyRef<'_, Self>,
+        include_generation_rows: bool,
+    ) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.inner_ref()?.clone();
+        future_into_py(self_.py(), async move {
+            let stats = inner
+                .get_lsm_stats(include_generation_rows)
+                .await
+                .infer_error()?;
+            Python::attach(|py| stats.map(|s| lsm_stats_to_py(py, &s)).transpose())
+        })
+    }
+
    pub fn close_lsm_writers(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
        let inner = self_.inner_ref()?.clone();
        future_into_py(self_.py(), async move {
@@ -1330,7 +1500,12 @@ impl Table {

        let inner = self_.inner_ref()?.clone();
        future_into_py(self_.py(), async move {
-            let result = inner.add_columns(definitions, None).await.infer_error()?;
+            let result = inner
+                .add_columns()
+                .transform(definitions)
+                .execute()
+                .await
+                .infer_error()?;
            Ok(AddColumnsResult::from(result))
        })
    }
@@ -1344,7 +1519,12 @@ impl Table {

        let inner = self_.inner_ref()?.clone();
        future_into_py(self_.py(), async move {
-            let result = inner.add_columns(transform, None).await.infer_error()?;
+            let result = inner
+                .add_columns()
+                .transform(transform)
+                .execute()
+                .await
+                .infer_error()?;
            Ok(AddColumnsResult::from(result))
        })
    }
@@ -1,2 +1,2 @@
 [toolchain]
-channel = "1.95.0"
+channel = "1.97.0"
@@ -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 transitive GooseFS SDK until the 0.1.6 compile break is fixed upstream.
-goosefs-sdk = { version = "=0.1.5", optional = true }
 moka = { workspace = true }
 pin-project = { workspace = true }
 tokio = { version = "1.23", features = ["rt-multi-thread", "sync"] }
@@ -75,6 +73,8 @@ reqwest = { version = "0.12.0", default-features = false, features = [
    "http2",
    "json",
    "macos-system-configuration",
+    # Avoid linking OpenSSL into Python wheels, which breaks on FIPS hosts.
+    "rustls-tls-native-roots",
    "stream",
 ], optional = true }
 http = { version = "1", optional = true } # Matching what is in reqwest
@@ -98,7 +98,8 @@ anyhow = "1"
 lance-testing = { workspace = true }
 tempfile = "3.5.0"
 random_word = { version = "0.4.3", features = ["en"] }
-tokio = { version = "1.23", features = ["io-util", "macros", "net", "rt-multi-thread", "sync"] }
+roaring = "0.11.4"
+tokio = { version = "1.23", features = ["io-util", "macros", "net", "rt-multi-thread", "sync", "test-util"] }
 uuid = { version = "1.7.0", features = ["v4"] }
 walkdir = "2"
 aws-sdk-dynamodb = { version = "1.55.0" }
@@ -133,7 +134,6 @@ azure = [
 ]
 cos = ["lance/tencent", "lance-io/tencent"]
 goosefs = [
-    "dep:goosefs-sdk",
    "lance/goosefs",
    "lance-io/goosefs",
    "lance-namespace-impls/dir-goosefs",
@@ -188,6 +188,9 @@ required-features = ["bedrock"]
 [[example]]
 name = "bench_streaming_dataloader"

+[[example]]
+name = "bench_open_missing_table"
+
 [[example]]
 name = "simple"

@@ -0,0 +1,150 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+// Release benchmark for opening a missing table as sibling-table cardinality grows.
+//
+// The fixture uses real `.lance` directories and marker files. Fixture creation is
+// outside the timed section. Defaults intentionally cover 1k, 10k, and 100k siblings
+// with 10 warmups and 100 distinct missing-table opens per scale:
+//
+// ```text
+// cargo run --release -p lancedb --example bench_open_missing_table
+// ```
+//
+// `BENCH_SIBLINGS`, `BENCH_WARMUPS`, and `BENCH_TRIALS` override those defaults.
+// Reduced settings are useful only as a smoke test. Performance comparisons require
+// the same machine, filesystem, fixture sizes, settings, lockfile, and alternating
+// baseline/candidate execution order.
+
+use std::time::{Duration, Instant};
+
+use anyhow::{Context, Result, bail};
+use lancedb::connection::Connection;
+use lancedb::{Error, connect};
+use object_store::ObjectStoreExt as _;
+use object_store::path::Path;
+
+const MAX_SIBLINGS: usize = 1_000_000;
+const MAX_WARMUPS: usize = 10_000;
+const MAX_TRIALS: usize = 100_000;
+
+fn env_usize(key: &str, default: usize, max: usize) -> Result<usize> {
+    let value = match std::env::var(key) {
+        Ok(value) => value
+            .parse()
+            .with_context(|| format!("invalid {key} value: {value}"))?,
+        Err(std::env::VarError::NotPresent) => default,
+        Err(error) => return Err(error).with_context(|| format!("reading {key}")),
+    };
+    if value == 0 || value > max {
+        bail!("{key} must be between 1 and {max}");
+    }
+    Ok(value)
+}
+
+fn sibling_counts() -> Result<Vec<usize>> {
+    let raw = std::env::var("BENCH_SIBLINGS").unwrap_or_else(|_| "1000,10000,100000".into());
+    let mut counts = raw
+        .split(',')
+        .map(|value| {
+            value
+                .trim()
+                .parse::<usize>()
+                .with_context(|| format!("invalid BENCH_SIBLINGS value: {value}"))
+        })
+        .collect::<Result<Vec<_>>>()?;
+    counts.sort_unstable();
+    counts.dedup();
+    if counts.is_empty() || counts[0] == 0 || counts[counts.len() - 1] > MAX_SIBLINGS {
+        bail!("BENCH_SIBLINGS values must be between 1 and {MAX_SIBLINGS}");
+    }
+    Ok(counts)
+}
+
+async fn add_siblings(
+    store: &object_store::local::LocalFileSystem,
+    start: usize,
+    end: usize,
+) -> Result<()> {
+    for index in start..end {
+        let marker = Path::from(format!("sibling_{index:06}.lance/_marker"));
+        store
+            .put(&marker, bytes::Bytes::new().into())
+            .await
+            .with_context(|| format!("creating benchmark marker {marker}"))?;
+    }
+    Ok(())
+}
+
+async fn time_missing_open(db: &Connection, name: &str) -> Result<Duration> {
+    let started = Instant::now();
+    let result = db.open_table(name).execute().await;
+    let elapsed = started.elapsed();
+    match result {
+        Err(Error::TableNotFound { .. }) => Ok(elapsed),
+        Err(error) => bail!("expected TableNotFound for {name}, got {error:?}"),
+        Ok(_) => bail!("benchmark missing-table name unexpectedly exists: {name}"),
+    }
+}
+
+fn percentile(sorted: &[Duration], percentile: usize) -> Duration {
+    let rank = (sorted.len() * percentile).div_ceil(100).saturating_sub(1);
+    sorted[rank]
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    let counts = sibling_counts()?;
+    let warmups = env_usize("BENCH_WARMUPS", 10, MAX_WARMUPS)?;
+    let trials = env_usize("BENCH_TRIALS", 100, MAX_TRIALS)?;
+
+    let fixture = tempfile::tempdir().context("creating benchmark fixture")?;
+    let database_path = fixture.path();
+    let fixture_store = object_store::local::LocalFileSystem::new_with_prefix(database_path)
+        .context("creating benchmark object store")?;
+    let db = connect(database_path.to_str().context("non-UTF-8 fixture path")?)
+        .execute()
+        .await?;
+
+    println!(
+        "config: siblings={counts:?} warmups={warmups} trials={trials} profile={} os={} arch={}",
+        if cfg!(debug_assertions) {
+            "debug"
+        } else {
+            "release"
+        },
+        std::env::consts::OS,
+        std::env::consts::ARCH,
+    );
+    println!("lower is better; fixture setup and teardown are excluded");
+    println!("| siblings | samples | p50 | p95 | max |");
+    println!("| ---: | ---: | ---: | ---: | ---: |");
+
+    let mut created = 0;
+    for sibling_count in counts {
+        add_siblings(&fixture_store, created, sibling_count).await?;
+        created = sibling_count;
+
+        for index in 0..warmups {
+            let name = format!("__missing_warmup_{sibling_count}_{index}");
+            let _ = time_missing_open(&db, &name).await?;
+        }
+
+        let mut samples = Vec::with_capacity(trials);
+        for index in 0..trials {
+            let name = format!("__missing_trial_{sibling_count}_{index}");
+            samples.push(time_missing_open(&db, &name).await?);
+        }
+        samples.sort_unstable();
+
+        println!(
+            "| {sibling_count} | {} | {:?} | {:?} | {:?} |",
+            samples.len(),
+            percentile(&samples, 50),
+            percentile(&samples, 95),
+            samples[samples.len() - 1],
+        );
+    }
+
+    Ok(())
+}
@@ -9,6 +9,7 @@
 //!
 //! Blob tables require Lance file format >= 2.2 and stable row ids at create.

+use std::ops::Range;
 use std::sync::Arc;

 use arrow_array::LargeBinaryArray;
@@ -16,11 +17,203 @@ use arrow_array::builder::LargeBinaryBuilder;
 use arrow_schema::{DataType, Field, Schema};
 use lance::dataset::{BlobRangeRequest as LanceBlobRangeRequest, Dataset, WriteParams};
 use lance_arrow::FieldExt;
-use lance_encoding::version::LanceFileVersion;
+use lance_file::version::{ConcreteFileVersion, LanceFileVersion};
+use lance_io::object_store::ObjectStore;
+use object_store::path::Path;

 use crate::error::{Error, Result};

-pub use lance::dataset::BlobFile;
+/// Seekable handle for one blob value, backed by local storage or a remote
+/// HTTP byte-range endpoint.
+#[derive(Debug)]
+pub struct BlobFile {
+    inner: BlobFileInner,
+}
+
+#[derive(Debug)]
+enum BlobFileInner {
+    Native(lance::dataset::BlobFile),
+    #[cfg(feature = "remote")]
+    Remote(Box<crate::remote::table::blobs::RemoteBlobFile>),
+}
+
+impl From<lance::dataset::BlobFile> for BlobFile {
+    fn from(value: lance::dataset::BlobFile) -> Self {
+        Self {
+            inner: BlobFileInner::Native(value),
+        }
+    }
+}
+
+#[cfg(feature = "remote")]
+impl From<crate::remote::table::blobs::RemoteBlobFile> for BlobFile {
+    fn from(value: crate::remote::table::blobs::RemoteBlobFile) -> Self {
+        Self {
+            inner: BlobFileInner::Remote(Box::new(value)),
+        }
+    }
+}
+
+impl BlobFile {
+    /// Inline reader over a data-file slice.
+    pub fn new_inline(
+        object_store: Arc<ObjectStore>,
+        path: Path,
+        position: u64,
+        size: u64,
+    ) -> Self {
+        lance::dataset::BlobFile::new_inline(object_store, path, position, size).into()
+    }
+
+    /// Dedicated sidecar-file reader.
+    pub fn new_dedicated(object_store: Arc<ObjectStore>, path: Path, size: u64) -> Self {
+        lance::dataset::BlobFile::new_dedicated(object_store, path, size).into()
+    }
+
+    /// Packed reader for a slice in a shared sidecar.
+    pub fn new_packed(
+        object_store: Arc<ObjectStore>,
+        path: Path,
+        position: u64,
+        size: u64,
+    ) -> Self {
+        lance::dataset::BlobFile::new_packed(object_store, path, position, size).into()
+    }
+
+    /// External reader at a resolved object location.
+    pub fn new_external(
+        object_store: Arc<ObjectStore>,
+        path: Path,
+        uri: String,
+        position: u64,
+        size: u64,
+    ) -> Self {
+        lance::dataset::BlobFile::new_external(object_store, path, uri, position, size).into()
+    }
+
+    /// Close the handle.
+    pub async fn close(&self) -> lance_core::Result<()> {
+        match &self.inner {
+            BlobFileInner::Native(file) => file.close().await,
+            #[cfg(feature = "remote")]
+            BlobFileInner::Remote(file) => file.close().await,
+        }
+    }
+
+    /// Whether the handle is closed.
+    pub async fn is_closed(&self) -> bool {
+        match &self.inner {
+            BlobFileInner::Native(file) => file.is_closed().await,
+            #[cfg(feature = "remote")]
+            BlobFileInner::Remote(file) => file.is_closed(),
+        }
+    }
+
+    /// Read a range without moving the cursor.
+    pub async fn read_range(&self, range: Range<u64>) -> lance_core::Result<bytes::Bytes> {
+        match &self.inner {
+            BlobFileInner::Native(file) => file.read_range(range).await,
+            #[cfg(feature = "remote")]
+            BlobFileInner::Remote(file) => file.read_range(range).await,
+        }
+    }
+
+    /// Read ranges without moving the cursor.
+    pub async fn read_ranges(
+        &self,
+        ranges: &[Range<u64>],
+    ) -> lance_core::Result<Vec<bytes::Bytes>> {
+        match &self.inner {
+            BlobFileInner::Native(file) => file.read_ranges(ranges).await,
+            #[cfg(feature = "remote")]
+            BlobFileInner::Remote(file) => file.read_ranges(ranges).await,
+        }
+    }
+
+    /// Read from the cursor to the end.
+    pub async fn read(&self) -> lance_core::Result<bytes::Bytes> {
+        match &self.inner {
+            BlobFileInner::Native(file) => file.read().await,
+            #[cfg(feature = "remote")]
+            BlobFileInner::Remote(file) => file.read().await,
+        }
+    }
+
+    /// Read up to `len` bytes and advance the cursor.
+    pub async fn read_up_to(&self, len: usize) -> lance_core::Result<bytes::Bytes> {
+        match &self.inner {
+            BlobFileInner::Native(file) => file.read_up_to(len).await,
+            #[cfg(feature = "remote")]
+            BlobFileInner::Remote(file) => file.read_up_to(len).await,
+        }
+    }
+
+    /// Move the cursor to `new_cursor`.
+    pub async fn seek(&self, new_cursor: u64) -> lance_core::Result<()> {
+        match &self.inner {
+            BlobFileInner::Native(file) => file.seek(new_cursor).await,
+            #[cfg(feature = "remote")]
+            BlobFileInner::Remote(file) => file.seek(new_cursor).await,
+        }
+    }
+
+    /// Current cursor position.
+    pub async fn tell(&self) -> lance_core::Result<u64> {
+        match &self.inner {
+            BlobFileInner::Native(file) => file.tell().await,
+            #[cfg(feature = "remote")]
+            BlobFileInner::Remote(file) => file.tell().await,
+        }
+    }
+
+    /// Blob length in bytes.
+    pub fn size(&self) -> u64 {
+        match &self.inner {
+            BlobFileInner::Native(file) => file.size(),
+            #[cfg(feature = "remote")]
+            BlobFileInner::Remote(file) => file.size(),
+        }
+    }
+
+    /// Physical byte offset in the data file. `None` on remote handles. The
+    /// Cloud byte-range route does not expose storage layout.
+    pub fn position(&self) -> Option<u64> {
+        match &self.inner {
+            BlobFileInner::Native(file) => Some(file.position()),
+            #[cfg(feature = "remote")]
+            BlobFileInner::Remote(_) => None,
+        }
+    }
+
+    /// Path of the data file holding the blob. `None` on remote handles. The
+    /// Cloud byte-range route does not expose storage layout.
+    pub fn data_path(&self) -> Option<&Path> {
+        match &self.inner {
+            BlobFileInner::Native(file) => Some(file.data_path()),
+            #[cfg(feature = "remote")]
+            BlobFileInner::Remote(_) => None,
+        }
+    }
+
+    /// Native storage layout. `None` on remote handles. The Cloud byte-range
+    /// route does not expose layout.
+    pub fn kind(&self) -> Option<lance_core::datatypes::BlobKind> {
+        match &self.inner {
+            BlobFileInner::Native(file) => Some(file.kind()),
+            #[cfg(feature = "remote")]
+            BlobFileInner::Remote(_) => None,
+        }
+    }
+
+    /// External URI for native handles. Remote handles do not expose storage URIs.
+    pub fn uri(&self) -> Option<&str> {
+        match &self.inner {
+            BlobFileInner::Native(file) => file.uri(),
+            #[cfg(feature = "remote")]
+            BlobFileInner::Remote(_) => None,
+        }
+    }
+}

 /// One row-specific blob range read request.
 ///
@@ -140,7 +333,10 @@ pub(crate) fn ensure_blob_storage_version(schema: &Schema, params: &mut WritePar
        .data_storage_version
        .unwrap_or(LanceFileVersion::Stable)
        .resolve();
-    if resolved < LanceFileVersion::V2_2 {
+    if matches!(
+        resolved,
+        ConcreteFileVersion::V1 | ConcreteFileVersion::V2_0 | ConcreteFileVersion::V2_1
+    ) {
        params.data_storage_version = Some(LanceFileVersion::V2_2);
    }
 }
@@ -264,7 +460,10 @@ pub(crate) async fn take_blob_files_aligned(

    let handles = dataset.take_blobs(row_ids, column).await?;
    ensure_all_row_ids_resolved(column, row_ids.len(), handles.len())?;
-    Ok(handles)
+    Ok(handles
+        .into_iter()
+        .map(|handle| handle.map(Into::into))
+        .collect())
 }

 #[cfg(test)]
@@ -303,7 +502,7 @@ mod tests {
        ensure_blob_storage_version(&blob_schema(), &mut params);
        assert_eq!(
            params.data_storage_version.unwrap().resolve(),
-            LanceFileVersion::V2_2
+            ConcreteFileVersion::V2_2
        );
    }

@@ -316,7 +515,7 @@ mod tests {
        ensure_blob_storage_version(&blob_schema(), &mut params);
        assert_eq!(
            params.data_storage_version.unwrap().resolve(),
-            LanceFileVersion::V2_2
+            ConcreteFileVersion::V2_2
        );
    }

@@ -23,8 +23,8 @@ use crate::connection::create_table::CreateTableBuilder;
 use crate::data::scannable::Scannable;
 use crate::database::listing::ListingDatabase;
 use crate::database::{
-    CloneTableRequest, Database, DatabaseOptions, OpenTableRequest, ReadConsistency,
-    TableNamesRequest,
+    CloneTableRequest, Database, DatabaseOptions, JobDescription, JobInfo, OpenTableRequest,
+    ReadConsistency, TableNamesRequest,
 };
 use crate::embeddings::{EmbeddingRegistry, MemoryRegistry};
 use crate::error::{Error, Result};
@@ -34,7 +34,7 @@ use crate::remote::{
    db::{OPT_REMOTE_API_KEY, OPT_REMOTE_HOST_OVERRIDE, OPT_REMOTE_REGION},
 };
 use lance::io::ObjectStoreParams;
-pub use lance_encoding::version::LanceFileVersion;
+pub use lance_file::version::LanceFileVersion;
 #[cfg(feature = "remote")]
 use lance_io::object_store::StorageOptions;
 use lance_io::object_store::{StorageOptionsAccessor, StorageOptionsProvider};
@@ -409,6 +409,11 @@ impl Connection {
    ///
    /// The names will be returned in lexicographical order (ascending)
    ///
+    /// Listing databases discover physical `*.lance` entries without opening every
+    /// dataset. The result is a point-in-time discovery snapshot: an entry may still be
+    /// under creation, may contain only uncommitted storage, or may be concurrently
+    /// dropped before it is opened.
+    ///
    /// The parameters `page_token` and `limit` can be used to paginate the results
    pub fn table_names(&self) -> TableNamesBuilder {
        TableNamesBuilder::new(self.internal.clone())
@@ -456,6 +461,9 @@ impl Connection {
    ///
    /// # Returns
    /// Created [`TableRef`], or [`Error::TableNotFound`] if the table does not exist.
+    /// On listing databases, a committed Lance manifest is authoritative for table
+    /// existence. Uncommitted files or a physical `<name>.lance` directory alone do not
+    /// make a table openable.
    pub fn open_table(&self, name: impl Into<String>) -> OpenTableBuilder {
        OpenTableBuilder::new(
            self.internal.clone(),
@@ -513,6 +521,39 @@ impl Connection {
        self.internal.read_consistency().await
    }

+    /// A [`crate::job::Job`] handle for a server-side job by id, suitable for
+    /// waiting on or cancelling the job.
+    ///
+    /// The handle is constructed without a server round trip; an unknown id
+    /// surfaces when the handle is used. Only server-backed databases support
+    /// job handles by id.
+    pub fn job(&self, job_id: impl AsRef<str>) -> Result<crate::job::Job> {
+        self.internal.job(job_id.as_ref())
+    }
+
+    /// List server-side jobs across the database's tables.
+    pub async fn list_jobs(&self) -> Result<Vec<JobInfo>> {
+        self.internal.list_jobs().await
+    }
+
+    /// Describe a single server-side job by id. `None` when the server has no
+    /// such job.
+    pub async fn get_job(&self, job_id: impl AsRef<str>) -> Result<Option<JobDescription>> {
+        self.internal.get_job(job_id.as_ref()).await
+    }
+
+    /// Request cancellation of a server-side job by id. Returns true if the
+    /// server accepted the cancellation, false if no such job exists.
+    pub async fn cancel_job(&self, job_id: impl AsRef<str>) -> Result<bool> {
+        self.internal.cancel_job(job_id.as_ref()).await
+    }
+
+    /// The lifecycle event history of a server-side job (all jobs when
+    /// `job_id` is `None`), as recorded Arrow batches.
+    pub async fn job_history(&self, job_id: Option<&str>) -> Result<Vec<RecordBatch>> {
+        self.internal.job_history(job_id).await
+    }
+
    /// Drop a table in the database.
    ///
    /// # Arguments
@@ -202,6 +202,17 @@ mod tests {
        assert_eq!(table.count_rows(None).await.unwrap(), 0);
    }

+    #[tokio::test]
+    async fn create_table_in_named_memory_database() {
+        let db = connect("memory://foo").execute().await.unwrap();
+        let batch = record_batch!(("id", Int64, [1, 2, 3])).unwrap();
+
+        let table = db.create_table("my_table", batch).execute().await.unwrap();
+
+        assert_eq!(table.uri().await.unwrap(), "memory://foo/my_table.lance");
+        assert_eq!(table.count_rows(None).await.unwrap(), 3);
+    }
+
    async fn test_create_table_with_data<T>(data: T)
    where
        T: Scannable + 'static,
@@ -427,10 +438,9 @@ mod tests {
            .await
            .unwrap()
            .data_storage_format
-            .lance_file_version()
-            .unwrap();
+            .lance_file_format();
        // Compare resolved versions since Stable/Next are aliases that resolve at storage time
-        assert_eq!(storage_format.resolve(), data_storage_version.resolve());
+        assert_eq!(storage_format, data_storage_version.resolve());
    }

    #[tokio::test]
@@ -18,6 +18,8 @@ use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::Duration;

+use arrow_array::RecordBatch;
+
 use lance::dataset::ReadParams;
 use lance_namespace::LanceNamespace;
 use lance_namespace::models::{
@@ -200,6 +202,45 @@ pub enum ReadConsistency {
    Strong,
 }

+/// A row from [`Database::list_jobs`]: one server-side job (index build,
+/// compaction, column refresh, ...).
+#[derive(Debug, Clone)]
+pub struct JobInfo {
+    /// The job id -- what [`Database::get_job`] and [`Database::cancel_job`]
+    /// accept.
+    pub job_id: String,
+    /// The table the job runs against, without URI or namespace.
+    pub table: String,
+    pub job_type: String,
+    /// Lifecycle state: "running", "finished", "failed", or "cancelled".
+    pub state: String,
+    /// When the job was created, in milliseconds since the epoch.
+    pub created_at_millis: i64,
+}
+
+/// A described job from [`Database::get_job`]: lifecycle state plus the
+/// job-type-specific specification.
+#[derive(Debug, Clone)]
+pub struct JobDescription {
+    pub job_id: String,
+    pub job_type: String,
+    /// Lifecycle state: "running", "finished", "failed", or "cancelled".
+    pub state: String,
+    /// When the job was created, in milliseconds since the epoch.
+    pub creation_ms: i64,
+    /// The job-type-specific specification. Null when the server omits it.
+    pub spec: serde_json::Value,
+    /// Why the job failed, when the job is failed and the server reports a
+    /// reason.
+    pub failure: Option<crate::error::JobFailure>,
+}
+
+fn job_op_not_supported<T>(what: &str) -> Result<T> {
+    Err(crate::error::Error::NotSupported {
+        message: format!("{} is not supported by this database", what),
+    })
+}
+
 /// The `Database` trait defines the interface for database implementations.
 ///
 /// A database is responsible for managing tables and their metadata.
@@ -245,6 +286,31 @@ pub trait Database:
    ///
    /// See [`CloneTableRequest`] for detailed documentation and examples.
    async fn clone_table(&self, request: CloneTableRequest) -> Result<Arc<dyn BaseTable>>;
+    /// A [`crate::job::Job`] handle for a server-side job by id, suitable for
+    /// waiting on or cancelling the job. The handle is constructed without a
+    /// server round trip; an unknown id surfaces when the handle is used.
+    fn job(&self, _job_id: &str) -> Result<crate::job::Job> {
+        job_op_not_supported("job")
+    }
+    /// List server-side jobs across the database's tables.
+    async fn list_jobs(&self) -> Result<Vec<JobInfo>> {
+        job_op_not_supported("list_jobs")
+    }
+    /// Describe a single job by id. `None` when the server has no such job.
+    async fn get_job(&self, _job_id: &str) -> Result<Option<JobDescription>> {
+        job_op_not_supported("get_job")
+    }
+    /// Request cancellation of a job by id. Returns true if the server
+    /// accepted the cancellation, false if no such job exists. Cancelling an
+    /// already-terminal job is a no-op success.
+    async fn cancel_job(&self, _job_id: &str) -> Result<bool> {
+        job_op_not_supported("cancel_job")
+    }
+    /// The lifecycle event history of a job (all jobs when `job_id` is
+    /// `None`), as recorded Arrow batches.
+    async fn job_history(&self, _job_id: Option<&str>) -> Result<Vec<RecordBatch>> {
+        job_op_not_supported("job_history")
+    }
    /// Open a table in the database
    async fn open_table(&self, request: OpenTableRequest) -> Result<Arc<dyn BaseTable>>;
    /// Rename a table in the database
--- a/Show More
+++ b/Show More