diff --git a/.github/workflows/docs-link-check.yml b/.github/workflows/docs-link-check.yml
index 0e22100eb..1286819bc 100644
--- a/.github/workflows/docs-link-check.yml
+++ b/.github/workflows/docs-link-check.yml
@@ -36,7 +36,9 @@ jobs:
     permissions:
       contents: read
     outputs:
+      checker_outcome: ${{ steps.lychee.outcome }}
       exit_code: ${{ steps.lychee.outputs.exit_code }}
+      status: ${{ steps.validate.outputs.status }}
     steps:
       - name: Checkout
         uses: actions/checkout@v6
@@ -50,6 +52,7 @@ jobs:
 
       - name: Check links
         id: lychee
+        continue-on-error: true
         uses: lycheeverse/lychee-action@e7477775783ea5526144ba13e8db5eec57747ce8 # v2.9.0
         with:
           # Restricted to http(s) on purpose. Much of docs/src is generated
@@ -68,38 +71,50 @@ jobs:
           format: json
           output: ./lychee/out.json
           jobSummary: false
-          # The report, not a red build, is the signal for broken links. The
-          # validation step below still fails the run if the check itself
-          # breaks.
+          # The report issue, not a red workflow run, is the signal for link
+          # findings and checker failures alike.
           fail: false
 
       - name: Validate report
+        id: validate
         # lychee does not reserve exit code 2 for broken links: its CLI
         # parser also exits 2 on an invalid option, before any link was
         # checked or any report written. Only a parseable report whose
-        # counts agree with the exit code counts as a link verdict; anything
-        # else fails here, and the report job below is skipped entirely, so
-        # the tracking issue is never touched. Exit 2 covers timeouts as
-        # well as errors, and a timed-out host is exactly the transient
-        # unavailability this report exists to surface, so both count as
-        # findings. Requiring total > 0 also catches a glob that silently
-        # stopped matching any file.
-        if: steps.lychee.outputs.exit_code == 0 || steps.lychee.outputs.exit_code == 2
+        # counts agree with a completed exit code (0 or 2) counts as a link
+        # verdict. Everything else becomes a checker-error report instead of
+        # failing the workflow. Exit 2 covers timeouts as well as errors, and a
+        # timed-out host is exactly the transient unavailability this report
+        # exists to surface, so both count as findings. Requiring total > 0
+        # also catches a glob that silently stopped matching any file.
+        if: always()
         env:
+          CHECKER_OUTCOME: ${{ steps.lychee.outcome }}
           EXIT_CODE: ${{ steps.lychee.outputs.exit_code }}
         run: |
-          jq -e --argjson code "$EXIT_CODE" '
-            (.total > 0) and
-            (if $code == 0
-             then .errors == 0 and .timeouts == 0
-               and (.error_map | length == 0) and (.timeout_map | length == 0)
-             else (.errors + .timeouts) > 0
-               and ((.error_map | length) + (.timeout_map | length)) > 0
-             end)
-          ' ./lychee/out.json
+          status=checker-error
+          if [[ "$CHECKER_OUTCOME" == success ]] &&
+             [[ "$EXIT_CODE" == 0 || "$EXIT_CODE" == 2 ]] &&
+             jq -e --argjson code "$EXIT_CODE" '
+               (.total > 0) and
+               (if $code == 0
+                then .errors == 0 and .timeouts == 0
+                  and (.error_map | length == 0) and (.timeout_map | length == 0)
+                else (.errors + .timeouts) > 0
+                  and ((.error_map | length) + (.timeout_map | length)) > 0
+                end)
+             ' ./lychee/out.json
+          then
+            if [[ "$EXIT_CODE" == 0 ]]; then
+              status=healthy
+            else
+              status=findings
+            fi
+          fi
+          echo "status=$status" >> "$GITHUB_OUTPUT"
+          echo "Validated link check as $status"
 
       - name: Upload report
-        if: steps.lychee.outputs.exit_code == 2
+        if: steps.validate.outputs.status == 'findings'
         uses: actions/upload-artifact@v7
         with:
           name: link-report
@@ -115,26 +130,11 @@ jobs:
     permissions:
       issues: write
     env:
+      CHECKER_OUTCOME: ${{ needs.scan.outputs.checker_outcome }}
       EXIT_CODE: ${{ needs.scan.outputs.exit_code }}
+      STATUS: ${{ needs.scan.outputs.status }}
       GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
     steps:
-      - name: Classify checker result
-        # lychee exits 0 when every link resolves and 2 when links fail,
-        # both already cross-checked against the report by the scan job's
-        # validation step. Anything else (1 runtime, 3 bad config) means the
-        # check never produced a link verdict, which must surface as a failed
-        # run rather than be published as "broken documentation links".
-        run: |
-          case "$EXIT_CODE" in
-            0|2)
-              echo "lychee exit code $EXIT_CODE"
-              ;;
-            *)
-              echo "::error::lychee exited with '$EXIT_CODE': the link check did not complete. Leaving the report issue untouched."
-              exit 1
-              ;;
-          esac
-
       - name: Find existing report issue
         id: report
         # Matched on title alone, and through search rather than a listing:
@@ -144,7 +144,7 @@ jobs:
         # Closed issues are included because a healthy run closes the report:
         # an open-only lookup would forget that identity and the next failing
         # run would open a duplicate. The oldest match stays the canonical
-        # report and is reopened below when links break again.
+        # report and is reopened below when a problem recurs.
         run: |
           match=$(gh issue list --repo "$GITHUB_REPOSITORY" --state all \
             --search "in:title \"$REPORT_TITLE\" author:app/github-actions" \
@@ -154,14 +154,14 @@ jobs:
           echo "state=$(jq -r '.state // empty' <<<"$match")" >> "$GITHUB_OUTPUT"
 
       - name: Download report
-        if: env.EXIT_CODE == 2
+        if: env.STATUS == 'findings'
         uses: actions/download-artifact@v8
         with:
           name: link-report
           path: ./lychee
 
       - name: Compose report
-        if: env.EXIT_CODE == 2
+        if: env.STATUS == 'findings'
         run: |
           run_url="$GITHUB_SERVER_URL/$GITHUB_REPOSITORY/actions/runs/$GITHUB_RUN_ID"
           {
@@ -185,22 +185,41 @@ jobs:
             ' ./lychee/out.json
           } > ./lychee/issue.md
 
+      - name: Compose checker error report
+        if: env.STATUS == 'checker-error'
+        run: |
+          mkdir -p ./lychee
+          run_url="$GITHUB_SERVER_URL/$GITHUB_REPOSITORY/actions/runs/$GITHUB_RUN_ID"
+          {
+            echo "The documentation link check did not complete in [the latest run]($run_url)."
+            echo
+            echo "This issue is rewritten by every scheduled run and closed automatically once a trustworthy run finds that all links resolve."
+            echo
+            echo "The checker did not produce a trustworthy link verdict. Treat the previous result, if any, as stale until a later run completes."
+            echo
+            echo "* Action outcome: \`$CHECKER_OUTCOME\`"
+            echo "* Exit code: \`${EXIT_CODE:-not reported}\`"
+            echo "* Verdict validation: \`failed\`"
+          } > ./lychee/issue.md
+
       - name: Reopen report issue
         # A healthy run closes the report, and the issue action below only
         # rewrites the body of whatever number it is given. Without an
-        # explicit reopen, the 2 -> 0 -> 2 sequence would keep rewriting a
-        # closed issue while links are broken. A CLOSED state implies the
-        # lookup found a canonical issue, so no separate emptiness check.
-        if: env.EXIT_CODE == 2 && steps.report.outputs.state == 'CLOSED'
+        # explicit reopen, a later finding or checker error would rewrite a
+        # closed issue. A CLOSED state implies the lookup found a canonical
+        # issue, so no separate emptiness check.
+        if: >-
+          env.STATUS != 'healthy' &&
+          steps.report.outputs.state == 'CLOSED'
         env:
           ISSUE_NUMBER: ${{ steps.report.outputs.number }}
         run: |
           run_url="$GITHUB_SERVER_URL/$GITHUB_REPOSITORY/actions/runs/$GITHUB_RUN_ID"
           gh issue reopen "$ISSUE_NUMBER" --repo "$GITHUB_REPOSITORY" \
-            --comment "Broken documentation links found again in [the latest run]($run_url)."
+            --comment "The documentation link checker reported a problem again in [the latest run]($run_url)."
 
-      - name: Report broken links
-        if: env.EXIT_CODE == 2
+      - name: Report link-check problem
+        if: env.STATUS != 'healthy'
         uses: peter-evans/create-issue-from-file@fca9117c27cdc29c6c4db3b86c48e4115a786710 # v6.0.0
         with:
           # Empty on the first failing run, which creates the issue; afterwards
@@ -213,7 +232,9 @@ jobs:
       - name: Close report issue once links are healthy
         # An OPEN state implies the lookup found a canonical issue; a report
         # that is already closed needs nothing.
-        if: env.EXIT_CODE == 0 && steps.report.outputs.state == 'OPEN'
+        if: >-
+          env.STATUS == 'healthy' &&
+          steps.report.outputs.state == 'OPEN'
         env:
           ISSUE_NUMBER: ${{ steps.report.outputs.number }}
         run: |
diff --git a/.github/workflows/pypi-publish.yml b/.github/workflows/pypi-publish.yml
index 74b7d05e6..4f5a927dc 100644
--- a/.github/workflows/pypi-publish.yml
+++ b/.github/workflows/pypi-publish.yml
@@ -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
diff --git a/Cargo.lock b/Cargo.lock
index cad82613e..f1f0ae467 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -3455,8 +3455,8 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"
 
 [[package]]
 name = "fsst"
-version = "11.0.0-beta.3"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
+version = "11.0.0-beta.7"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
 dependencies = [
  "arrow-array",
  "rand 0.9.5",
@@ -4815,8 +4815,8 @@ checksum = "e037a2e1d8d5fdbd49b16a4ea09d5d6401c1f29eca5ff29d03d3824dba16256a"
 
 [[package]]
 name = "lance"
-version = "11.0.0-beta.3"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
+version = "11.0.0-beta.7"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
 dependencies = [
  "arc-swap",
  "arrow",
@@ -4890,8 +4890,8 @@ dependencies = [
 
 [[package]]
 name = "lance-arrow"
-version = "11.0.0-beta.3"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
+version = "11.0.0-beta.7"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
 dependencies = [
  "arrow-array",
  "arrow-buffer",
@@ -4913,7 +4913,7 @@ dependencies = [
 [[package]]
 name = "lance-arrow-scalar"
 version = "58.0.0"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
 dependencies = [
  "arrow-array",
  "arrow-buffer",
@@ -4927,7 +4927,7 @@ dependencies = [
 [[package]]
 name = "lance-arrow-stats"
 version = "58.0.0"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
 dependencies = [
  "arrow-array",
  "arrow-schema",
@@ -4936,8 +4936,8 @@ dependencies = [
 
 [[package]]
 name = "lance-bitpacking"
-version = "11.0.0-beta.3"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
+version = "11.0.0-beta.7"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
 dependencies = [
  "arrayref",
  "crunchy",
@@ -4947,8 +4947,8 @@ dependencies = [
 
 [[package]]
 name = "lance-core"
-version = "11.0.0-beta.3"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
+version = "11.0.0-beta.7"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
 dependencies = [
  "arrow-array",
  "arrow-buffer",
@@ -4988,8 +4988,8 @@ dependencies = [
 
 [[package]]
 name = "lance-datafusion"
-version = "11.0.0-beta.3"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
+version = "11.0.0-beta.7"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
 dependencies = [
  "arrow",
  "arrow-array",
@@ -5019,8 +5019,8 @@ dependencies = [
 
 [[package]]
 name = "lance-datagen"
-version = "11.0.0-beta.3"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
+version = "11.0.0-beta.7"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
 dependencies = [
  "arrow",
  "arrow-array",
@@ -5037,8 +5037,8 @@ dependencies = [
 
 [[package]]
 name = "lance-derive"
-version = "11.0.0-beta.3"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
+version = "11.0.0-beta.7"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -5047,8 +5047,8 @@ dependencies = [
 
 [[package]]
 name = "lance-encoding"
-version = "11.0.0-beta.3"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
+version = "11.0.0-beta.7"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
 dependencies = [
  "arrow-arith",
  "arrow-array",
@@ -5082,8 +5082,8 @@ dependencies = [
 
 [[package]]
 name = "lance-file"
-version = "11.0.0-beta.3"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
+version = "11.0.0-beta.7"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
 dependencies = [
  "arrow-arith",
  "arrow-array",
@@ -5114,8 +5114,8 @@ dependencies = [
 
 [[package]]
 name = "lance-index"
-version = "11.0.0-beta.3"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
+version = "11.0.0-beta.7"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
 dependencies = [
  "arc-swap",
  "arrow",
@@ -5182,8 +5182,8 @@ dependencies = [
 
 [[package]]
 name = "lance-index-core"
-version = "11.0.0-beta.3"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
+version = "11.0.0-beta.7"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
 dependencies = [
  "arrow-array",
  "arrow-schema",
@@ -5205,8 +5205,8 @@ dependencies = [
 
 [[package]]
 name = "lance-io"
-version = "11.0.0-beta.3"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
+version = "11.0.0-beta.7"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
 dependencies = [
  "arrow",
  "arrow-array",
@@ -5242,8 +5242,8 @@ dependencies = [
 
 [[package]]
 name = "lance-linalg"
-version = "11.0.0-beta.3"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
+version = "11.0.0-beta.7"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
 dependencies = [
  "arrow-array",
  "arrow-buffer",
@@ -5259,8 +5259,8 @@ dependencies = [
 
 [[package]]
 name = "lance-namespace"
-version = "11.0.0-beta.3"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
+version = "11.0.0-beta.7"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
 dependencies = [
  "arrow",
  "async-trait",
@@ -5272,8 +5272,8 @@ dependencies = [
 
 [[package]]
 name = "lance-namespace-impls"
-version = "11.0.0-beta.3"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
+version = "11.0.0-beta.7"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
 dependencies = [
  "arrow",
  "arrow-ipc",
@@ -5326,8 +5326,8 @@ dependencies = [
 
 [[package]]
 name = "lance-select"
-version = "11.0.0-beta.3"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
+version = "11.0.0-beta.7"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
 dependencies = [
  "arrow-array",
  "arrow-buffer",
@@ -5342,8 +5342,8 @@ dependencies = [
 
 [[package]]
 name = "lance-table"
-version = "11.0.0-beta.3"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
+version = "11.0.0-beta.7"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
 dependencies = [
  "arrow",
  "arrow-array",
@@ -5383,8 +5383,8 @@ dependencies = [
 
 [[package]]
 name = "lance-testing"
-version = "11.0.0-beta.3"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
+version = "11.0.0-beta.7"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
 dependencies = [
  "arrow-array",
  "arrow-schema",
@@ -5397,8 +5397,8 @@ dependencies = [
 
 [[package]]
 name = "lance-tokenizer"
-version = "11.0.0-beta.3"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
+version = "11.0.0-beta.7"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
 dependencies = [
  "frostem",
  "icu_segmenter",
@@ -5447,7 +5447,6 @@ dependencies = [
  "datafusion-physical-plan",
  "datafusion-sql",
  "futures",
- "goosefs-sdk",
  "half",
  "hf-hub",
  "http 1.5.0",
diff --git a/Cargo.toml b/Cargo.toml
index 936660d78..33bf7e09b 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -13,20 +13,20 @@ categories = ["database-implementations"]
 rust-version = "1.91.0"
 
 [workspace.dependencies]
-lance = { "version" = "=11.0.0-beta.3", default-features = false, "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
-lance-core = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
-lance-datagen = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
-lance-file = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
-lance-io = { "version" = "=11.0.0-beta.3", default-features = false, "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
-lance-index = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
-lance-linalg = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace-impls = { "version" = "=11.0.0-beta.3", default-features = false, "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
-lance-table = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
-lance-testing = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
-lance-datafusion = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
-lance-encoding = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
-lance-arrow = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
+lance = { "version" = "=11.0.0-beta.7", default-features = false, "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
+lance-core = { "version" = "=11.0.0-beta.7", "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
+lance-datagen = { "version" = "=11.0.0-beta.7", "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
+lance-file = { "version" = "=11.0.0-beta.7", "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
+lance-io = { "version" = "=11.0.0-beta.7", default-features = false, "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
+lance-index = { "version" = "=11.0.0-beta.7", "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
+lance-linalg = { "version" = "=11.0.0-beta.7", "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace = { "version" = "=11.0.0-beta.7", "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace-impls = { "version" = "=11.0.0-beta.7", default-features = false, "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
+lance-table = { "version" = "=11.0.0-beta.7", "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
+lance-testing = { "version" = "=11.0.0-beta.7", "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
+lance-datafusion = { "version" = "=11.0.0-beta.7", "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
+lance-encoding = { "version" = "=11.0.0-beta.7", "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
+lance-arrow = { "version" = "=11.0.0-beta.7", "tag" = "v11.0.0-beta.7", "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 }
diff --git a/deny.toml b/deny.toml
index 034b48c25..d94c9d536 100644
--- a/deny.toml
+++ b/deny.toml
@@ -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" },
 ]
 
 # ---------------------------------------------------------------------------
diff --git a/java/pom.xml b/java/pom.xml
index f85b5c906..4fdf77e81 100644
--- a/java/pom.xml
+++ b/java/pom.xml
@@ -28,7 +28,7 @@
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <arrow.version>15.0.0</arrow.version>
-        <lance-core.version>11.0.0-beta.3</lance-core.version>
+        <lance-core.version>11.0.0-beta.7</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>
diff --git a/python/python/lancedb/streaming.py b/python/python/lancedb/streaming.py
index 525ed3d63..b27e606a4 100644
--- a/python/python/lancedb/streaming.py
+++ b/python/python/lancedb/streaming.py
@@ -11,6 +11,11 @@ Provides StreamingDataset, a PyTorch IterableDataset that guarantees:
 - **Resumability**: state_dict / load_state_dict capture per-split consumption
   counts so training can resume from an exact mid-epoch position even when the
   distributed topology changes between runs.
+
+Transform failures on bad rows (e.g. nulls or NaNs from incomplete data) can
+be tolerated with ``on_transform_error="skip"``; see the parameter
+documentation on StreamingDataset for how this interacts with the guarantees
+above.
 """
 
 import ctypes
@@ -22,7 +27,7 @@ import time
 from collections import deque
 from concurrent.futures import ThreadPoolExecutor
 from multiprocessing import RawArray
-from typing import Any, Callable, Iterator, Optional
+from typing import Any, Callable, Iterator, Optional, Union
 
 from torch.utils.data import IterableDataset, get_worker_info
 
@@ -127,6 +132,49 @@ class StreamingDataset(IterableDataset):
         Maximum number of transforms to run concurrently.  Must be greater
         than zero.  When ``None`` (the default), uses ``os.cpu_count()`` or 1
         when the CPU count is unavailable.
+    on_transform_error:
+        What to do when the transform raises an exception:
+
+        - ``"raise"`` (the default): the exception propagates and iteration
+          aborts.
+        - ``"skip"``: the failing rows are dropped and iteration continues.
+        - ``"warn"``: like ``"skip"``, but a warning is logged for each
+          failing batch.
+        - a callable ``handler(exc) -> bool``: called with the exception;
+          return ``True`` to skip the failing rows or ``False`` to re-raise.
+          Useful to skip only expected error types (compatible with
+          ``webdataset.handlers`` style handlers).
+
+        When a batch fails, the transform is re-invoked on each single-row
+        slice of the batch so that only the rows that actually fail are
+        dropped.  Transforms should therefore be deterministic and accept
+        batches of any size (including one row).  Skipped rows are counted in
+        ``rows_skipped``.
+
+        Skipping weakens the elastic-determinism guarantee at the end of the
+        epoch: splits that lose more rows than others run dry earlier, and
+        each rank's iterator ends at the last cycle where every split *it
+        owns* still has a row.  Because bad rows are not distributed evenly
+        across splits, this means one rank's iterator can yield noticeably
+        fewer or more steps than another rank's *in the same run* — there is
+        no cross-rank coordination that stops every rank at the same global
+        step.  This is generally safe for asynchronous or single-rank use,
+        but synchronous distributed training (e.g. ranks that call
+        ``all_reduce`` every step) can hang or deadlock if one rank's
+        iterator is exhausted while others are still stepping; callers doing
+        synchronous multi-rank training with ``on_transform_error != "raise"``
+        are responsible for their own cross-rank stopping mechanism (e.g.
+        broadcasting a stop signal on ``StopIteration``).  The final few
+        global steps can also differ across topologies (bounded by the skew
+        in bad-row counts across splits).  The sequence of samples yielded
+        from each split remains deterministic.  Mid-epoch
+        checkpoints remain exact provided the transform fails
+        deterministically; in multi-rank training each rank must save its
+        own ``state_dict`` and the states must be combined with
+        ``merge_state_dicts`` before resuming on a different topology.
+        Prefer the ``filter`` parameter when bad rows can be expressed as a
+        SQL predicate (e.g. ``"col IS NOT NULL"``) — filtering happens before
+        splits are built, so every guarantee is fully preserved.
     worker_info_override:
         If set, used in place of ``torch.utils.data.get_worker_info()`` to
         determine the DataLoader worker assignment.  Intended for unit tests
@@ -152,6 +200,7 @@ class StreamingDataset(IterableDataset):
         filter: Optional[str] = None,
         transform: Optional[Callable] = None,
         transform_parallelism: Optional[int] = None,
+        on_transform_error: Union[str, Callable[[Exception], bool]] = "raise",
         connection_factory: Optional[Callable[[str], Any]] = None,
         worker_info_override=None,
     ):
@@ -167,6 +216,13 @@ class StreamingDataset(IterableDataset):
             )
         if transform_parallelism is not None and transform_parallelism <= 0:
             raise ValueError("transform_parallelism must be greater than 0")
+        if on_transform_error not in ("raise", "skip", "warn") and not callable(
+            on_transform_error
+        ):
+            raise ValueError(
+                "on_transform_error must be 'raise', 'skip', 'warn', or a "
+                f"callable, got {on_transform_error!r}"
+            )
 
         self._table = table
         self._num_splits = num_splits
@@ -182,6 +238,7 @@ class StreamingDataset(IterableDataset):
         self._filter = filter
         self._transform = transform
         self._transform_parallelism = transform_parallelism
+        self._on_transform_error = on_transform_error
         self._connection_factory = connection_factory
         self._worker_info_override = worker_info_override
 
@@ -199,19 +256,28 @@ class StreamingDataset(IterableDataset):
         # in the main process.  RawArray is picklable via the forkserver
         # reduction protocol so it survives the dataset pickle round-trip.
         # Layout: [unscanned_rows, raw_rows, cooked_rows, consumed_rows,
-        #          bytes_loaded, fetch_time_us, transform_time_us]
-        self._worker_stats: RawArray = RawArray(ctypes.c_int64, 7)
+        #          bytes_loaded, fetch_time_us, transform_time_us,
+        #          rows_skipped]
+        self._worker_stats: RawArray = RawArray(ctypes.c_int64, 8)
 
         # Cumulative bytes of Arrow buffer data fetched across all iterations.
         self._bytes_loaded: int = 0
         # Cumulative seconds spent in LanceDB I/O and in transform functions.
         self._fetch_time: float = 0.0
         self._transform_time: float = 0.0
+        # Cumulative rows dropped by on_transform_error across all iterations.
+        self._rows_skipped: int = 0
 
         # Number of samples each split has already been consumed.  At global
         # step boundaries all splits have consumed this many samples, so a
         # single scalar captures the topology-independent checkpoint state.
         self._resume_offset: int = 0
+        # Permutation position each split has consumed through, keyed by
+        # global split index.  Equal to _resume_offset for every split unless
+        # on_transform_error skipped rows, in which case skipped positions
+        # push the watermark of the affected splits further ahead.  Splits
+        # this instance has never iterated have no entry.
+        self._resume_positions: dict[int, int] = {}
 
         # Build the permutation table once, deterministically.
         builder = permutation_builder(table)
@@ -275,6 +341,7 @@ class StreamingDataset(IterableDataset):
         # Set identity transform on each Permutation so __getitems__ returns
         # the raw RecordBatch.  Stage 2 applies the real transform.
         permutations: list[Permutation] = []
+        initial_positions: list[int] = []
         for split_idx in my_splits:
             perm = Permutation.from_tables(
                 self._table, self._perm_table, split=split_idx
@@ -282,14 +349,20 @@ class StreamingDataset(IterableDataset):
             if self._columns is not None:
                 perm = perm.select_columns(self._columns)
             perm = perm.with_transform(lambda batch: batch)
-            if self._resume_offset > 0:
-                perm = perm.with_skip(self._resume_offset)
+            start_pos = self._resume_positions.get(split_idx, self._resume_offset)
+            if start_pos > 0:
+                perm = perm.with_skip(start_pos)
+            initial_positions.append(start_pos)
             permutations.append(perm)
 
         n = len(permutations)
         split_sizes = [perm.num_rows for perm in permutations]
         initial_offset = self._resume_offset
         local_consumed = [0] * n
+        # Permutation position each split has consumed through (absolute,
+        # i.e. counted from the start of the unskipped split).  Runs ahead of
+        # initial + local_consumed when rows are skipped.
+        pos_consumed = list(initial_positions)
 
         batch_size = self._read_batch_size
         max_prefetch = self._prefetch_batches
@@ -302,12 +375,14 @@ class StreamingDataset(IterableDataset):
             self._transform if self._transform is not None else Transforms.arrow2python
         )
 
-        # Per-split pipeline state.
+        # Per-split pipeline state.  Batches are paired with the absolute
+        # permutation position of their first row so that skipped rows can be
+        # accounted for in pos_consumed.
         fetch_head = [0] * n
-        io_pending = [deque() for _ in range(n)]  # Future[RecordBatch]
-        raw_batches = [deque() for _ in range(n)]  # RecordBatch — fetched, awaiting tx
-        tx_pending = [deque() for _ in range(n)]  # Future[list[Any]]
-        cooked = [deque() for _ in range(n)]  # rows ready to yield
+        io_pending = [deque() for _ in range(n)]  # (abs_start, Future[RecordBatch])
+        raw_batches = [deque() for _ in range(n)]  # (abs_start, RecordBatch)
+        tx_pending = [deque() for _ in range(n)]  # Future[list[(abs_pos, row)]]
+        cooked = [deque() for _ in range(n)]  # (abs_pos, row) ready to yield
 
         # Limit simultaneous transforms to transform_workers across all splits.
         tx_semaphore = threading.Semaphore(transform_workers)
@@ -330,7 +405,8 @@ class StreamingDataset(IterableDataset):
             fetch_head[i] += fetch
             perm_i = permutations[i]
             indices = list(range(start, start + fetch))
-            io_pending[i].append(io_pool.submit(_io_call, perm_i, indices))
+            abs_start = initial_positions[i] + start
+            io_pending[i].append((abs_start, io_pool.submit(_io_call, perm_i, indices)))
 
         def _fill_io(i: int) -> None:
             while len(io_pending[i]) < max_prefetch and fetch_head[i] < split_sizes[i]:
@@ -338,15 +414,72 @@ class StreamingDataset(IterableDataset):
 
         def _drain_io(i: int) -> None:
             """Move completed I/O futures into raw_batches non-blockingly."""
-            while io_pending[i] and io_pending[i][0].done():
-                raw_batches[i].append(io_pending[i].popleft().result())
+            while io_pending[i] and io_pending[i][0][1].done():
+                abs_start, fut = io_pending[i].popleft()
+                raw_batches[i].append((abs_start, fut.result()))
 
         # ── Stage 2 helpers ───────────────────────────────────────────────────
 
-        def _tx_call_guarded(batch):
+        on_error = self._on_transform_error
+
+        def _should_skip(exc: Exception) -> bool:
+            if on_error == "raise":
+                return False
+            if callable(on_error):
+                return bool(on_error(exc))
+            return True  # "skip" or "warn"
+
+        def _check_row_count(rows: list, num_rows: int) -> None:
+            if len(rows) != num_rows:
+                raise ValueError(
+                    f"transform returned {len(rows)} rows for a batch of "
+                    f"{num_rows}; transforms must return exactly one output "
+                    "row per input row.  To drop bad rows, raise inside the "
+                    "transform and pass on_transform_error='skip'."
+                )
+
+        def _transform_isolated(abs_start, batch, batch_exc):
+            """Re-run the transform on single-row slices, dropping failures."""
+            out = []
+            skipped = 0
+            first_exc = None
+            for j in range(batch.num_rows):
+                try:
+                    rows = list(final_transform(batch.slice(j, 1)))
+                except Exception as exc:
+                    if not _should_skip(exc):
+                        raise
+                    skipped += 1
+                    if first_exc is None:
+                        first_exc = exc
+                    continue
+                _check_row_count(rows, 1)
+                out.append((abs_start + j, rows[0]))
+            self._rows_skipped += skipped
+            if skipped and on_error == "warn":
+                logger.warning(
+                    "Skipped %d of %d rows whose transform failed (first error: %r)",
+                    skipped,
+                    batch.num_rows,
+                    first_exc if first_exc is not None else batch_exc,
+                )
+            return out
+
+        def _transform_batch(abs_start, batch):
+            """Apply the transform, returning [(abs_pos, row), ...]."""
+            try:
+                rows = list(final_transform(batch))
+            except Exception as exc:
+                if not _should_skip(exc):
+                    raise
+                return _transform_isolated(abs_start, batch, exc)
+            _check_row_count(rows, batch.num_rows)
+            return [(abs_start + j, row) for j, row in enumerate(rows)]
+
+        def _tx_call_guarded(abs_start, batch):
             try:
                 t0 = time.perf_counter()
-                result = final_transform(batch)
+                result = _transform_batch(abs_start, batch)
                 self._transform_time += time.perf_counter() - t0
                 return result
             finally:
@@ -355,8 +488,8 @@ class StreamingDataset(IterableDataset):
         def _try_submit_tx(i: int) -> None:
             """Submit transforms for raw_batches[i] up to available capacity."""
             while raw_batches[i] and tx_semaphore.acquire(blocking=False):
-                batch = raw_batches[i].popleft()
-                tx_pending[i].append(tx_pool.submit(_tx_call_guarded, batch))
+                abs_start, batch = raw_batches[i].popleft()
+                tx_pending[i].append(tx_pool.submit(_tx_call_guarded, abs_start, batch))
 
         def _drain_tx(i: int) -> None:
             """Move completed transform futures into cooked non-blockingly."""
@@ -384,11 +517,14 @@ class StreamingDataset(IterableDataset):
                     # Acquire a transform slot (may block briefly if all
                     # transform_workers are busy with other splits).
                     tx_semaphore.acquire()
-                    batch = raw_batches[i].popleft()
-                    tx_pending[i].append(tx_pool.submit(_tx_call_guarded, batch))
+                    abs_start, batch = raw_batches[i].popleft()
+                    tx_pending[i].append(
+                        tx_pool.submit(_tx_call_guarded, abs_start, batch)
+                    )
                 elif io_pending[i]:
                     # Block on the oldest in-flight I/O fetch.
-                    raw_batches[i].append(io_pending[i].popleft().result())
+                    abs_start, fut = io_pending[i].popleft()
+                    raw_batches[i].append((abs_start, fut.result()))
                     _advance(i)
                 else:
                     break  # split exhausted
@@ -407,15 +543,28 @@ class StreamingDataset(IterableDataset):
                         _fill_io(i)
 
                     while True:
-                        # Stop when any split is exhausted (all exhaust
-                        # simultaneously: equal split sizes + round-robin).
-                        if any(local_consumed[i] >= split_sizes[i] for i in range(n)):
+                        # A cycle only runs if every split can still produce a
+                        # row.  Without skips all splits exhaust simultaneously
+                        # (equal split sizes + round-robin); when
+                        # on_transform_error drops rows a split can run dry
+                        # early, ending the epoch at the last complete cycle.
+                        # This check only sees splits owned by this rank/worker
+                        # (my_splits) — there is no cross-rank coordination, so
+                        # a different rank with fewer skipped rows keeps going;
+                        # see the on_transform_error docstring.
+                        exhausted = False
+                        for i in range(n):
+                            _ensure_cooked(i)
+                            if not cooked[i]:
+                                exhausted = True
+                                break
+                        if exhausted:
                             break
 
                         for i in range(n):
-                            _ensure_cooked(i)
-                            row = cooked[i].popleft()
+                            pos, row = cooked[i].popleft()
                             local_consumed[i] += 1
+                            pos_consumed[i] = pos + 1
                             _advance(i)
 
                             # After the last split in each cycle: update the
@@ -424,21 +573,39 @@ class StreamingDataset(IterableDataset):
                             # even when __iter__ runs in a worker process.
                             if i == n - 1:
                                 self._resume_offset = initial_offset + local_consumed[i]
+                                for j, split_idx in enumerate(my_splits):
+                                    self._resume_positions[split_idx] = pos_consumed[j]
                                 ws = self._worker_stats
                                 ws[0] = sum(
                                     split_sizes[j] - fetch_head[j] for j in range(n)
                                 )
                                 ws[1] = sum(
-                                    batch.num_rows for q in raw_batches for batch in q
+                                    batch.num_rows
+                                    for q in raw_batches
+                                    for _, batch in q
                                 )
                                 ws[2] = sum(len(q) for q in cooked)
                                 ws[3] = sum(local_consumed)
                                 ws[4] = self._bytes_loaded
                                 ws[5] = int(self._fetch_time * 1_000_000)
                                 ws[6] = int(self._transform_time * 1_000_000)
+                                ws[7] = self._rows_skipped
 
                             yield row
                 finally:
+                    # Final stats flush: the per-cycle write above never runs
+                    # when iteration ends mid-cycle (e.g. a split whose rows
+                    # were all skipped before completing a single cycle), so
+                    # counters like rows_skipped would otherwise be stale.
+                    ws = self._worker_stats
+                    ws[0] = sum(split_sizes[j] - fetch_head[j] for j in range(n))
+                    ws[1] = 0  # queue-depth properties document 0 when idle
+                    ws[2] = 0
+                    ws[3] = sum(local_consumed)
+                    ws[4] = self._bytes_loaded
+                    ws[5] = int(self._fetch_time * 1_000_000)
+                    ws[6] = int(self._transform_time * 1_000_000)
+                    ws[7] = self._rows_skipped
                     self._raw_batches_ref = None
                     self._cooked_ref = None
                     self._fetch_head_ref = None
@@ -492,7 +659,7 @@ class StreamingDataset(IterableDataset):
         batches.  Returns 0 when not iterating.
         """
         if self._raw_batches_ref is not None:
-            return sum(batch.num_rows for q in self._raw_batches_ref for batch in q)
+            return sum(batch.num_rows for q in self._raw_batches_ref for _, batch in q)
         return int(self._worker_stats[1])
 
     @property
@@ -522,6 +689,19 @@ class StreamingDataset(IterableDataset):
             )
         return int(self._worker_stats[0])
 
+    @property
+    def rows_skipped(self) -> int:
+        """Number of rows dropped because their transform raised an exception.
+
+        Only ever non-zero when ``on_transform_error`` is set to ``"skip"``,
+        ``"warn"``, or a callable that returned ``True``.  Accumulates across
+        multiple iterations of the same dataset instance and is never reset
+        automatically.
+        """
+        if self._raw_batches_ref is not None:
+            return self._rows_skipped
+        return int(self._worker_stats[7])
+
     @property
     def consumed_rows(self) -> int:
         """Number of rows already yielded to the caller across all splits.
@@ -587,12 +767,27 @@ class StreamingDataset(IterableDataset):
         every split has been consumed the same number of times (by the
         round-robin design), so the per-split count is a single uniform value
         that is identical across all ranks and DataLoader workers.
+
+        ``positions_consumed_per_split`` records how far into each split's
+        permutation iteration has advanced.  It only differs from
+        ``samples_consumed_per_split`` when ``on_transform_error`` skipped
+        rows, in which case entries are exact for the splits this instance
+        iterated and a lower bound (the sample count) for splits owned by
+        other ranks or workers.  Combine the state dicts from all ranks with
+        [merge_state_dicts][lancedb.streaming.StreamingDataset.merge_state_dicts]
+        to recover the exact value for every split before resuming on a
+        different topology.
         """
+        positions = [
+            self._resume_positions.get(split, self._resume_offset)
+            for split in range(self._num_splits)
+        ]
         return {
             "shuffle_seed": self._shuffle_seed,
             "num_splits": self._num_splits,
             "epoch": self._epoch,
             "samples_consumed_per_split": [self._resume_offset] * self._num_splits,
+            "positions_consumed_per_split": positions,
         }
 
     def load_state_dict(self, state: dict) -> None:
@@ -618,3 +813,96 @@ class StreamingDataset(IterableDataset):
             self._resume_offset = consumed[0] if consumed else 0
         else:
             self._resume_offset = int(consumed)
+        # Older checkpoints predate positions_consumed_per_split; without
+        # skipped rows positions equal sample counts, so falling back to
+        # _resume_offset (the .get default in __iter__) is exact.
+        positions = state.get("positions_consumed_per_split")
+        if positions is None:
+            self._resume_positions = {}
+        else:
+            self._resume_positions = {
+                split: int(pos) for split, pos in enumerate(positions)
+            }
+
+    @staticmethod
+    def merge_state_dicts(states: list[dict]) -> dict:
+        """Merge state dicts saved by different ranks into one exact state.
+
+        Only needed when ``on_transform_error`` skips rows in multi-rank
+        training: each rank then knows the exact permutation position only for
+        its own splits, and records a lower bound for the rest.  Because
+        exactly one rank owns each split, the elementwise maximum across all
+        ranks' ``positions_consumed_per_split`` recovers the exact position of
+        every split.  Without skipped rows every rank's state is already
+        identical and merging is a no-op.
+
+        Raises ``ValueError`` if the states are empty or were not produced by
+        the same run (mismatched seed, split count, epoch, or sample counts).
+
+        The merge is always all-to-all and topology-agnostic: collect the
+        ``state_dict()`` from every rank of the *previous* run into one list,
+        merge that whole list, and hand the identical merged result to every
+        rank of the *next* run — regardless of whether the rank count grew,
+        shrank, or stayed the same. There is no pairwise or subset merging
+        step, because each split's exact position is only known to whichever
+        rank owned that split, and the elementwise maximum needs every rank's
+        contribution to be correct.
+
+        For example, checkpointing 8 ranks and resuming on 4 (the same
+        pattern applies when growing, e.g. 4 ranks resuming on 8)::
+
+            states = [ds.state_dict() for ds in previous_run_datasets]  # 8
+            merged = StreamingDataset.merge_state_dicts(states)
+            for ds in resumed_datasets:  # now only 4 ranks
+                ds.load_state_dict(merged)  # same dict on every rank
+
+        The rank count on either side never affects the merge itself, since
+        ``merge_state_dicts`` only cares about the list of states it is
+        given.  Each split's position is recovered by elementwise maximum;
+        here rank 0 owned split 0 (and skipped two rows there) while rank 1
+        owned split 1 (and skipped one row):
+
+        >>> rank0 = {
+        ...     "shuffle_seed": 0, "num_splits": 2, "epoch": 0,
+        ...     "samples_consumed_per_split": [3, 3],
+        ...     "positions_consumed_per_split": [5, 3],
+        ... }
+        >>> rank1 = {
+        ...     "shuffle_seed": 0, "num_splits": 2, "epoch": 0,
+        ...     "samples_consumed_per_split": [3, 3],
+        ...     "positions_consumed_per_split": [3, 4],
+        ... }
+        >>> merged = StreamingDataset.merge_state_dicts([rank0, rank1])
+        >>> merged["positions_consumed_per_split"]
+        [5, 4]
+        """
+        if not states:
+            raise ValueError("merge_state_dicts requires at least one state dict")
+        first = states[0]
+        for state in states[1:]:
+            for key in ("shuffle_seed", "num_splits", "epoch"):
+                if state[key] != first[key]:
+                    raise ValueError(
+                        f"{key} mismatch across state dicts: "
+                        f"{state[key]} != {first[key]}"
+                    )
+            if (
+                state["samples_consumed_per_split"]
+                != first["samples_consumed_per_split"]
+            ):
+                raise ValueError(
+                    "samples_consumed_per_split mismatch across state dicts; "
+                    "state_dict() must be called at the same global step "
+                    "boundary on every rank"
+                )
+        merged = dict(first)
+        all_positions = [
+            state.get(
+                "positions_consumed_per_split", state["samples_consumed_per_split"]
+            )
+            for state in states
+        ]
+        merged["positions_consumed_per_split"] = [
+            max(per_split) for per_split in zip(*all_positions)
+        ]
+        return merged
diff --git a/python/python/tests/test_elastic_dataloader.py b/python/python/tests/test_elastic_dataloader.py
index 22918082b..734f835c6 100644
--- a/python/python/tests/test_elastic_dataloader.py
+++ b/python/python/tests/test_elastic_dataloader.py
@@ -1456,6 +1456,408 @@ def test_shuffle_clump_size_yields_all_rows(lance_table):
     )
 
 
+# ---------------------------------------------------------------------------
+# on_transform_error tests
+# ---------------------------------------------------------------------------
+
+
+class BadRowError(ValueError):
+    """Raised by the failing transforms below when a batch contains a bad id."""
+
+
+def _failing_transform(bad_ids: set):
+    """A transform that raises BadRowError whenever the batch has a bad id.
+
+    Raises on the full batch and on any single-row slice containing a bad id,
+    so per-row isolation drops exactly the bad rows.
+    """
+
+    def transform(batch: pa.RecordBatch) -> list:
+        ids = batch.column("id").to_pylist()
+        bad = sorted(set(ids) & bad_ids)
+        if bad:
+            raise BadRowError(f"bad ids in batch: {bad}")
+        return [{"id": i} for i in ids]
+
+    return transform
+
+
+def _sequential_split_members(table) -> list[list[int]]:
+    """Return each split's ids in yield order for shuffle=False.
+
+    With a single rank and no workers the round-robin yields one row per split
+    per cycle, so item k of a clean run belongs to split k % NUM_SPLITS.
+    """
+    ds = StreamingDataset(table, num_splits=NUM_SPLITS, shuffle=False)
+    members: list[list[int]] = [[] for _ in range(NUM_SPLITS)]
+    for k, row in enumerate(ds):
+        members[k % NUM_SPLITS].append(row["id"])
+    return members
+
+
+def test_on_transform_error_default_raises(lance_table):
+    """By default a transform exception propagates and aborts iteration."""
+    ds = StreamingDataset(
+        lance_table,
+        num_splits=NUM_SPLITS,
+        shuffle_seed=SHUFFLE_SEED,
+        transform=_failing_transform({7}),
+    )
+    with pytest.raises(BadRowError):
+        list(ds)
+
+
+def test_on_transform_error_invalid_value(lance_table):
+    with pytest.raises(ValueError, match="on_transform_error"):
+        StreamingDataset(lance_table, num_splits=NUM_SPLITS, on_transform_error="bogus")
+
+
+def test_on_transform_error_skip_drops_bad_rows(lance_table):
+    """With one bad row per split, 'skip' yields every good row exactly once
+    and counts the dropped rows in rows_skipped."""
+    members = _sequential_split_members(lance_table)
+    bad_ids = {members[i][4] for i in range(NUM_SPLITS)}
+
+    ds = StreamingDataset(
+        lance_table,
+        num_splits=NUM_SPLITS,
+        shuffle=False,
+        transform=_failing_transform(bad_ids),
+        on_transform_error="skip",
+    )
+    assert ds.rows_skipped == 0
+
+    ids = [row["id"] for row in ds]
+
+    assert sorted(ids) == sorted(set(range(NUM_ROWS)) - bad_ids)
+    assert ds.rows_skipped == NUM_SPLITS
+
+
+def test_on_transform_error_skip_uneven_ends_at_last_complete_cycle(lance_table):
+    """When one split loses more rows than the others, the epoch ends at the
+    last cycle where every split still has a row — no crash, no bad rows, and
+    every step remains one sample per split."""
+    members = _sequential_split_members(lance_table)
+    bad_ids = set(members[0][:3])  # all 3 bad rows in split 0
+
+    ds = StreamingDataset(
+        lance_table,
+        num_splits=NUM_SPLITS,
+        shuffle=False,
+        transform=_failing_transform(bad_ids),
+        on_transform_error="skip",
+    )
+    items = [row["id"] for row in ds]
+
+    rows_per_split = NUM_ROWS // NUM_SPLITS
+    expected_cycles = rows_per_split - len(bad_ids)
+    assert len(items) == expected_cycles * NUM_SPLITS
+    assert len(set(items)) == len(items), "duplicate samples yielded"
+    assert not set(items) & bad_ids, "a bad row was yielded"
+    # Split 0 contributed exactly its surviving rows, in order, one per cycle.
+    survivors = [i for i in members[0] if i not in bad_ids]
+    assert items[0::NUM_SPLITS] == survivors[:expected_cycles]
+
+
+def test_on_transform_error_warn_logs(lance_table, caplog):
+    """'warn' skips like 'skip' but logs a warning for the failing batch."""
+    members = _sequential_split_members(lance_table)
+    bad_ids = {members[i][3] for i in range(NUM_SPLITS)}
+
+    ds = StreamingDataset(
+        lance_table,
+        num_splits=NUM_SPLITS,
+        shuffle=False,
+        transform=_failing_transform(bad_ids),
+        on_transform_error="warn",
+    )
+    with caplog.at_level(logging.WARNING, logger="lancedb.streaming"):
+        items = list(ds)
+
+    assert len(items) == NUM_ROWS - NUM_SPLITS
+    assert ds.rows_skipped == NUM_SPLITS
+    assert "Skipped" in caplog.text
+    assert "BadRowError" in caplog.text
+
+
+def test_on_transform_error_callable_selective(lance_table):
+    """A callable handler can skip expected errors and re-raise the rest."""
+    members = _sequential_split_members(lance_table)
+    bad_ids = {members[i][0] for i in range(NUM_SPLITS)}
+
+    handled: list[Exception] = []
+
+    def handler(exc: Exception) -> bool:
+        handled.append(exc)
+        return isinstance(exc, BadRowError)
+
+    ds = StreamingDataset(
+        lance_table,
+        num_splits=NUM_SPLITS,
+        shuffle=False,
+        transform=_failing_transform(bad_ids),
+        on_transform_error=handler,
+    )
+    items = list(ds)
+    assert len(items) == NUM_ROWS - NUM_SPLITS
+    assert handled and all(isinstance(exc, BadRowError) for exc in handled)
+
+    def broken_transform(batch: pa.RecordBatch) -> list:
+        raise TypeError("boom")
+
+    ds2 = StreamingDataset(
+        lance_table,
+        num_splits=NUM_SPLITS,
+        shuffle=False,
+        transform=broken_transform,
+        on_transform_error=handler,
+    )
+    with pytest.raises(TypeError, match="boom"):
+        list(ds2)
+
+
+def test_transform_wrong_row_count_raises(lance_table):
+    """A transform that returns the wrong number of rows is an error even with
+    on_transform_error='skip' — silent shrinkage would corrupt accounting."""
+
+    def drops_rows(batch: pa.RecordBatch) -> list:
+        return batch.column("id").to_pylist()[:-1]
+
+    ds = StreamingDataset(
+        lance_table,
+        num_splits=NUM_SPLITS,
+        shuffle_seed=SHUFFLE_SEED,
+        transform=drops_rows,
+        on_transform_error="skip",
+    )
+    with pytest.raises(ValueError, match="one output row per input row"):
+        list(ds)
+
+
+def test_skip_deterministic_across_runs(lance_table):
+    """With a fixed seed, skipping produces the identical sample sequence on
+    every run — skips are data-dependent, not run-dependent."""
+    bad_ids = {5, 17, 46}
+
+    def run() -> tuple[list[int], int]:
+        ds = StreamingDataset(
+            lance_table,
+            num_splits=NUM_SPLITS,
+            shuffle_seed=SHUFFLE_SEED,
+            transform=_failing_transform(bad_ids),
+            on_transform_error="skip",
+        )
+        return [row["id"] for row in ds], ds.rows_skipped
+
+    ids_a, skipped_a = run()
+    ids_b, skipped_b = run()
+    assert ids_a == ids_b
+    assert skipped_a == skipped_b
+    assert not set(ids_a) & bad_ids
+
+
+def test_skip_elastic_det_across_world_sizes(lance_table):
+    """With equal bad-row counts per split, skipping preserves the full
+    elastic-determinism guarantee: identical global batches at every step for
+    every compatible world_size."""
+    members = _sequential_split_members(lance_table)
+    bad_ids = {members[i][6] for i in range(NUM_SPLITS)}
+
+    def collect(world_size: int) -> list[frozenset[int]]:
+        micro = GLOBAL_BATCH_SIZE // world_size
+        iters = [
+            iter(
+                StreamingDataset(
+                    lance_table,
+                    num_splits=NUM_SPLITS,
+                    shuffle=False,
+                    rank=rank,
+                    world_size=world_size,
+                    transform=_failing_transform(bad_ids),
+                    on_transform_error="skip",
+                )
+            )
+            for rank in range(world_size)
+        ]
+        _STOP = object()
+        batches: list[frozenset[int]] = []
+        while True:
+            step_samples: set[int] = set()
+            exhausted = 0
+            for it in iters:
+                for _ in range(micro):
+                    val = next(it, _STOP)
+                    if val is _STOP:
+                        exhausted += 1
+                        break
+                    step_samples.add(val["id"])
+            if exhausted == len(iters):
+                break
+            assert exhausted == 0, (
+                "Rank iterators exhausted at different steps despite equal "
+                "bad-row counts per split"
+            )
+            batches.append(frozenset(step_samples))
+        return batches
+
+    reference = collect(1)
+    assert len(reference) == NUM_ROWS // NUM_SPLITS - 1
+    for ws in (2, 3, 4):
+        assert collect(ws) == reference, f"world_size={ws} diverged"
+
+
+def test_resumability_with_skips_same_topology(lance_table):
+    """Checkpointing mid-epoch with skipped rows resumes exactly: no sample
+    repeated, no sample lost, skipped rows stay skipped."""
+    members = _sequential_split_members(lance_table)
+    # Uneven skips: positions diverge across splits (2 bad in split 0, 1 in
+    # split 5), which only a position-based checkpoint can resume exactly.
+    bad_ids = {members[0][2], members[0][3], members[5][7]}
+    kwargs = dict(
+        num_splits=NUM_SPLITS,
+        shuffle=False,
+        transform=_failing_transform(bad_ids),
+        on_transform_error="skip",
+    )
+
+    reference = [row["id"] for row in StreamingDataset(lance_table, **kwargs)]
+    rows_per_split = NUM_ROWS // NUM_SPLITS
+    assert len(reference) == (rows_per_split - 2) * NUM_SPLITS
+
+    steps = 3
+    ds = StreamingDataset(lance_table, **kwargs)
+    it = iter(ds)
+    consumed = [next(it)["id"] for _ in range(steps * NUM_SPLITS)]
+    checkpoint = ds.state_dict()
+    it.close()
+
+    # Split 0 skipped positions 2 and 3 within its first 3 yields; split 5's
+    # bad row is beyond the checkpoint.  Everything else is at 3 = the sample
+    # count.
+    positions = checkpoint["positions_consumed_per_split"]
+    assert positions[0] == 5
+    assert positions[1:] == [3] * (NUM_SPLITS - 1)
+    assert checkpoint["samples_consumed_per_split"] == [3] * NUM_SPLITS
+
+    ds2 = StreamingDataset(lance_table, **kwargs)
+    ds2.load_state_dict(checkpoint)
+    resumed = [row["id"] for row in ds2]
+
+    assert consumed == reference[: steps * NUM_SPLITS]
+    assert resumed == reference[steps * NUM_SPLITS :]
+
+
+def test_resumability_with_skips_elastic_merge(lance_table):
+    """Elastic resume with skips: each rank's checkpoint knows exact positions
+    only for its own splits; merge_state_dicts recovers the global state, and
+    a run on a different world_size continues exactly."""
+    members = _sequential_split_members(lance_table)
+    # Bad rows early in split 0 (rank 0) and split 6 (rank 1 of a ws=2 run) so
+    # both ranks' position vectors diverge before the checkpoint.
+    bad_ids = {members[0][0], members[0][2], members[6][1]}
+    kwargs = dict(
+        num_splits=NUM_SPLITS,
+        shuffle=False,
+        transform=_failing_transform(bad_ids),
+        on_transform_error="skip",
+    )
+
+    reference = [row["id"] for row in StreamingDataset(lance_table, **kwargs)]
+
+    steps = 3
+    world_size = 2
+    micro = GLOBAL_BATCH_SIZE // world_size
+    datasets = [
+        StreamingDataset(lance_table, rank=rank, world_size=world_size, **kwargs)
+        for rank in range(world_size)
+    ]
+    iters = [iter(ds) for ds in datasets]
+    seen: list[frozenset[int]] = []
+    for _ in range(steps):
+        step_samples = set()
+        for it in iters:
+            for _ in range(micro):
+                step_samples.add(next(it)["id"])
+        seen.append(frozenset(step_samples))
+    states = [ds.state_dict() for ds in datasets]
+    for it in iters:
+        it.close()
+
+    merged = StreamingDataset.merge_state_dicts(states)
+    expected_positions = [3] * NUM_SPLITS
+    expected_positions[0] = 5  # skipped positions 0 and 2
+    expected_positions[6] = 4  # skipped position 1
+    assert merged["positions_consumed_per_split"] == expected_positions
+
+    # The first 3 global batches match the world_size=1 reference.
+    ref_batches = [
+        frozenset(reference[s * NUM_SPLITS : (s + 1) * NUM_SPLITS])
+        for s in range(len(reference) // NUM_SPLITS)
+    ]
+    assert seen == ref_batches[:steps]
+
+    # Resume on world_size=1 from the merged state.
+    ds_resume = StreamingDataset(lance_table, **kwargs)
+    ds_resume.load_state_dict(merged)
+    resumed = [row["id"] for row in ds_resume]
+    assert resumed == reference[steps * NUM_SPLITS :]
+
+
+def test_rows_skipped_flushed_when_split_entirely_bad(lance_table):
+    """A split whose rows all fail never completes a cycle, so the epoch ends
+    immediately — but rows_skipped must still report the drops after the
+    iterator exits (the shared-memory counter is flushed on exhaustion)."""
+    members = _sequential_split_members(lance_table)
+    bad_ids = set(members[0])  # every row of split 0 is bad
+
+    ds = StreamingDataset(
+        lance_table,
+        num_splits=NUM_SPLITS,
+        shuffle=False,
+        transform=_failing_transform(bad_ids),
+        on_transform_error="skip",
+    )
+    assert list(ds) == []
+    assert ds.rows_skipped == len(bad_ids)
+
+
+def test_merge_state_dicts_validates_consistency(lance_table):
+    ds = StreamingDataset(lance_table, num_splits=NUM_SPLITS, shuffle_seed=SHUFFLE_SEED)
+    state = ds.state_dict()
+    other = dict(state, shuffle_seed=SHUFFLE_SEED + 1)
+    with pytest.raises(ValueError, match="shuffle_seed mismatch"):
+        StreamingDataset.merge_state_dicts([state, other])
+    with pytest.raises(ValueError, match="at least one"):
+        StreamingDataset.merge_state_dicts([])
+
+
+def test_load_state_dict_without_positions_key(lance_table):
+    """Checkpoints from before positions_consumed_per_split existed still
+    resume exactly (positions equal sample counts when nothing is skipped)."""
+    reference = [
+        row["id"]
+        for row in StreamingDataset(
+            lance_table, num_splits=NUM_SPLITS, shuffle_seed=SHUFFLE_SEED
+        )
+    ]
+
+    steps = 4
+    ds = StreamingDataset(lance_table, num_splits=NUM_SPLITS, shuffle_seed=SHUFFLE_SEED)
+    it = iter(ds)
+    for _ in range(steps * NUM_SPLITS):
+        next(it)
+    checkpoint = ds.state_dict()
+    it.close()
+    del checkpoint["positions_consumed_per_split"]
+
+    ds2 = StreamingDataset(
+        lance_table, num_splits=NUM_SPLITS, shuffle_seed=SHUFFLE_SEED
+    )
+    ds2.load_state_dict(checkpoint)
+    resumed = [row["id"] for row in ds2]
+    assert resumed == reference[steps * NUM_SPLITS :]
+
+
 def test_num_splits_defaults_to_world_size(lance_table):
     """Omitting num_splits gives world_size splits (one per rank)."""
     ds = StreamingDataset(
diff --git a/python/src/index.rs b/python/src/index.rs
index 8c81dcecf..dd362373e 100644
--- a/python/src/index.rs
+++ b/python/src/index.rs
@@ -289,7 +289,7 @@ struct IvfHnswFlatParams {
     target_partition_size: Option<u32>,
 }
 
-#[pyclass(get_all)]
+#[pyclass(module = "lancedb._lancedb", get_all)]
 /// A description of an index currently configured on a column
 pub struct IndexConfig {
     /// The type of the index
diff --git a/python/src/session.rs b/python/src/session.rs
index 891e61e44..4d58dd269 100644
--- a/python/src/session.rs
+++ b/python/src/session.rs
@@ -11,7 +11,7 @@ use pyo3::{PyResult, pyclass, pymethods};
 /// Sessions allow you to configure cache sizes for index and metadata caches,
 /// which can significantly impact memory use and performance. They can
 /// also be re-used across multiple connections to share the same cache state.
-#[pyclass(from_py_object)]
+#[pyclass(module = "lancedb._lancedb", from_py_object)]
 #[derive(Clone)]
 pub struct Session {
     pub(crate) inner: Arc<LanceSession>,
diff --git a/python/src/table.rs b/python/src/table.rs
index 20a93556f..cae6b5d9a 100644
--- a/python/src/table.rs
+++ b/python/src/table.rs
@@ -579,7 +579,7 @@ impl PyBlobFile {
     }
 }
 
-#[pyclass(get_all, from_py_object)]
+#[pyclass(module = "lancedb._lancedb", get_all, from_py_object)]
 #[derive(Clone, Debug)]
 pub struct FtsToken {
     pub text: String,
diff --git a/rust/lancedb/Cargo.toml b/rust/lancedb/Cargo.toml
index 1bc31f82b..89bee920e 100644
--- a/rust/lancedb/Cargo.toml
+++ b/rust/lancedb/Cargo.toml
@@ -49,8 +49,6 @@ lance-namespace = { workspace = true }
 lance-namespace-impls = { workspace = true }
 metrics = { workspace = true, optional = true }
 metrics-util = { workspace = true, optional = true }
-# Pin the GooseFS SDK to the version required by Lance's OpenDAL dependency.
-goosefs-sdk = { version = "=0.1.9", optional = true }
 moka = { workspace = true }
 pin-project = { workspace = true }
 tokio = { version = "1.23", features = ["rt-multi-thread", "sync"] }
@@ -141,7 +139,6 @@ azure = [
 ]
 cos = ["lance/tencent", "lance-io/tencent"]
 goosefs = [
-    "dep:goosefs-sdk",
     "lance/goosefs",
     "lance-io/goosefs",
     "lance-namespace-impls/dir-goosefs",
@@ -196,6 +193,9 @@ required-features = ["bedrock"]
 [[example]]
 name = "bench_streaming_dataloader"
 
+[[example]]
+name = "bench_open_missing_table"
+
 [[example]]
 name = "simple"
 
diff --git a/rust/lancedb/examples/bench_open_missing_table.rs b/rust/lancedb/examples/bench_open_missing_table.rs
new file mode 100644
index 000000000..8e6b16e11
--- /dev/null
+++ b/rust/lancedb/examples/bench_open_missing_table.rs
@@ -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(())
+}
diff --git a/rust/lancedb/src/blob.rs b/rust/lancedb/src/blob.rs
index e1c18dd84..d59123ec3 100644
--- a/rust/lancedb/src/blob.rs
+++ b/rust/lancedb/src/blob.rs
@@ -17,7 +17,7 @@ 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_file::version::LanceFileVersion;
+use lance_file::version::{ConcreteFileVersion, LanceFileVersion};
 use lance_io::object_store::ObjectStore;
 use object_store::path::Path;
 
@@ -333,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);
     }
 }
@@ -499,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
         );
     }
 
@@ -512,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
         );
     }
 
diff --git a/rust/lancedb/src/connection.rs b/rust/lancedb/src/connection.rs
index dd53a2d2e..1f2708d4e 100644
--- a/rust/lancedb/src/connection.rs
+++ b/rust/lancedb/src/connection.rs
@@ -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,10 +461,9 @@ impl Connection {
     ///
     /// # Returns
     /// Created [`TableRef`], or [`Error::TableNotFound`] if the table does not exist.
-    /// If the table's storage is present but holds no readable dataset (for example a
-    /// `<name>.lance` directory left behind by an interrupted drop and re-create, which
-    /// [`Self::table_names`] still lists) this returns [`Error::TableCorrupted`]
-    /// instead.
+    /// 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(),
diff --git a/rust/lancedb/src/connection/create_table.rs b/rust/lancedb/src/connection/create_table.rs
index b10141beb..39cc82ec0 100644
--- a/rust/lancedb/src/connection/create_table.rs
+++ b/rust/lancedb/src/connection/create_table.rs
@@ -438,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]
diff --git a/rust/lancedb/src/database/listing.rs b/rust/lancedb/src/database/listing.rs
index 6a22e98a8..8b7251525 100644
--- a/rust/lancedb/src/database/listing.rs
+++ b/rust/lancedb/src/database/listing.rs
@@ -1326,6 +1326,7 @@ impl Database for ListingDatabase {
 mod tests {
     use super::*;
     use crate::Table;
+    use crate::arrow::{SendableRecordBatchStream, SimpleRecordBatchStream};
     use crate::connection::ConnectRequest;
     use crate::data::scannable::Scannable;
     use crate::database::{CreateTableMode, CreateTableRequest};
@@ -1333,11 +1334,15 @@ mod tests {
     use crate::query::QueryRequest;
     use crate::table::{AnyQuery, WriteOptions};
     use arrow_array::{Int32Array, RecordBatch, StringArray};
-    use arrow_schema::{DataType, Field, Schema};
-    use futures::TryStreamExt;
+    use arrow_schema::{DataType, Field, Schema, SchemaRef};
+    use futures::{TryStreamExt, stream::once};
     use std::path::PathBuf;
+    use std::sync::Arc;
     use std::sync::atomic::{AtomicUsize, Ordering};
+    use std::time::Duration;
     use tempfile::tempdir;
+    use tokio::sync::Barrier;
+    use tokio::time::timeout;
 
     #[derive(Debug)]
     struct PassthroughStoreWrapper(Arc<AtomicUsize>);
@@ -1375,6 +1380,114 @@ mod tests {
         (tempdir, db)
     }
 
+    struct BarrierScannable {
+        batch: RecordBatch,
+        barrier: Arc<Barrier>,
+    }
+
+    impl Scannable for BarrierScannable {
+        fn schema(&self) -> SchemaRef {
+            self.batch.schema()
+        }
+
+        fn scan_as_stream(&mut self) -> SendableRecordBatchStream {
+            let batch = self.batch.clone();
+            let schema = batch.schema();
+            let barrier = self.barrier.clone();
+            Box::pin(SimpleRecordBatchStream {
+                schema,
+                stream: once(async move {
+                    barrier.wait().await;
+                    Ok(batch)
+                }),
+            })
+        }
+    }
+
+    fn create_request(name: &str, data: Box<dyn Scannable>) -> CreateTableRequest {
+        CreateTableRequest {
+            name: name.to_string(),
+            namespace_path: vec![],
+            data,
+            mode: CreateTableMode::Create,
+            write_options: Default::default(),
+            location: None,
+            namespace_client: None,
+        }
+    }
+
+    #[tokio::test]
+    async fn test_create_ignores_uncommitted_storage_without_manifest() {
+        let (tmp_dir, db) = setup_database().await;
+        let data_dir = tmp_dir.path().join("test.lance/data");
+        std::fs::create_dir_all(&data_dir).unwrap();
+        std::fs::write(data_dir.join("orphan.lance"), b"uncommitted").unwrap();
+
+        let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
+        let batch =
+            RecordBatch::try_new(schema, vec![Arc::new(Int32Array::from(vec![1]))]).unwrap();
+
+        let table = db
+            .create_table(create_request("test", Box::new(batch)))
+            .await
+            .unwrap();
+        assert_eq!(table.count_rows(None).await.unwrap(), 1);
+    }
+
+    #[tokio::test]
+    async fn test_concurrent_create_is_arbitrated_by_manifest_commit() {
+        let uri = format!("memory:///concurrent-create-{}", uuid::Uuid::new_v4());
+        let db = crate::connect(&uri).execute().await.unwrap();
+        let store: Arc<dyn object_store::ObjectStore> =
+            Arc::new(object_store::memory::InMemory::new());
+        let table_url = url::Url::parse("memory:///database/test.lance").unwrap();
+        let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
+        let batch =
+            RecordBatch::try_new(schema, vec![Arc::new(Int32Array::from(vec![1]))]).unwrap();
+        let barrier = Arc::new(Barrier::new(2));
+
+        #[allow(deprecated)]
+        let request = |batch, barrier| {
+            let mut request = create_request("test", Box::new(BarrierScannable { batch, barrier }));
+            request.write_options = WriteOptions {
+                lance_write_params: Some(lance::dataset::WriteParams {
+                    store_params: Some(ObjectStoreParams {
+                        object_store: Some((store.clone(), table_url.clone())),
+                        ..Default::default()
+                    }),
+                    commit_handler: Some(Arc::new(
+                        lance_table::io::commit::ConditionalPutCommitHandler,
+                    )),
+                    ..Default::default()
+                }),
+            };
+            request
+        };
+
+        let left = db
+            .database()
+            .create_table(request(batch.clone(), barrier.clone()));
+        let right = db.database().create_table(request(batch, barrier));
+        let (left, right) = timeout(Duration::from_secs(30), async { tokio::join!(left, right) })
+            .await
+            .expect("concurrent creates deadlocked");
+
+        let results = [left, right];
+        assert_eq!(
+            results.iter().filter(|result| result.is_ok()).count(),
+            1,
+            "expected one successful create, got {results:?}"
+        );
+        assert_eq!(
+            results
+                .iter()
+                .filter(|result| matches!(result, Err(Error::TableAlreadyExists { .. })))
+                .count(),
+            1,
+            "expected one manifest conflict, got {results:?}"
+        );
+    }
+
     #[tokio::test]
     async fn test_listing_database_root_ops_do_not_create_manifest() {
         let tempdir = tempdir().unwrap();
diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs
index a4c5ea196..e05cd8bad 100644
--- a/rust/lancedb/src/table.rs
+++ b/rust/lancedb/src/table.rs
@@ -50,7 +50,6 @@ use crate::DistanceType;
 use crate::blob::BlobRangeRequest;
 use crate::data::scannable::{PeekedScannable, Scannable, estimate_write_partitions};
 use crate::database::Database;
-use crate::database::listing::LANCE_FILE_EXTENSION;
 use crate::database::read_freshness::TableFreshness;
 use crate::embeddings::{EmbeddingDefinition, EmbeddingRegistry, MemoryRegistry};
 use crate::error::{Error, Result};
@@ -152,55 +151,6 @@ pub(crate) fn map_namespace_lance_error(err: lance::Error, table_name: &str) ->
     }
 }
 
-/// Map a `lance::Error::DatasetNotFound` for the table at `uri` into a `lancedb::Error`.
-///
-/// Lance reports "there is nothing at this location" and "there is a table directory
-/// here but nothing loadable inside it" with the same error. Only the first is a
-/// `TableNotFound`: a `<name>.lance` directory left behind by an interrupted drop and
-/// re-create is still reported by `Connection::table_names`, so callers need to be able
-/// to tell "never existed" from "exists but is broken".
-///
-/// See <https://github.com/lancedb/lancedb/issues/3127>.
-async fn map_dataset_not_found(
-    uri: &str,
-    name: &str,
-    params: ReadParams,
-    err: lance::Error,
-) -> Error {
-    let name = name.to_string();
-    let source = Box::new(err);
-    if table_dir_exists(uri, params).await.unwrap_or(false) {
-        Error::TableCorrupted { name, source }
-    } else {
-        Error::TableNotFound { name, source }
-    }
-}
-
-/// Whether a table directory is present at `uri`, even though no dataset could be
-/// loaded from it.
-///
-/// This looks for a `<name>.lance` entry in the parent directory, which is exactly what
-/// `ListingDatabase::table_names` lists, so the two APIs agree on whether a table is
-/// present. Probing `uri` itself would not work: object stores have no empty
-/// directories to probe, and on a local filesystem the interesting case is precisely an
-/// empty directory.
-async fn table_dir_exists(uri: &str, params: ReadParams) -> Result<bool> {
-    let (object_store, path, _) = DatasetBuilder::from_uri(uri)
-        .with_read_params(params)
-        .build_object_store()
-        .await?;
-    // Only `*.lance` entries are ever reported as tables, so nothing else can produce
-    // the list-then-open mismatch this guards against.
-    if path.extension() != Some(LANCE_FILE_EXTENSION) {
-        return Ok(false);
-    }
-    let (Some(parent), Some(dir_name)) = (path.parent(), path.filename()) else {
-        return Ok(false);
-    };
-    let entries = object_store.read_dir(parent).await?;
-    Ok(entries.iter().any(|entry| entry.as_str() == dir_name))
-}
-
 /// Defines the type of column
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub enum ColumnKind {
@@ -2429,8 +2379,6 @@ impl NativeTable {
             None => false,
         };
 
-        // Kept so that a `DatasetNotFound` can be re-checked against storage below.
-        let recovery_params = params.clone();
         let mut builder = DatasetBuilder::from_uri(uri).with_read_params(params);
 
         // Set up commit handler when managed_versioning is enabled
@@ -2449,7 +2397,12 @@ impl NativeTable {
         let dataset = match builder.load().await {
             Ok(dataset) => dataset,
             Err(e @ lance::Error::DatasetNotFound { .. }) => {
-                return Err(map_dataset_not_found(uri, name, recovery_params, e).await);
+                // The manifest load is the existence check. A physical prefix may be
+                // from a concurrent or abandoned create, so it cannot refine this error.
+                return Err(Error::TableNotFound {
+                    name: name.to_string(),
+                    source: Box::new(e),
+                });
             }
             Err(e) => return Err(e.into()),
         };
@@ -3736,7 +3689,7 @@ pub struct FragmentSummaryStats {
 #[cfg(test)]
 #[allow(deprecated)]
 mod tests {
-    use std::sync::atomic::{AtomicBool, Ordering};
+    use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
     use std::sync::{Arc, Mutex};
     use std::time::Duration;
 
@@ -3819,73 +3772,50 @@ mod tests {
         );
     }
 
-    /// Write a table and then break it, leaving the `<name>.lance` directory in place.
-    ///
-    /// `remove_all` reproduces an interrupted drop + re-create (the directory is left
-    /// empty); otherwise only the manifests are removed, leaving the data files behind.
-    async fn write_then_corrupt_table(dir: &std::path::Path, remove_all: bool) -> String {
-        let dataset_path = dir.join("test.lance");
-        let uri = dataset_path.to_str().unwrap().to_string();
-
-        let batch = make_test_batches();
-        let reader = RecordBatchIterator::new(vec![Ok(batch.clone())], batch.schema());
-        Dataset::write(reader, &uri, None).await.unwrap();
-
-        if remove_all {
-            for entry in std::fs::read_dir(&dataset_path).unwrap() {
-                let entry = entry.unwrap();
-                if entry.file_type().unwrap().is_dir() {
-                    std::fs::remove_dir_all(entry.path()).unwrap();
-                } else {
-                    std::fs::remove_file(entry.path()).unwrap();
-                }
-            }
-            assert_eq!(std::fs::read_dir(&dataset_path).unwrap().count(), 0);
-        } else {
-            let versions = dataset_path.join("_versions");
-            assert!(versions.is_dir(), "expected manifests under {versions:?}");
-            std::fs::remove_dir_all(&versions).unwrap();
-            assert!(std::fs::read_dir(&dataset_path).unwrap().count() > 0);
-        }
-
-        uri
-    }
-
     #[tokio::test]
-    async fn test_open_corrupt_empty_dir() {
+    async fn test_open_not_found_when_empty_directory_exists() {
         let tmp_dir = tempdir().unwrap();
-        let uri = write_then_corrupt_table(tmp_dir.path(), true).await;
+        let dataset_path = tmp_dir.path().join("test.lance");
+        std::fs::create_dir(&dataset_path).unwrap();
 
-        let err = NativeTable::open(&uri).await.unwrap_err();
+        let err = NativeTable::open(dataset_path.to_str().unwrap())
+            .await
+            .unwrap_err();
         assert!(
-            matches!(&err, Error::TableCorrupted { name, .. } if name == "test"),
+            matches!(&err, Error::TableNotFound { name, .. } if name == "test"),
             "got {err:?}"
         );
     }
 
     #[tokio::test]
-    async fn test_open_corrupt_missing_manifest() {
+    async fn test_open_not_found_when_only_uncommitted_storage_exists() {
         let tmp_dir = tempdir().unwrap();
-        let uri = write_then_corrupt_table(tmp_dir.path(), false).await;
+        let dataset_path = tmp_dir.path().join("test.lance");
+        let data_dir = dataset_path.join("data");
+        std::fs::create_dir_all(&data_dir).unwrap();
+        std::fs::write(data_dir.join("orphan.lance"), b"uncommitted").unwrap();
 
-        let err = NativeTable::open(&uri).await.unwrap_err();
+        let err = NativeTable::open(dataset_path.to_str().unwrap())
+            .await
+            .unwrap_err();
         assert!(
-            matches!(&err, Error::TableCorrupted { name, .. } if name == "test"),
+            matches!(&err, Error::TableNotFound { name, .. } if name == "test"),
             "got {err:?}"
         );
     }
 
-    /// A table listed by `table_names()` must not be reported as missing by
-    /// `open_table()`. See <https://github.com/lancedb/lancedb/issues/3127>.
+    /// Listing databases discover physical `*.lance` entries. That snapshot is not an
+    /// authoritative table-existence check: only a committed manifest makes a table
+    /// openable, and the entry could also be concurrently created or dropped.
     #[tokio::test]
-    async fn test_open_table_corrupt_is_still_listed() {
+    async fn test_table_names_may_include_uncommitted_storage() {
         let tmp_dir = tempdir().unwrap();
         let db = connect(tmp_dir.path().to_str().unwrap())
             .execute()
             .await
             .unwrap();
 
-        write_then_corrupt_table(tmp_dir.path(), true).await;
+        std::fs::create_dir(tmp_dir.path().join("test.lance")).unwrap();
 
         assert_eq!(
             db.table_names().execute().await.unwrap(),
@@ -3893,12 +3823,177 @@ mod tests {
         );
         let err = db.open_table("test").execute().await.unwrap_err();
         assert!(
-            matches!(&err, Error::TableCorrupted { name, .. } if name == "test"),
+            matches!(&err, Error::TableNotFound { name, .. } if name == "test"),
+            "physical storage without a committed manifest is not a table: {err:?}"
+        );
+    }
+
+    #[derive(Debug)]
+    struct ParentListGuardStore {
+        inner: Arc<dyn object_store::ObjectStore>,
+        parent: object_store::path::Path,
+        parent_list_calls: Arc<AtomicUsize>,
+    }
+
+    impl std::fmt::Display for ParentListGuardStore {
+        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+            f.write_str("ParentListGuardStore")
+        }
+    }
+
+    #[async_trait::async_trait]
+    #[deny(clippy::missing_trait_methods)]
+    impl object_store::ObjectStore for ParentListGuardStore {
+        async fn put_opts(
+            &self,
+            location: &object_store::path::Path,
+            payload: object_store::PutPayload,
+            opts: object_store::PutOptions,
+        ) -> object_store::Result<object_store::PutResult> {
+            self.inner.put_opts(location, payload, opts).await
+        }
+
+        async fn put_multipart_opts(
+            &self,
+            location: &object_store::path::Path,
+            opts: object_store::PutMultipartOptions,
+        ) -> object_store::Result<Box<dyn object_store::MultipartUpload>> {
+            self.inner.put_multipart_opts(location, opts).await
+        }
+
+        async fn get_opts(
+            &self,
+            location: &object_store::path::Path,
+            options: object_store::GetOptions,
+        ) -> object_store::Result<object_store::GetResult> {
+            self.inner.get_opts(location, options).await
+        }
+
+        async fn get_ranges(
+            &self,
+            location: &object_store::path::Path,
+            ranges: &[std::ops::Range<u64>],
+        ) -> object_store::Result<Vec<bytes::Bytes>> {
+            self.inner.get_ranges(location, ranges).await
+        }
+
+        fn delete_stream(
+            &self,
+            locations: futures::stream::BoxStream<
+                'static,
+                object_store::Result<object_store::path::Path>,
+            >,
+        ) -> futures::stream::BoxStream<'static, object_store::Result<object_store::path::Path>>
+        {
+            self.inner.delete_stream(locations)
+        }
+
+        fn list(
+            &self,
+            prefix: Option<&object_store::path::Path>,
+        ) -> futures::stream::BoxStream<'static, object_store::Result<object_store::ObjectMeta>>
+        {
+            if prefix == Some(&self.parent) {
+                self.parent_list_calls.fetch_add(1, Ordering::Relaxed);
+            }
+            self.inner.list(prefix)
+        }
+
+        fn list_with_offset(
+            &self,
+            prefix: Option<&object_store::path::Path>,
+            offset: &object_store::path::Path,
+        ) -> futures::stream::BoxStream<'static, object_store::Result<object_store::ObjectMeta>>
+        {
+            if prefix == Some(&self.parent) {
+                self.parent_list_calls.fetch_add(1, Ordering::Relaxed);
+            }
+            self.inner.list_with_offset(prefix, offset)
+        }
+
+        async fn list_with_delimiter(
+            &self,
+            prefix: Option<&object_store::path::Path>,
+        ) -> object_store::Result<object_store::ListResult> {
+            if prefix == Some(&self.parent) {
+                self.parent_list_calls.fetch_add(1, Ordering::Relaxed);
+            }
+            self.inner.list_with_delimiter(prefix).await
+        }
+
+        async fn copy_opts(
+            &self,
+            from: &object_store::path::Path,
+            to: &object_store::path::Path,
+            options: object_store::CopyOptions,
+        ) -> object_store::Result<()> {
+            self.inner.copy_opts(from, to, options).await
+        }
+
+        async fn rename_opts(
+            &self,
+            from: &object_store::path::Path,
+            to: &object_store::path::Path,
+            options: object_store::RenameOptions,
+        ) -> object_store::Result<()> {
+            self.inner.rename_opts(from, to, options).await
+        }
+    }
+
+    #[derive(Debug)]
+    struct ParentListGuardWrapper {
+        parent_list_calls: Arc<AtomicUsize>,
+    }
+
+    impl WrappingObjectStore for ParentListGuardWrapper {
+        fn wrap(
+            &self,
+            _store_prefix: &str,
+            inner: Arc<dyn object_store::ObjectStore>,
+        ) -> Arc<dyn object_store::ObjectStore> {
+            Arc::new(ParentListGuardStore {
+                inner,
+                parent: object_store::path::Path::from("database"),
+                parent_list_calls: self.parent_list_calls.clone(),
+            })
+        }
+    }
+
+    #[tokio::test]
+    async fn test_open_missing_never_lists_database_parent() {
+        let parent_list_calls = Arc::new(AtomicUsize::new(0));
+        let params = ReadParams {
+            store_options: Some(ObjectStoreParams {
+                object_store_wrapper: Some(Arc::new(ParentListGuardWrapper {
+                    parent_list_calls: parent_list_calls.clone(),
+                })),
+                ..Default::default()
+            }),
+            ..Default::default()
+        };
+
+        let err = NativeTable::open_with_params(
+            "memory:///database/missing.lance",
+            "missing",
+            Vec::new(),
+            None,
+            Some(params),
+            None,
+            None,
+            HashSet::new(),
+            None,
+        )
+        .await
+        .unwrap_err();
+
+        assert!(
+            matches!(&err, Error::TableNotFound { name, .. } if name == "missing"),
             "got {err:?}"
         );
-        assert!(
-            err.to_string().contains("exists but could not be loaded"),
-            "got {err}"
+        assert_eq!(
+            parent_list_calls.load(Ordering::Relaxed),
+            0,
+            "opening one missing table must not enumerate sibling tables"
         );
     }
 
@@ -5428,7 +5523,7 @@ mod tests {
     pub async fn test_stats_includes_index_and_overlay_files() {
         use lance::dataset::WriteDestination;
         use lance::dataset::transaction::{DataOverlayGroup, Operation};
-        use lance_file::version::{ConcreteFileVersion, LanceFileVersion};
+        use lance_file::version::stable_file_version;
         use lance_file::writer::FileWriterOptions;
         use lance_io::utils::CachedFileSize;
         use lance_table::format::DataFile;
@@ -5494,7 +5589,7 @@ mod tests {
         let fragment_id = dataset.get_fragments()[0].id() as u64;
         let foo_field_id = dataset.schema().field("foo").unwrap().id;
         let overlay_schema = dataset.schema().project_by_ids(&[foo_field_id], true);
-        let file_version = ConcreteFileVersion::from(LanceFileVersion::Stable);
+        let file_version = stable_file_version();
 
         let filename = "overlay.lance".to_string();
         let store = dataset.object_store(None).await.unwrap();
diff --git a/rust/lancedb/tests/blob_integration.rs b/rust/lancedb/tests/blob_integration.rs
index 77d49abd9..b92f961f4 100644
--- a/rust/lancedb/tests/blob_integration.rs
+++ b/rust/lancedb/tests/blob_integration.rs
@@ -10,7 +10,7 @@ use arrow_array::{
 use arrow_schema::{DataType, Field, Fields, Schema};
 use futures::TryStreamExt;
 use lance::Dataset;
-use lance_file::version::LanceFileVersion;
+use lance_file::version::{ConcreteFileVersion, LanceFileVersion};
 use lancedb::{
     Connection, Error, Result, Table,
     blob::{BlobRangeRequest, blob},
@@ -61,7 +61,7 @@ async fn create_inline_blob_table(
     Ok(table)
 }
 
-async fn storage_format_version(table: &Table) -> LanceFileVersion {
+async fn storage_format_version(table: &Table) -> ConcreteFileVersion {
     table
         .as_native()
         .unwrap()
@@ -69,9 +69,14 @@ async fn storage_format_version(table: &Table) -> LanceFileVersion {
         .await
         .unwrap()
         .data_storage_format
-        .lance_file_version()
-        .unwrap()
-        .resolve()
+        .lance_file_format()
+}
+
+fn supports_blob_v2(version: ConcreteFileVersion) -> bool {
+    matches!(
+        version,
+        ConcreteFileVersion::V2_2 | ConcreteFileVersion::V2_3
+    )
 }
 
 async fn uses_stable_row_ids(table: &Table) -> bool {
@@ -112,7 +117,7 @@ async fn declaring_blob_column_bumps_format_and_enables_stable_row_ids() -> Resu
         .execute()
         .await?;
 
-    assert!(storage_format_version(&table).await >= LanceFileVersion::V2_2);
+    assert!(supports_blob_v2(storage_format_version(&table).await));
     assert!(uses_stable_row_ids(&table).await);
     Ok(())
 }
@@ -127,7 +132,7 @@ async fn explicit_stable_row_id_setting_wins_over_blob_default() -> Result<()> {
         .execute()
         .await?;
 
-    assert!(storage_format_version(&table).await >= LanceFileVersion::V2_2);
+    assert!(supports_blob_v2(storage_format_version(&table).await));
     assert!(!uses_stable_row_ids(&table).await);
     Ok(())
 }
@@ -139,7 +144,7 @@ async fn non_blob_table_keeps_default_format_and_row_id_setting() -> Result<()>
     let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)]));
     let table = db.create_empty_table("t", schema).execute().await?;
 
-    assert!(storage_format_version(&table).await < LanceFileVersion::V2_2);
+    assert!(!supports_blob_v2(storage_format_version(&table).await));
     assert!(!uses_stable_row_ids(&table).await);
     Ok(())
 }
@@ -171,7 +176,7 @@ async fn creating_with_blob_data_bumps_format() -> Result<()> {
     .unwrap();
     let table = db.create_table("t", batch).execute().await?;
 
-    assert!(storage_format_version(&table).await >= LanceFileVersion::V2_2);
+    assert!(supports_blob_v2(storage_format_version(&table).await));
     assert!(uses_stable_row_ids(&table).await);
     assert_eq!(table.count_rows(None).await?, 1);
     Ok(())
@@ -281,7 +286,7 @@ async fn connection_level_stable_row_id_setting_wins_over_blob_default() -> Resu
         .execute()
         .await?;
 
-    assert!(storage_format_version(&table).await >= LanceFileVersion::V2_2);
+    assert!(supports_blob_v2(storage_format_version(&table).await));
     assert!(!uses_stable_row_ids(&table).await);
     Ok(())
 }
@@ -297,7 +302,7 @@ async fn namespace_create_applies_blob_defaults() -> Result<()> {
         .execute()
         .await?;
 
-    assert!(storage_format_version(&table).await >= LanceFileVersion::V2_2);
+    assert!(supports_blob_v2(storage_format_version(&table).await));
     assert!(uses_stable_row_ids(&table).await);
     Ok(())
 }
@@ -474,7 +479,7 @@ async fn fetch_blobs_round_trips_nested_blob_column() -> Result<()> {
     let batch = RecordBatch::try_new(schema, vec![Arc::new(info_array) as ArrayRef]).unwrap();
     let table = db.create_table("t", batch).execute().await?;
 
-    assert!(storage_format_version(&table).await >= LanceFileVersion::V2_2);
+    assert!(supports_blob_v2(storage_format_version(&table).await));
     assert!(uses_stable_row_ids(&table).await);
 
     let ids = collect_row_ids(&table).await?;
@@ -1305,7 +1310,7 @@ async fn optimize_preserves_blob_v2_null_and_empty_distinction() -> Result<()> {
         .await?;
     table.add(null_empty_input_batch()).execute().await?;
     assert!(
-        storage_format_version(&table).await >= LanceFileVersion::V2_2,
+        supports_blob_v2(storage_format_version(&table).await),
         "blob v2 columns require storage >= 2.2"
     );