Bump version: 0.37.1-beta.1 → 0.38.0-beta.0

feat: add asynchronous drop table API (#3936 )
## Summary - add `drop_table_async` and return a job handle while preserving `drop_table` - consume remote 202 responses with cleanup job IDs and retain older-server compatibility - expose the API through Python and TypeScript connection wrappers
2026-08-26 07:58:31 +00:00 · 2026-08-14 01:09:15 +00:00 · 2026-08-13 18:05:44 -07:00 · 2026-08-13 11:26:58 -07:00 · 2026-08-13 13:23:37 -04:00 · 2026-08-14 00:00:18 +08:00
48 changed files with 1837 additions and 368 deletions
@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "0.37.1-beta.1"
+current_version = "0.38.0-beta.0"
 parse = """(?x)
    (?P<major>0|[1-9]\\d*)\\.
    (?P<minor>0|[1-9]\\d*)\\.
@@ -4,14 +4,14 @@ on:
  workflow_call:
    inputs:
      tag:
-        description: "Tag name from Lance. If omitted, the skill will use the latest Lance release that needs an update."
+        description: "Tag name from Lance (e.g. `v7.2.0-beta.1`). If omitted, the newest release is resolved automatically — stable releases are preferred over pre-releases — and the run is skipped if it is not newer than the version currently pinned in Cargo.toml."
        required: false
        default: ""
        type: string
  workflow_dispatch:
    inputs:
      tag:
-        description: "Tag name from Lance. Leave empty to use the latest Lance release that needs an update."
+        description: "Tag name from Lance (e.g. `v7.2.0-beta.1`). Leave empty to resolve the newest release automatically — stable releases are preferred over pre-releases — and skip the run if it is not newer than the version currently pinned in Cargo.toml."
        required: false
        default: ""
        type: string
@@ -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: |
@@ -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
@@ -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.8"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
 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.8"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
 dependencies = [
 "arc-swap",
 "arrow",
@@ -4832,7 +4832,6 @@ dependencies = [
 "async-recursion",
 "async-trait",
 "async_cell",
- "aws-credential-types",
 "aws-sdk-dynamodb",
 "byteorder",
 "bytes",
@@ -4848,7 +4847,6 @@ dependencies = [
 "either",
 "fst",
 "futures",
- "half",
 "humantime",
 "itertools 0.14.0",
 "lance-arrow",
@@ -4890,8 +4888,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.8"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4913,7 +4911,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.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4927,7 +4925,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.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -4936,8 +4934,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.8"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
 dependencies = [
 "arrayref",
 "crunchy",
@@ -4947,8 +4945,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.8"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4956,12 +4954,10 @@ dependencies = [
 "arrow-schema",
 "async-trait",
 "blake3",
- "byteorder",
 "bytes",
 "datafusion-common",
 "datafusion-sql",
 "futures",
- "itertools 0.14.0",
 "lance-arrow",
 "lance-derive",
 "libc",
@@ -4979,7 +4975,6 @@ dependencies = [
 "snafu 0.9.0",
 "tempfile",
 "tokio",
- "tokio-stream",
 "tokio-util",
 "tracing",
 "twox-hash",
@@ -4988,8 +4983,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.8"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5019,8 +5014,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.8"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5037,8 +5032,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.8"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
 dependencies = [
 "proc-macro2",
 "quote",
@@ -5047,8 +5042,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.8"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
 dependencies = [
 "arrow-arith",
 "arrow-array",
@@ -5073,7 +5068,6 @@ dependencies = [
 "num-traits",
 "prost",
 "prost-build",
- "rand 0.9.5",
 "tokio",
 "tracing",
 "xxhash-rust",
@@ -5082,8 +5076,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.8"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
 dependencies = [
 "arrow-arith",
 "arrow-array",
@@ -5114,8 +5108,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.8"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
 dependencies = [
 "arc-swap",
 "arrow",
@@ -5130,7 +5124,6 @@ dependencies = [
 "async-trait",
 "bitvec",
 "bytes",
- "chrono",
 "crossbeam-queue",
 "datafusion",
 "datafusion-common",
@@ -5148,7 +5141,6 @@ dependencies = [
 "lance-bitpacking",
 "lance-core",
 "lance-datafusion",
- "lance-datagen",
 "lance-encoding",
 "lance-file",
 "lance-index-core",
@@ -5177,13 +5169,12 @@ dependencies = [
 "tempfile",
 "tokio",
 "tracing",
- "uuid",
 ]

 [[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.8"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -5205,8 +5196,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.8"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5220,7 +5211,6 @@ dependencies = [
 "futures",
 "http 1.5.0",
 "io-uring",
- "lance-arrow",
 "lance-core",
 "lance-namespace",
 "log",
@@ -5238,29 +5228,28 @@ dependencies = [
 "tokio",
 "tracing",
 "url",
+ "uuid",
 ]

 [[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.8"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
 dependencies = [
 "arrow-array",
- "arrow-buffer",
 "arrow-schema",
 "cc",
 "half",
 "lance-arrow",
 "lance-core",
 "num-traits",
- "rand 0.9.5",
 "rayon",
 ]

 [[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.8"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
 dependencies = [
 "arrow",
 "async-trait",
@@ -5272,8 +5261,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.8"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
 dependencies = [
 "arrow",
 "arrow-ipc",
@@ -5326,14 +5315,13 @@ 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.8"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
 "arrow-schema",
 "byteorder",
- "bytes",
 "itertools 0.14.0",
 "lance-core",
 "roaring",
@@ -5342,8 +5330,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.8"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5383,8 +5371,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.8"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -5397,8 +5385,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.8"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
 dependencies = [
 "frostem",
 "icu_segmenter",
@@ -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.8", default-features = false, "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-core = { "version" = "=11.0.0-beta.8", "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-datagen = { "version" = "=11.0.0-beta.8", "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-file = { "version" = "=11.0.0-beta.8", "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-io = { "version" = "=11.0.0-beta.8", default-features = false, "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-index = { "version" = "=11.0.0-beta.8", "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-linalg = { "version" = "=11.0.0-beta.8", "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace = { "version" = "=11.0.0-beta.8", "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace-impls = { "version" = "=11.0.0-beta.8", default-features = false, "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-table = { "version" = "=11.0.0-beta.8", "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-testing = { "version" = "=11.0.0-beta.8", "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-datafusion = { "version" = "=11.0.0-beta.8", "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-encoding = { "version" = "=11.0.0-beta.8", "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-arrow = { "version" = "=11.0.0-beta.8", "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
 ahash = "0.8"
 # Note that this one does not include pyarrow
 arrow = { version = "58.0.0", optional = false }
@@ -101,6 +101,13 @@ ignore = [
    # https://rustsec.org/advisories/RUSTSEC-2026-0195
    { id = "RUSTSEC-2026-0194", reason = "transitive via inferno/lance/opendal; XML from trusted cloud endpoints, not attacker-controlled" },
    { id = "RUSTSEC-2026-0195", reason = "transitive via inferno/lance/opendal; XML from trusted cloud endpoints, not attacker-controlled" },
+    # smartstring: unmaintained — the repository was archived by its author on
+    # 2026-05-03. Not a vulnerability. Reached only transitively through polars
+    # (polars-core/-io/-ops/-time/-utils); nothing in LanceDB depends on it directly.
+    # The advisory states no safe upgrade is available: upstream recommends
+    # compact_str/smol_str, so clearing this requires polars to migrate.
+    # https://rustsec.org/advisories/RUSTSEC-2026-0249
+    { id = "RUSTSEC-2026-0249", reason = "smartstring unmaintained via polars; no fixed upstream release" },
 ]

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

@@ -386,6 +386,29 @@ Drop an existing table.

 ***

+### dropTableAsync()
+
+```ts
+abstract dropTableAsync(name, namespacePath?): Promise<Job>
+```
+
+Start dropping a table and return its cleanup job.
+
+The table may become unavailable before its data files are removed. Wait
+on the returned job to know when cleanup has finished.
+
+#### Parameters
+
+* **name**: `string`
+
+* **namespacePath?**: `string`[]
+
+#### Returns
+
+`Promise`&lt;[`Job`](Job.md)&gt;
+
+***
+
 ### getJob()

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

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

    <groupId>com.lancedb</groupId>
    <artifactId>lancedb-parent</artifactId>
-    <version>0.37.1-beta.1</version>
+    <version>0.38.0-beta.0</version>
    <packaging>pom</packaging>
    <name>${project.artifactId}</name>
    <description>LanceDB Java SDK Parent POM</description>
@@ -28,7 +28,7 @@
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <arrow.version>15.0.0</arrow.version>
-        <lance-core.version>11.0.0-beta.3</lance-core.version>
+        <lance-core.version>11.0.0-beta.8</lance-core.version>
        <spotless.skip>false</spotless.skip>
        <spotless.version>2.30.0</spotless.version>
        <spotless.java.googlejavaformat.version>1.7</spotless.java.googlejavaformat.version>
@@ -1,7 +1,7 @@
 [package]
 name = "lancedb-nodejs"
 edition.workspace = true
-version = "0.37.1-beta.1"
+version = "0.38.0-beta.0"
 publish = false
 license.workspace = true
 description.workspace = true
@@ -89,6 +89,16 @@ describe("given a connection", () => {
    await db.createTable("test4", [{ id: 1 }, { id: 2 }]);
  });

+  it("should return a completed job when dropping a local table", async () => {
+    await db.createTable("async-drop", [{ id: 1 }]);
+
+    const job = await db.dropTableAsync("async-drop");
+    expect(job.id).toBeNull();
+    await expect(job.status()).resolves.toBe("finished");
+    await job.wait();
+    await expect(db.tableNames()).resolves.toEqual([]);
+  });
+
  it("should fail if creating table twice, unless overwrite is true", async () => {
    let tbl = await db.createTable("test", [{ id: 1 }, { id: 2 }]);
    await expect(tbl.countRows()).resolves.toBe(2);
@@ -327,6 +327,14 @@ export abstract class Connection {
   */
  abstract dropTable(name: string, namespacePath?: string[]): Promise<void>;

+  /**
+   * Start dropping a table and return its cleanup job.
+   *
+   * The table may become unavailable before its data files are removed. Wait
+   * on the returned job to know when cleanup has finished.
+   */
+  abstract dropTableAsync(name: string, namespacePath?: string[]): Promise<Job>;
+
  /**
   * Drop all tables in the database.
   * @param {string[]} namespacePath The namespace path to drop tables from (defaults to root namespace).
@@ -705,6 +713,10 @@ export class LocalConnection extends Connection {
    return this.inner.dropTable(name, namespacePath ?? []);
  }

+  async dropTableAsync(name: string, namespacePath?: string[]): Promise<Job> {
+    return this.inner.dropTableAsync(name, namespacePath ?? []);
+  }
+
  async dropAllTables(namespacePath?: string[]): Promise<void> {
    return this.inner.dropAllTables(namespacePath ?? []);
  }
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-darwin-arm64",
-	"version": "0.37.1-beta.1",
+	"version": "0.38.0-beta.0",
 	"os": ["darwin"],
 	"cpu": ["arm64"],
 	"main": "lancedb.darwin-arm64.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-gnu",
-	"version": "0.37.1-beta.1",
+	"version": "0.38.0-beta.0",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-gnu.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-musl",
-	"version": "0.37.1-beta.1",
+	"version": "0.38.0-beta.0",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-musl.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-gnu",
-	"version": "0.37.1-beta.1",
+	"version": "0.38.0-beta.0",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-gnu.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-musl",
-	"version": "0.37.1-beta.1",
+	"version": "0.38.0-beta.0",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-musl.node",
@@ -1,6 +1,6 @@
 {
  "name": "@lancedb/lancedb-win32-arm64-msvc",
-  "version": "0.37.1-beta.1",
+  "version": "0.38.0-beta.0",
  "os": [
    "win32"
  ],
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-win32-x64-msvc",
-	"version": "0.37.1-beta.1",
+	"version": "0.38.0-beta.0",
 	"os": ["win32"],
 	"cpu": ["x64"],
 	"main": "lancedb.win32-x64-msvc.node",
@@ -11,7 +11,7 @@
    "ann"
  ],
  "private": false,
-  "version": "0.37.1-beta.1",
+  "version": "0.38.0-beta.0",
  "main": "dist/index.js",
  "exports": {
    ".": "./dist/index.js",
@@ -334,6 +334,22 @@ impl Connection {
            .default_error()
    }

+    /// Start dropping a table and return its cleanup job.
+    #[napi(catch_unwind)]
+    pub async fn drop_table_async(
+        &self,
+        name: String,
+        namespace_path: Option<Vec<String>>,
+    ) -> napi::Result<crate::job::Job> {
+        let ns = namespace_path.unwrap_or_default();
+        let job = self
+            .get_inner()?
+            .drop_table_async(&name, &ns)
+            .await
+            .default_error()?;
+        Ok(crate::job::Job::new(job))
+    }
+
    #[napi(catch_unwind)]
    pub async fn drop_all_tables(&self, namespace_path: Option<Vec<String>>) -> napi::Result<()> {
        let ns = namespace_path.unwrap_or_default();
@@ -1,6 +1,6 @@
 [package]
 name = "lancedb-python"
-version = "0.37.1-beta.1"
+version = "0.38.0-beta.0"
 publish = false
 edition.workspace = true
 description = "Python bindings for LanceDB"
@@ -198,6 +198,9 @@ class Connection(object):
    async def drop_table(
        self, name: str, namespace_path: Optional[List[str]] = None
    ) -> None: ...
+    async def drop_table_async(
+        self, name: str, namespace_path: Optional[List[str]] = None
+    ) -> Job: ...
    async def drop_all_tables(
        self, namespace_path: Optional[List[str]] = None
    ) -> None: ...
@@ -524,6 +524,12 @@ class DBConnection(EnforceOverrides):
            namespace_path = []
        raise NotImplementedError

+    def drop_table_async(
+        self, name: str, namespace_path: Optional[List[str]] = None
+    ) -> Job:
+        """Start dropping a table and return its cleanup job."""
+        raise NotImplementedError
+
    def rename_table(
        self,
        cur_name: str,
@@ -1186,6 +1192,20 @@ class LanceDBConnection(DBConnection):
            )
        )

+    @override
+    def drop_table_async(
+        self, name: str, namespace_path: Optional[List[str]] = None
+    ) -> Job:
+        """Start dropping a table and return its cleanup job.
+
+        The table may become unavailable before its data files are removed.
+        Call :meth:`Job.wait` to wait for cleanup to finish.
+        """
+        if namespace_path is None:
+            namespace_path = []
+        job = LOOP.run(self._conn.drop_table_async(name, namespace_path=namespace_path))
+        return Job(job if isinstance(job, AsyncJob) else AsyncJob(job))
+
    @override
    def drop_all_tables(self, namespace_path: Optional[List[str]] = None):
        if namespace_path is None:
@@ -1963,6 +1983,23 @@ class AsyncConnection(object):
            if f"Table '{name}' was not found" not in str(e):
                raise e

+    async def drop_table_async(
+        self,
+        name: str,
+        *,
+        namespace_path: Optional[List[str]] = None,
+    ) -> AsyncJob:
+        """Start dropping a table and return its cleanup job.
+
+        The table may become unavailable before its data files are removed.
+        Await :meth:`AsyncJob.wait` to wait for cleanup to finish.
+        """
+        if namespace_path is None:
+            namespace_path = []
+        return AsyncJob(
+            await self._inner.drop_table_async(name, namespace_path=namespace_path)
+        )
+
    async def drop_all_tables(self, namespace_path: Optional[List[str]] = None):
        """Drop all tables from the database.

@@ -49,6 +49,7 @@ from lancedb._lancedb import (
 )
 from lancedb.background_loop import LOOP
 from lancedb.db import AsyncConnection, DBConnection
+from lancedb.job import AsyncJob, Job
 from lance_namespace import (
    LanceNamespace,
    connect as namespace_connect,
@@ -624,6 +625,18 @@ class LanceNamespaceDBConnection(DBConnection):
            namespace_path = []
        LOOP.run(self._inner.drop_table(name, namespace_path=namespace_path))

+    @override
+    def drop_table_async(
+        self, name: str, namespace_path: Optional[List[str]] = None
+    ) -> Job:
+        """Start dropping a table and return its cleanup job."""
+        if namespace_path is None:
+            namespace_path = []
+        job = LOOP.run(
+            self._inner.drop_table_async(name, namespace_path=namespace_path)
+        )
+        return Job(job if isinstance(job, AsyncJob) else AsyncJob(job))
+
    @override
    def rename_table(
        self,
@@ -1134,6 +1147,14 @@ class AsyncLanceNamespaceDBConnection:
            namespace_path = []
        await self._inner.drop_table(name, namespace_path=namespace_path)

+    async def drop_table_async(
+        self, name: str, namespace_path: Optional[List[str]] = None
+    ) -> AsyncJob:
+        """Start dropping a table and return its cleanup job."""
+        if namespace_path is None:
+            namespace_path = []
+        return await self._inner.drop_table_async(name, namespace_path=namespace_path)
+
    async def rename_table(
        self,
        cur_name: str,
@@ -23,7 +23,7 @@ import pyarrow as pa

 from ..common import DATA
 from ..db import DBConnection, LOOP
-from ..job import Job
+from ..job import AsyncJob, Job

 if TYPE_CHECKING:
    from .._lancedb import JobDescription, JobInfo
@@ -663,6 +663,16 @@ class RemoteDBConnection(DBConnection):
            namespace_path = []
        LOOP.run(self._conn.drop_table(name, namespace_path=namespace_path))

+    @override
+    def drop_table_async(
+        self, name: str, namespace_path: Optional[List[str]] = None
+    ) -> Job:
+        """Start dropping a table and return its cleanup job."""
+        if namespace_path is None:
+            namespace_path = []
+        job = LOOP.run(self._conn.drop_table_async(name, namespace_path=namespace_path))
+        return Job(job if isinstance(job, AsyncJob) else AsyncJob(job))
+
    @override
    def rename_table(
        self,
@@ -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
@@ -755,8 +755,7 @@ def test_delete_table(tmp_db: lancedb.DBConnection):
    assert tmp_db.table_names() == []


-@pytest.mark.asyncio
-async def test_delete_table_async(tmp_db: lancedb.DBConnection):
+def test_drop_table_async(tmp_db: lancedb.DBConnection):
    data = pd.DataFrame(
        {
            "vector": [[3.1, 4.1], [5.9, 26.5]],
@@ -772,7 +771,10 @@ async def test_delete_table_async(tmp_db: lancedb.DBConnection):

    assert tmp_db.table_names() == ["test"]

-    tmp_db.drop_table("test")
+    job = tmp_db.drop_table_async("test")
+    assert job.id is None
+    assert job.status() == "finished"
+    job.wait()
    assert tmp_db.table_names() == []

    tmp_db.create_table("test", data=data)
@@ -781,6 +783,17 @@ async def test_delete_table_async(tmp_db: lancedb.DBConnection):
    tmp_db.drop_table("does_not_exist", ignore_missing=True)


+@pytest.mark.asyncio
+async def test_drop_table_async_connection(tmp_db_async: lancedb.AsyncConnection):
+    await tmp_db_async.create_table("test", data=pa.table({"id": [1, 2]}))
+
+    job = await tmp_db_async.drop_table_async("test")
+    assert job.id is None
+    assert await job.status() == "finished"
+    await job.wait()
+    assert await tmp_db_async.table_names() == []
+
+
 def test_drop_database(tmp_db: lancedb.DBConnection):
    data = pd.DataFrame(
        {
@@ -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(
@@ -346,6 +346,23 @@ impl Connection {
        })
    }

+    #[pyo3(signature = (name, namespace_path=None))]
+    pub fn drop_table_async(
+        self_: PyRef<'_, Self>,
+        name: String,
+        namespace_path: Option<Vec<String>>,
+    ) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.get_inner()?.clone();
+        let ns_path = namespace_path.unwrap_or_default();
+        future_into_py(self_.py(), async move {
+            inner
+                .drop_table_async(name, &ns_path)
+                .await
+                .infer_error()
+                .map(crate::job::Job::new)
+        })
+    }
+
    #[pyo3(signature = (namespace_path=None,))]
    pub fn drop_all_tables(
        self_: PyRef<'_, Self>,
@@ -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
@@ -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>,
@@ -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,
@@ -1,6 +1,6 @@
 [package]
 name = "lancedb"
-version = "0.37.1-beta.1"
+version = "0.38.0-beta.0"
 edition.workspace = true
 description = "LanceDB: A serverless, low-latency vector database for AI applications"
 license.workspace = true
@@ -188,6 +188,9 @@ required-features = ["bedrock"]
 [[example]]
 name = "bench_streaming_dataloader"

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

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

@@ -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(),
@@ -561,6 +565,21 @@ impl Connection {
            .await
    }

+    /// Start dropping a table and return a handle to the cleanup job.
+    ///
+    /// The table may become unavailable before its physical data is removed.
+    /// Call [`crate::job::Job::wait`] to wait for cleanup to finish. Local
+    /// backends may complete the drop before returning the handle.
+    pub async fn drop_table_async(
+        &self,
+        name: impl AsRef<str>,
+        namespace_path: &[String],
+    ) -> Result<crate::job::Job> {
+        self.internal
+            .drop_table_async(name.as_ref(), namespace_path)
+            .await
+    }
+
    /// Drop the database
    ///
    /// This is the same as dropping all of the tables
@@ -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]
@@ -323,6 +323,18 @@ pub trait Database:
    ) -> Result<()>;
    /// Drop a table in the database
    async fn drop_table(&self, name: &str, namespace_path: &[String]) -> Result<()>;
+    /// Start dropping a table and return a handle to the cleanup job.
+    ///
+    /// Backends without asynchronous cleanup complete the drop before
+    /// returning an already-finished job.
+    async fn drop_table_async(
+        &self,
+        name: &str,
+        namespace_path: &[String],
+    ) -> Result<crate::job::Job> {
+        self.drop_table(name, namespace_path).await?;
+        Ok(crate::job::Job::new_done())
+    }
    /// Drop all tables in the database
    async fn drop_all_tables(&self, namespace_path: &[String]) -> Result<()>;
    fn as_any(&self) -> &dyn std::any::Any;
@@ -1291,16 +1291,21 @@ 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};
    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::time::Duration;
    use tempfile::tempdir;
+    use tokio::sync::Barrier;
+    use tokio::time::timeout;

    async fn setup_database() -> (tempfile::TempDir, ListingDatabase) {
        let tempdir = tempdir().unwrap();
@@ -1324,6 +1329,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();
@@ -19,6 +19,15 @@ const ARROW_FILE_CONTENT_TYPE: &str = "application/vnd.apache.arrow.file";
 #[cfg(test)]
 const JSON_CONTENT_TYPE: &str = "application/json";

+fn extract_job_id(body: &str) -> Option<String> {
+    serde_json::from_str::<serde_json::Value>(body)
+        .ok()?
+        .get("job_id")?
+        .as_str()
+        .filter(|job_id| !job_id.is_empty())
+        .map(str::to_string)
+}
+
 pub use client::{ClientConfig, HeaderProvider, RetryConfig, TimeoutConfig, TlsConfig};
 pub use db::{RemoteDatabaseOptions, RemoteDatabaseOptionsBuilder};
 pub use oauth::{OAuthConfig, OAuthFlow, OAuthHeaderProvider};
@@ -9,6 +9,7 @@ use http::StatusCode;
 use lance_io::object_store::StorageOptions;
 use lance_namespace_impls::{DynamicContextProvider, OperationInfo};
 use moka::future::Cache;
+use reqwest::Response;
 use reqwest::header::CONTENT_TYPE;

 use lance_namespace::models::{
@@ -23,15 +24,17 @@ use crate::database::{
    JobDescription, JobInfo, OpenTableRequest, ReadConsistency, TableNamesRequest,
 };
 use crate::error::Result;
+use crate::job::Job;
+use crate::remote::job::RemoteJob;
 use crate::remote::util::stream_as_body;
 use crate::table::BaseTable;

-use super::ARROW_STREAM_CONTENT_TYPE;
 use super::client::{
    ClientConfig, HeaderProvider, HttpSend, RequestResultExt, RestfulLanceDbClient, Sender,
 };
 use super::table::RemoteTable;
 use super::util::parse_server_version;
+use super::{ARROW_STREAM_CONTENT_TYPE, extract_job_id};

 // Request structure for the remote clone table API
 #[derive(serde::Serialize)]
@@ -326,6 +329,22 @@ impl RemoteDatabase {
    }
 }

+impl<S: HttpSend> RemoteDatabase<S> {
+    async fn submit_drop_table(
+        &self,
+        name: &str,
+        namespace_path: &[String],
+    ) -> Result<(String, Response)> {
+        let identifier = build_table_identifier(name, namespace_path, &self.client.id_delimiter);
+        let cache_key = build_cache_key(name, namespace_path);
+        let req = self.client.post(&format!("/v1/table/{}/drop/", identifier));
+        let (request_id, resp) = self.client.send(req).await?;
+        let resp = self.client.check_response(&request_id, resp).await?;
+        self.table_cache.remove(&cache_key).await;
+        Ok((request_id, resp))
+    }
+}
+
 #[cfg(all(test, feature = "remote"))]
 mod test_utils {
    use super::*;
@@ -894,13 +913,28 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
    }

    async fn drop_table(&self, name: &str, namespace_path: &[String]) -> Result<()> {
-        let identifier = build_table_identifier(name, namespace_path, &self.client.id_delimiter);
-        let cache_key = build_cache_key(name, namespace_path);
-        let req = self.client.post(&format!("/v1/table/{}/drop/", identifier));
-        let (request_id, resp) = self.client.send(req).await?;
-        self.client.check_response(&request_id, resp).await?;
-        self.table_cache.remove(&cache_key).await;
-        Ok(())
+        self.submit_drop_table(name, namespace_path)
+            .await
+            .map(|_| ())
+    }
+
+    async fn drop_table_async(&self, name: &str, namespace_path: &[String]) -> Result<Job> {
+        let (request_id, response) = self.submit_drop_table(name, namespace_path).await?;
+        let status = response.status();
+        let body = response.text().await.err_to_http(request_id.clone())?;
+        let job_id = extract_job_id(&body);
+        Ok(match job_id {
+            Some(job_id) => Job::new(Box::new(RemoteJob::new(self.client.clone(), job_id))),
+            None if status == StatusCode::ACCEPTED => {
+                return Err(Error::Http {
+                    source: "asynchronous drop-table response did not contain a valid job_id"
+                        .into(),
+                    request_id,
+                    status_code: Some(status),
+                });
+            }
+            None => Job::new_done(),
+        })
    }

    async fn drop_all_tables(&self, namespace_path: &[String]) -> Result<()> {
@@ -1492,6 +1526,67 @@ mod tests {
        // NOTE: the API will return 200 even if the table does not exist. So we shouldn't expect 404.
    }

+    #[tokio::test]
+    async fn test_drop_table_does_not_read_response_body() {
+        let conn = Connection::new_with_handler(|_| {
+            http::Response::builder()
+                .status(200)
+                .body(vec![0xff])
+                .unwrap()
+        });
+
+        conn.drop_table("table1", &[]).await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn test_drop_table_async_returns_job() {
+        let conn = Connection::new_with_handler(|request| {
+            assert_eq!(request.method(), &reqwest::Method::POST);
+            assert_eq!(request.url().path(), "/v1/table/table1/drop/");
+            http::Response::builder()
+                .status(202)
+                .body(r#"{"job_id":"drop-job-123"}"#)
+                .unwrap()
+        });
+
+        let job = conn.drop_table_async("table1", &[]).await.unwrap();
+        assert_eq!(job.id(), Some("drop-job-123"));
+    }
+
+    #[tokio::test]
+    async fn test_drop_table_async_old_server_returns_done_job() {
+        let conn = Connection::new_with_handler(|_| {
+            http::Response::builder().status(200).body("").unwrap()
+        });
+
+        let job = conn.drop_table_async("table1", &[]).await.unwrap();
+        assert_eq!(job.id(), None);
+        assert_eq!(job.status().await.unwrap(), "finished");
+    }
+
+    #[tokio::test]
+    async fn test_drop_table_async_rejects_accepted_response_without_job_id() {
+        let conn = Connection::new_with_handler(|_| {
+            http::Response::builder().status(202).body("{}").unwrap()
+        });
+
+        let error = conn.drop_table_async("table1", &[]).await.err().unwrap();
+        assert!(error.to_string().contains("valid job_id"));
+    }
+
+    #[tokio::test]
+    async fn test_drop_table_async_rejects_empty_job_id() {
+        let conn = Connection::new_with_handler(|_| {
+            http::Response::builder()
+                .status(202)
+                .body(r#"{"job_id":""}"#)
+                .unwrap()
+        });
+
+        let error = conn.drop_table_async("table1", &[]).await.err().unwrap();
+        assert!(error.to_string().contains("valid job_id"));
+    }
+
    #[tokio::test]
    async fn test_rename_table() {
        let conn = Connection::new_with_handler(|request| {
@@ -8,7 +8,7 @@ use self::insert::{RemoteWriteExec, WriteOp};
 use super::client::RequestResultExt;
 use super::client::{HttpSend, RestfulLanceDbClient, Sender};
 use super::db::ServerVersion;
-use super::{ARROW_FILE_CONTENT_TYPE, ARROW_STREAM_CONTENT_TYPE};
+use super::{ARROW_FILE_CONTENT_TYPE, ARROW_STREAM_CONTENT_TYPE, extract_job_id};
 use crate::blob::BlobFile;
 use crate::data::scannable::{PeekedScannable, Scannable, estimate_write_partitions};
 use crate::expr::expr_to_sql_string;
@@ -392,13 +392,7 @@ impl<S: HttpSend> RemoteTable<S> {
            .text()
            .await
            .ok()
-            .and_then(|body| serde_json::from_str::<serde_json::Value>(&body).ok())
-            .and_then(|value| {
-                value
-                    .get("job_id")
-                    .and_then(|id| id.as_str())
-                    .map(str::to_string)
-            });
+            .and_then(|body| extract_job_id(&body));

        if let Some(wait_timeout) = index.wait_timeout {
            let index_name = index.name.unwrap_or_else(|| format!("{}_idx", column));
@@ -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 {
@@ -2420,8 +2370,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
@@ -2440,7 +2388,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()),
        };
@@ -3708,7 +3661,7 @@ pub struct FragmentSummaryStats {
 #[allow(deprecated)]
 mod tests {
    use std::sync::Arc;
-    use std::sync::atomic::{AtomicBool, Ordering};
+    use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
    use std::time::Duration;

    use arrow_array::{
@@ -3790,73 +3743,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(),
@@ -3864,12 +3794,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"
        );
    }

@@ -5339,7 +5434,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;
@@ -5405,7 +5500,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();
@@ -84,9 +84,8 @@ pub(super) async fn create_lsm_plan(
    let pk_columns = pk_columns(&ds_ref)?;
    // The base index an indexed arm relies on may lag compaction; resolve it so the
    // snapshot retains SSTables the index has not yet caught up to.
-    let arm_index = arm_maintained_index_name(&ds_ref, &query, &details).await?;
-    let (snapshots, in_memory) =
-        build_read_context(table, &ds_ref, &details, arm_index.as_deref()).await?;
+    let arm_indexes = arm_maintained_index_names(&ds_ref, &query, &details).await?;
+    let (snapshots, in_memory) = build_read_context(table, &ds_ref, &details, &arm_indexes).await?;

    let limit = query.base.limit;
    let offset = query.base.offset;
@@ -232,28 +231,36 @@ fn pk_columns(dataset: &Dataset) -> Result<Vec<String>> {
    Ok(pk)
 }

-/// Per-shard SSTable exclusion watermark: the generation at or below which SSTables
-/// are safe to drop for this arm. A generation is droppable only once it is
-/// compacted into the base table AND covered by `index_name`'s catch-up (for an
-/// indexed arm); a plain scan (`index_name == None`) uses the compaction watermark
-/// alone. Capping at the index catch-up keeps rows the base index has not yet
-/// indexed visible through their SSTable. First occurrence per shard mirrors Lance's
-/// `compacted_generation_for_shard`.
+/// Per-shard SSTable exclusion watermark: the generation at or below which
+/// SSTables are safe to drop for this query.
+///
+/// A generation is droppable only once it is compacted into the base table AND
+/// covered by the catch-up of every index the query relies on, so the watermark
+/// is the minimum across `index_names`. Gating on fewer than all of them would
+/// drop SSTables holding rows an uncounted index has not yet indexed, and that
+/// arm would silently return fewer rows.
+///
+/// See [`arm_maintained_index_names`] for which indexes are collected today: a
+/// vector search with a scalar prefilter is not yet among them.
+///
+/// An empty `index_names` (a plain scan) uses the compaction watermark alone.
+/// First occurrence per shard mirrors Lance's `compacted_generation_for_shard`.
 fn exclusion_watermarks(
    details: &MemWalIndexDetails,
-    index_name: Option<&str>,
+    index_names: &[String],
 ) -> HashMap<Uuid, u64> {
    let mut exclude: HashMap<Uuid, u64> = HashMap::new();
    for entry in &details.compacted_sstables {
        let mut watermark = entry.generation;
-        if let Some(name) = index_name
-            && let Some(caught_up) = details
+        for name in index_names {
+            if let Some(caught_up) = details
                .index_catchup
                .iter()
-                .find(|icp| icp.index_name == name)
+                .find(|icp| icp.index_name == *name)
                .and_then(|icp| icp.caught_up_generation_for_shard(&entry.shard_id))
-        {
-            watermark = watermark.min(caught_up);
+            {
+                watermark = watermark.min(caught_up);
+            }
        }
        exclude.entry(entry.shard_id).or_insert(watermark);
    }
@@ -271,9 +278,9 @@ async fn build_read_context(
    table: &NativeTable,
    dataset: &Dataset,
    details: &MemWalIndexDetails,
-    index_name: Option<&str>,
+    index_names: &[String],
 ) -> Result<(Vec<ShardSnapshot>, HashMap<Uuid, InMemoryMemTables>)> {
-    let exclude = exclusion_watermarks(details, index_name);
+    let exclude = exclusion_watermarks(details, index_names);

    let shard_ids = dataset.list_mem_wal_latest_shard_ids().await?;
    // Use the dataset's own object store (not `ObjectStore::from_uri`, which
@@ -487,19 +494,33 @@ async fn index_maintained(
    }))
 }

-/// The maintained base index the query's arm relies on (vector index for ANN, FTS
-/// index for full-text), used to gate SSTable compaction exclusion by index catch-up.
-/// `None` for a plain scan or when no maintained index covers the searched column.
-async fn arm_maintained_index_name(
+/// Every maintained base index this query relies on, used to gate SSTable
+/// exclusion by index catch-up.
+///
+/// Returns a list because the watermark must be the lowest across every index a
+/// query relies on. Today it never holds more than one: `reject_unsupported`
+/// refuses hybrid search, so the vector and full-text arms are mutually
+/// exclusive.
+///
+/// The case that is genuinely multi-index -- a vector search with a scalar or
+/// bitmap prefilter -- is **not collected yet**. Identifying those needs the
+/// planner's chosen indexes, not the columns the filter names, and no Lance API
+/// exposes them. Until it does, such a query is gated on its vector index alone.
+///
+/// Empty for a plain scan, or when no maintained index covers the searched
+/// column.
+async fn arm_maintained_index_names(
    dataset: &Dataset,
    query: &VectorQueryRequest,
    details: &MemWalIndexDetails,
-) -> Result<Option<String>> {
+) -> Result<Vec<String>> {
    use lance::index::DatasetIndexExt;
-    // Resolve the arm's searched column, the index-detail type it relies on, and a
+
+    // Each arm's searched column, the index-detail type it relies on, and a
    // label for diagnostics — catch-up is taken from the vector/FTS index
    // specifically, not a BTree on the same column.
-    let (column, type_url_suffix, arm) = if !query.query_vector.is_empty() {
+    let mut arms: Vec<(String, &str, &str)> = Vec::new();
+    if !query.query_vector.is_empty() {
        let arrow_schema = ArrowSchema::from(dataset.schema());
        let column = match &query.column {
            Some(column) => column.clone(),
@@ -508,31 +529,43 @@ async fn arm_maintained_index_name(
                default_vector_column(&arrow_schema, dim)?
            }
        };
-        (column, "VectorIndexDetails", "vector")
-    } else if let Some(fts) = &query.base.full_text_search {
-        match fts.columns().into_iter().next() {
-            Some(column) => (column, "InvertedIndexDetails", "full-text"),
-            None => return Ok(None),
-        }
-    } else {
-        return Ok(None);
-    };
-    let Some(field) = dataset.schema().field(&column) else {
-        return Ok(None);
-    };
+        arms.push((column, "VectorIndexDetails", "vector"));
+    }
+    if let Some(fts) = &query.base.full_text_search
+        && let Some(column) = fts.columns().into_iter().next()
+    {
+        arms.push((column, "InvertedIndexDetails", "full-text"));
+    }
+    if arms.is_empty() {
+        return Ok(Vec::new());
+    }
+
    let indices = dataset.load_indices().await?;
-    let segment_names: Vec<String> = indices
-        .iter()
-        .filter(|idx| {
-            idx.fields.contains(&field.id)
-                && idx
-                    .index_details
-                    .as_ref()
-                    .is_some_and(|d| d.type_url.ends_with(type_url_suffix))
-        })
-        .map(|idx| idx.name.clone())
-        .collect();
-    resolve_single_index(segment_names, &details.maintained_indexes, arm, &column)
+    let mut names = Vec::with_capacity(arms.len());
+    for (column, type_url_suffix, arm) in arms {
+        let Some(field) = dataset.schema().field(&column) else {
+            continue;
+        };
+        let segment_names: Vec<String> = indices
+            .iter()
+            .filter(|idx| {
+                idx.fields.contains(&field.id)
+                    && idx
+                        .index_details
+                        .as_ref()
+                        .is_some_and(|d| d.type_url.ends_with(type_url_suffix))
+            })
+            .map(|idx| idx.name.clone())
+            .collect();
+        if let Some(name) =
+            resolve_single_index(segment_names, &details.maintained_indexes, arm, &column)?
+        {
+            names.push(name);
+        }
+    }
+    names.sort();
+    names.dedup();
+    Ok(names)
 }

 /// Resolve the single logical index from the names of its matching physical
@@ -734,24 +767,85 @@ mod tests {
        };

        // Plain scan: drop every compacted generation (through 5).
-        assert_eq!(exclusion_watermarks(&details, None).get(&shard), Some(&5));
+        assert_eq!(exclusion_watermarks(&details, &[]).get(&shard), Some(&5));

        // FTS arm with a lagging index: exclusion is capped at the index catch-up
        // (2), so SSTable generations 3..=5 are retained until the index covers
        // them — otherwise those documents would silently vanish from FTS results.
        assert_eq!(
-            exclusion_watermarks(&details, Some("fts_idx")).get(&shard),
+            exclusion_watermarks(&details, &["fts_idx".to_string()]).get(&shard),
            Some(&2)
        );

        // A caught-up index — or one untracked in index_catchup — falls back to the
        // compaction watermark.
        assert_eq!(
-            exclusion_watermarks(&details, Some("caught_up_idx")).get(&shard),
+            exclusion_watermarks(&details, &["caught_up_idx".to_string()]).get(&shard),
            Some(&5)
        );
    }

+    /// A hybrid search reads a vector and a full-text index, and either may lag.
+    /// Retaining to the lower of the two is what keeps both arms complete;
+    /// gating on one alone would drop SSTables the other has not indexed.
+    #[test]
+    fn exclusion_watermark_takes_the_minimum_across_every_index_used() {
+        let shard = Uuid::from_u128(1);
+        let details = MemWalIndexDetails {
+            compacted_sstables: vec![CompactedSsTable::new(shard, 9)],
+            index_catchup: vec![
+                IndexCatchupProgress::new(
+                    "vec_idx".to_string(),
+                    vec![CompactedSsTable::new(shard, 7)],
+                ),
+                IndexCatchupProgress::new(
+                    "fts_idx".to_string(),
+                    vec![CompactedSsTable::new(shard, 4)],
+                ),
+            ],
+            maintained_indexes: vec!["vec_idx".to_string(), "fts_idx".to_string()],
+            ..Default::default()
+        };
+
+        // Each index alone stops at its own catch-up.
+        assert_eq!(
+            exclusion_watermarks(&details, &["vec_idx".to_string()]).get(&shard),
+            Some(&7)
+        );
+        assert_eq!(
+            exclusion_watermarks(&details, &["fts_idx".to_string()]).get(&shard),
+            Some(&4)
+        );
+
+        // Used together, the lower one governs regardless of order.
+        let both = ["vec_idx".to_string(), "fts_idx".to_string()];
+        assert_eq!(exclusion_watermarks(&details, &both).get(&shard), Some(&4));
+        let reversed = ["fts_idx".to_string(), "vec_idx".to_string()];
+        assert_eq!(
+            exclusion_watermarks(&details, &reversed).get(&shard),
+            Some(&4)
+        );
+    }
+
+    /// An index with no catch-up entry contributes no cap today, so a lagging
+    /// sibling must still govern rather than being widened by the untracked one.
+    #[test]
+    fn an_untracked_index_does_not_widen_a_lagging_sibling() {
+        let shard = Uuid::from_u128(1);
+        let details = MemWalIndexDetails {
+            compacted_sstables: vec![CompactedSsTable::new(shard, 9)],
+            index_catchup: vec![IndexCatchupProgress::new(
+                "fts_idx".to_string(),
+                vec![CompactedSsTable::new(shard, 4)],
+            )],
+            maintained_indexes: vec!["fts_idx".to_string(), "untracked_idx".to_string()],
+            ..Default::default()
+        };
+
+        let both = ["fts_idx".to_string(), "untracked_idx".to_string()];
+        assert_eq!(exclusion_watermarks(&details, &both).get(&shard), Some(&4));
+    }
+
    #[test]
    fn resolve_single_index_dedupes_segments() {
        let maintained = vec!["fts_idx".to_string()];
@@ -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"
    );
Author	SHA1	Message	Date
Lance Release	2f57dd053e	Bump version: 0.37.1-beta.1 → 0.38.0-beta.0	2026-08-14 01:09:15 +00:00
Jack Ye	ffd35c1a8f	feat: add asynchronous drop table API (#3936 ) ## Summary - add `drop_table_async` and return a job handle while preserving `drop_table` - consume remote 202 responses with cleanup job IDs and retain older-server compatibility - expose the API through Python and TypeScript connection wrappers	2026-08-13 18:05:44 -07:00
Wyatt Alt	790d0c684c	docs(ci): clarify tag input on codex-update-lance-dependency (#3924 ) Say what resolving "latest" actually does: pick the newest release, preferring stable over pre-release, and skip the run if it is not newer than the version pinned in Cargo.toml. Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>	2026-08-13 11:26:58 -07:00
XY Zhan	251f194696	refactor(lsm): gate SSTable exclusion on every index a query relies on (#3780 ) `exclusion_watermarks` resolved a single index and capped SSTable exclusion at that index's catch-up watermark. It now takes every index the query relies on and retains to the lowest of them, and the resolver collects arms together rather than returning at the first match. This is groundwork, not a fix for a reachable bug: `reject_unsupported` refuses hybrid search, so the vector and full-text arms are mutually exclusive and the list never holds more than one entry today. The generalisation is what the remaining work below plugs into. Unchanged: a plain scan uses the compaction watermark alone, an index with no catch-up entry contributes no cap, and a caught-up index falls back to the compaction watermark. Taking a minimum over more indexes can only lower a watermark, so the failure direction is "read an SSTable unnecessarily", never "miss rows". ## Tests Three in `lsm`: the existing lagging-index test updated for the new signature; `exclusion_watermark_takes_the_minimum_across_every_index_used` (two indexes at 7 and 4 against compaction at 9 — each alone stops at its own watermark, together the lower governs, order-independent); and `an_untracked_index_does_not_widen_a_lagging_sibling`. `cargo test -p lancedb --lib` — 45 lsm tests, 484 in the crate. `cargo fmt --check` clean. ## Follow-ups This crate pins lance to a released tag, so anything needing unreleased Lance symbols waits for a bump. 1. Select legacy versus strict semantics from the feature bit. On a table with `FLAG_MEM_WAL_INDEX_CATCHUP` set, a missing entry must mean "not caught up" and retain the SSTables, instead of leaving the compaction watermark unchanged. Needs the bit from lance-format/lance#8263. This must land before any table is activated — otherwise the bit is set while queries still read permissively. 2. Collect scalar and bitmap-family prefilter indexes. The genuinely multi-index query is a vector search with a scalar prefilter, and it is gated on the vector index alone today. Identifying the others needs the planner's chosen indexes, not the columns the filter names, so it needs a Lance-side helper. 3. Verify a retained SSTable can actually answer. Both base and SSTable arms use `fast_search`; a source without a compatible index contributes nothing, so retention alone does not guarantee its rows are returned. Needs a flat-search fallback or an explicit error in Lance's `LsmScanner`. 4. Planner-level integration tests. Current tests exercise the watermark arithmetic directly. End-to-end coverage over real queries — prefilter forms, legacy versus activated, missing index and missing shard entries — depends on 1–3.	2026-08-13 13:23:37 -04:00
LanceDB Robot	4b7325bd74	chore: update lance dependency to v11.0.0-beta.8 (#3928 ) Updates the Rust workspace and Java lance-core dependency to Lance v11.0.0-beta.8, with refreshed Cargo lockfile metadata. No compatibility fixes were required. Lance tag: https://github.com/lance-format/lance/releases/tag/v11.0.0-beta.8	2026-08-14 00:00:18 +08:00
Yang Cen	1d75638dea	fix: make table existence manifest-authoritative (#3919 ) ## What is the bug? #3731 tries to distinguish a missing table from a corrupt table after Lance returns `DatasetNotFound`. It does that by listing the database parent and treating a physical `<name>.lance` entry as evidence that the table exists. That premise is not sound for a listing database. Table creation writes data before atomically committing the first manifest, so the same physical prefix can represent a live concurrent create, abandoned uncommitted data, or an old empty directory. It is not evidence of a committed table. The parent listing also makes every missing-table open, including the create-on-miss path, perform work proportional to the number of sibling tables. Cloud `list_with_delimiter` exhausts all pages before returning. ## How does this PR fix the problem? This PR makes the committed Lance manifest the sole table-existence authority for listing-database opens: - `DatasetNotFound` maps directly to `TableNotFound`; no parent or target storage probe runs. - Other Lance load errors continue to propagate unchanged. - A physical directory, object prefix, or uncommitted data file alone does not block `Create`. - Concurrent `Create` requests are arbitrated by the conditional version-1 manifest commit: one succeeds and the loser receives `TableAlreadyExists`. - `table_names` is documented as physical discovery, not an atomic table-existence check. Its snapshot can contain an entry that is still being created, has only uncommitted storage, or is concurrently dropped. This removes the need for a new Lance object-store capability. LanceDB remains on the official Lance `v11.0.0-beta.6` dependency from `main`; the merge commit for lance-format/lance#7722 is an ancestor of that tag, so the ambiguous-GCS-500 corruption-prevention fix is retained. ## Performance evidence Lower is better. The benchmark uses real `.lance` directories with marker objects on the local filesystem; fixture creation and teardown are outside the timed region. Baseline is `origin/main` at `6fb976cf`, candidate is `e1240751`. Both were built from the same lockfile on the same macOS arm64 machine with the repository's `release` profile (fat LTO), then executed in alternating baseline/candidate order for three pairs. Each run used 10 warmups and 100 distinct missing-table opens per scale. The table reports the median of the three run-level percentiles. \| Scenario / metric \| Baseline \| This PR \| Benefit \| \| --- \| ---: \| ---: \| ---: \| \| 1,000 real sibling directories, p50 \| 11.905 ms \| 21.042 us \| 566x speedup \| \| 10,000 real sibling directories, p50 \| 143.630 ms \| 18.375 us \| 7,817x speedup \| \| 100,000 real sibling directories, p50 \| 1.991 s \| 19.917 us \| 99,984x speedup \| \| 100,000 real sibling directories, p95 \| 2.346 s \| 25.792 us \| 90,965x speedup \| These results validate removal of the sibling-cardinality dependency in this local-filesystem workload; they are not an extrapolation to production GCS latency. A structural object-store regression test separately asserts that opening one missing table performs zero parent-scoped `list`, `list_with_offset`, or `list_with_delimiter` calls. Run with: ```bash BENCH_SIBLINGS=1000,10000,100000 BENCH_WARMUPS=10 BENCH_TRIALS=100 \ cargo run --locked --release --quiet -p lancedb --example bench_open_missing_table ``` ## Correctness and compatibility boundaries - An empty `.lance` directory or orphan data without a committed manifest now opens as `TableNotFound` and may be replaced by a successful `Create`. - Two synchronized creators sharing one object store deterministically produce one success and one conditional-manifest conflict mapped to `TableAlreadyExists`. - A readable manifest remains authoritative; non-`DatasetNotFound` corruption, external-manifest, authorization, and object-store errors are not folded into `TableNotFound`. - `TableCorrupted` remains in the public error enum for compatibility, but this listing-database fallback no longer synthesizes it from an ambiguous physical footprint. - Reliably distinguishing `Missing`, `Creating`, and `Corrupt` would require explicit authoritative lifecycle/catalog metadata (for example a leased creation record). It cannot be inferred from a directory or prefix, and is outside this incident fix. ## Validation - `cargo fmt --all -- --check` - `cargo check --quiet --locked -p lancedb --features remote --tests --examples` - `cargo clippy --quiet --locked -p lancedb --features remote --tests --examples -- -D warnings` - `cargo test --quiet --locked -p lancedb --features remote --tests` - library: 843 passed, 1 ignored - integration groups: 39 passed, 6 passed, 5 passed - focused coverage for empty directories, orphan data, physical listing snapshots, zero parent listings, and concurrent manifest arbitration	2026-08-13 21:22:42 +08:00
LanceDB Robot	031c3585a8	chore: update lance dependency to v11.0.0-beta.7 (#3925 ) Updates the Rust workspace Lance dependencies and Java lance-core dependency to v11.0.0-beta.7. No compatibility fixes were required; full-workspace Clippy passes with warnings denied. Lance tag: https://github.com/lance-format/lance/releases/tag/v11.0.0-beta.7 --------- Co-authored-by: Yang Cen <159225399+BubbleCal@users.noreply.github.com>	2026-08-13 20:37:19 +08:00
LanceDB Robot	6fb976cf89	chore: update lance dependency to v11.0.0-beta.6 (#3922 ) Updates the Rust workspace Lance dependencies and Java lance-core dependency to v11.0.0-beta.6. Includes compatibility updates for the new concrete Lance file-version API. Trigger: https://github.com/lance-format/lance/releases/tag/v11.0.0-beta.6 --------- Co-authored-by: XYZhan <zhaner08@hotmail.com>	2026-08-12 02:43:44 -04:00
Sravan Avvaru	a615306f39	feat(python): add on_transform_error fault tolerance to StreamingDataset (#3763 ) Closes #3704 ## Problem Transforms can fail on bad data (e.g. nulls/NaNs from incomplete user surveys). Today any transform exception aborts iteration, and there is no way to skip invalid rows during loading. ## Solution New `on_transform_error` parameter on `StreamingDataset`: - `"raise"` (default, matches current behavior and the convention in tf.data / WebDataset / Ray Data) - `"skip"` — drop the failing rows and continue - `"warn"` — like skip, plus a logged warning per failing batch - a WebDataset-style callable `handler(exc) -> bool`, so users can skip only expected error types Key design points: - Row-granular skipping: when a batch fails, the transform is re-run on single-row slices so only the rows that actually fail are dropped (avoids Ray-style whole-block loss). Skips are counted in a new `rows_skipped` property. - No crash on uneven skips: the round-robin loop now ends the epoch at the last cycle where every split still has a row, instead of hitting `IndexError` when a split runs dry early. - Exact resumability under skips: checkpoints are now position-based. `state_dict` gains `positions_consumed_per_split` (exact for owned splits), and a new `merge_state_dicts` static method combines per-rank states via elementwise max for elastic resume across topology changes. Old checkpoints without the new key still load. Positions equal sample counts when nothing is skipped, so existing behavior is unchanged. - Guardrail: transforms returning the wrong number of rows now raise a clear `ValueError` instead of silently corrupting split accounting. ### Answers to the issue's open questions - Can we do this? Yes — all transforms funnel through one guarded call in the Stage 2 pipeline. - What do other libraries do? tf.data `ignore_errors()`, WebDataset `handler=`, Ray `max_errored_blocks`; MosaicML StreamingDataset offers nothing (skipping conflicts with its determinism model). This design follows the common conventions: raise by default, opt-in skipping, count/log drops. - Error handling or pre-filtering? Both: the existing `filter=` remains the recommended tool for predictable bad data (splits are built post-filter, so all guarantees hold — now documented); `on_transform_error` covers failures not expressible as a predicate. - Impact on splits / elastic determinism? Per-split sample sequences stay deterministic (skips are data-dependent, not topology-dependent). With unequal bad-row counts across splits the last few global steps of an epoch can differ across topologies (bounded by the skew), which is documented on the parameter. With equal counts per split, full determinism is preserved — covered by a test. ## Testing 15 new tests in `test_elastic_dataloader.py` covering: default raise, invalid values, uniform and uneven skips (including epoch-end truncation), warn logging, selective callable handlers, wrong-row-count guardrail, determinism across runs and across world sizes (1/2/3/4) with skips, exact mid-epoch resume with skips on the same topology, elastic resume via `merge_state_dicts` (ws=2 → ws=1), merge validation, and backward-compat loading of old checkpoints. Note: relying on CI for the test run — my local machine OOMs during the final link of the native extension. The change itself is pure Python. --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>	2026-08-10 09:22:06 -07:00
Xuanwo	920fc0e455	fix(python): set native module metadata (#3913 ) PyO3 defaults native extension classes to `builtins`, so mkdocstrings/Griffe could not resolve the newly documented `lancedb.Session` alias and `Deploy docs to Pages` failed on `main`. Declare the extension module for the public native types referenced by the Python API docs so Griffe resolves them through `lancedb._lancedb` and Pages can build again. Validated with the docs toolchain used by CI (`griffe==0.49.0`, `mkdocstrings==0.25.2`, and `mkdocs==1.6.1`); `PYTHONPATH=. mkdocs build` succeeds.	2026-08-10 21:40:31 +08:00
Xuanwo	5acce6782e	ci(docs): report link checker failures through issues (#3909 )	2026-08-10 15:08:36 +08:00