chore: retain computed column support in beta.4 update

chore: update lance dependency to v12.0.0-beta.4
feat: let a computed-column batch read its own earlier declarations
2026-09-04 20:48:50 +00:00 · 2026-08-27 17:24:32 -07:00 · 2026-08-27 21:26:01 +00:00 · 2026-08-27 20:46:32 +00:00 · 2026-08-26 01:31:44 +00:00
81 changed files with 12229 additions and 5688 deletions
@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "0.39.0-beta.0"
+current_version = "0.38.0-beta.11"
 parse = """(?x)
    (?P<major>0|[1-9]\\d*)\\.
    (?P<minor>0|[1-9]\\d*)\\.
@@ -44,27 +44,3 @@ updates:
      python-deps:
        patterns:
          - "*"
-
-  # The npm ecosystem covers pnpm lockfiles. There are two separate installs:
-  # the bindings themselves and the examples, which have their own lockfile.
-  # As with cargo and pip above, only bump the lockfile — the version ranges
-  # in package.json are our consumers' constraints, not ours.
-  - package-ecosystem: npm
-    directory: /nodejs
-    schedule:
-      interval: weekly
-    versioning-strategy: lockfile-only
-    groups:
-      nodejs-deps:
-        patterns:
-          - "*"
-
-  - package-ecosystem: npm
-    directory: /nodejs/examples
-    schedule:
-      interval: weekly
-    versioning-strategy: lockfile-only
-    groups:
-      nodejs-examples-deps:
-        patterns:
-          - "*"
@@ -29,14 +29,12 @@ jobs:
    steps:
      - uses: actions/setup-node@v6
        with:
-          node-version: "24"
-      - uses: pnpm/action-setup@v6
-        with:
-          version: 11.1.1
+          node-version: "18"
      # These rules are disabled because Github will always ensure there
      # is a blank line between the title and the body and Github will
      # word wrap the description field to ensure a reasonable max line
      # length.
+      - run: npm install @commitlint/config-conventional
      - run: >
          echo 'module.exports = {
            "rules": {
@@ -45,11 +43,7 @@ jobs:
              "body-leading-blank": [0, "always"]
            }
          }' > .commitlintrc.js
-      - run: >
-          pnpm dlx
-          --package @commitlint/cli@21.2.2
-          --package @commitlint/config-conventional@21.2.2
-          commitlint --extends @commitlint/config-conventional --verbose <<< $COMMIT_MSG
+      - run: npx commitlint --extends @commitlint/config-conventional --verbose <<< $COMMIT_MSG
        env:
          COMMIT_MSG: >
            ${{ github.event.pull_request.title }}
@@ -60,7 +54,7 @@ jobs:
        with:
          script: |
            const message = `**ACTION NEEDED**
-
+              
              Lance follows the [Conventional Commits specification](https://www.conventionalcommits.org/en/v1.0.0/) for release automation.

              The PR title and description are used as the merge commit message.\
@@ -56,7 +56,7 @@ jobs:
        uses: lycheeverse/lychee-action@e7477775783ea5526144ba13e8db5eec57747ce8 # v2.9.0
        with:
          # Restricted to http(s) on purpose. Much of docs/src is generated
-          # API reference (the js/ tree comes from `pnpm run docs` in nodejs)
+          # API reference (the js/ tree comes from `npm run docs` in nodejs)
          # and the hand-written pages use mkdocstrings cross-references and
          # nav-relative paths that only resolve in the site mkdocs builds,
          # not in this checkout, so relative links would be reported as
@@ -55,7 +55,9 @@ jobs:
      - name: Set up node
        uses: actions/setup-node@v6
        with:
-          node-version: 24
+          node-version: 20
+          cache: 'npm'
+          cache-dependency-path: docs/package-lock.json
      - name: Install node dependencies
        working-directory: nodejs
        run: |
@@ -47,8 +47,9 @@ jobs:
        version: 11.1.1
    - uses: actions/setup-node@v6
      with:
-        # Build on a supported LTS; the matrix job below covers every
-        # Node version the library claims to support.
+        # pnpm 11 requires Node >= 22.13; use 24 since 22 hits EOL
+        # in October. The library itself still supports Node >= 18
+        # (see test matrix below).
        node-version: 24
        cache: 'pnpm'
        cache-dependency-path: nodejs/pnpm-lock.yaml
@@ -83,7 +84,7 @@ jobs:
    timeout-minutes: 30
    strategy:
      matrix:
-        node-version: [ "22", "24", "26" ]
+        node-version: [ "18", "20" ]
    runs-on: "ubuntu-22.04"
    defaults:
      run:
@@ -100,9 +101,9 @@ jobs:
    - uses: actions/setup-node@v6
      name: Setup Node.js 24 for build
      with:
-        # Build and install once on a fixed version so the generated docs
-        # are identical across matrix legs; the tests below then run on each
-        # supported Node version.
+        # pnpm 11 requires Node >= 22.13; use 24 since 22 hits EOL
+        # in October. Build/install runs on Node 24; tests run on the
+        # matrix version below using direct jest invocation.
        node-version: 24
        cache: 'pnpm'
        cache-dependency-path: nodejs/pnpm-lock.yaml
@@ -151,9 +152,9 @@ jobs:
        S3_TEST: "1"
        # Newer @smithy/core uses dynamic ESM imports.
        NODE_OPTIONS: "--experimental-vm-modules"
-      # Invoke the installed jest binary directly; the pnpm shim is set up
-      # against the build-phase Node, not the version selected above.
-      run: node_modules/.bin/jest --verbose
+      # Invoke jest directly because pnpm 11 itself requires Node 22+
+      # while the matrix tests on older Node versions.
+      run: npx jest --verbose
    - name: Test examples
      working-directory: ./
      env:
@@ -163,7 +164,7 @@ jobs:
      run: |
        python ci/mock_openai.py &
        cd nodejs/examples
-        node_modules/.bin/jest --testEnvironment jest-environment-node-single-context --verbose
+        npx jest --testEnvironment jest-environment-node-single-context --verbose
  macos:
    timeout-minutes: 30
    # macos-15 ships a newer linker; the older macos-14 linker fails to insert
@@ -184,7 +185,8 @@ jobs:
        version: 11.1.1
    - uses: actions/setup-node@v6
      with:
-        # pnpm 11 requires Node >= 22.13.
+        # pnpm 11 requires Node >= 22.13; use 24 since 22 hits EOL
+        # in October.
        node-version: 24
        cache: 'pnpm'
        cache-dependency-path: nodejs/pnpm-lock.yaml
@@ -40,31 +40,40 @@ jobs:
          - target: aarch64-apple-darwin
            host: macos-latest
            features: fp16kernels
-            # Fat LTO was ~111 of this job's ~113 minutes.
-            lto: thin
-            codegen_units: 16
            pre_build: |-
              brew install protobuf
+              # Fat LTO (the workspace default in .cargo/config.toml) is
+              # single-threaded and is the peak-memory step of the build. On
+              # this runner it accounted for ~111 of the job's ~113 minutes,
+              # making it the critical path of the entire publish pipeline.
+              # ThinLTO parallelizes it across the runner's cores, for a few
+              # percent of runtime performance.
+              export CARGO_PROFILE_RELEASE_LTO=thin
+              export CARGO_PROFILE_RELEASE_CODEGEN_UNITS=16
          - target: x86_64-pc-windows-msvc
            host: windows-2025
            features: ","
-            # The lower peak also keeps this on the standard 4-core runner.
-            lto: thin
-            codegen_units: 16
            pre_build: |-
              choco install --no-progress protoc ninja nasm
              tail -n 1000 /c/ProgramData/chocolatey/logs/chocolatey.log
              # There is an issue where choco doesn't add nasm to the path
              export PATH="$PATH:/c/Program Files/NASM"
              nasm -v
+              # See the ThinLTO note on aarch64-apple-darwin above. Keeping
+              # peak memory down is also what lets this run on the standard
+              # 4-core runner: the 8-core larger runner was only needed to
+              # stop fat LTO from OOMing rustc-LLVM.
+              export CARGO_PROFILE_RELEASE_LTO=thin
+              export CARGO_PROFILE_RELEASE_CODEGEN_UNITS=16
          - target: aarch64-pc-windows-msvc
            host: windows-2025
            features: ","
-            lto: thin
-            codegen_units: 16
            pre_build: |-
                choco install --no-progress protoc
                rustup target add aarch64-pc-windows-msvc
+                # See the ThinLTO note on aarch64-apple-darwin above.
+                export CARGO_PROFILE_RELEASE_LTO=thin
+                export CARGO_PROFILE_RELEASE_CODEGEN_UNITS=16
          - target: x86_64-unknown-linux-gnu
            host: ubuntu-latest
            features: fp16kernels
@@ -94,14 +103,6 @@ jobs:
            # https://github.com/napi-rs/napi-rs/blob/main/debian-aarch64.Dockerfile
            docker: ghcr.io/napi-rs/napi-rs/nodejs-rust:lts-debian-aarch64
            features: "fp16kernels"
-            # Fat LTO OOM-killed rustc every nightly; even with lld it peaked
-            # at 31391 MiB of the runner's 32 GiB.
-            lto: thin
-            codegen_units: 16
-            # arm64 Linux links through GNU `ld` where x86_64 defaults to
-            # `rust-lld`, which is why only arm64 OOM'd. lld cut the largest
-            # linker process 7.0 -> 4.0 GiB (lancedb/sophon#7313).
-            linker: /tmp/aarch64-lld-clang
            pre_build: |-
              set -e &&
              apt-get update &&
@@ -111,30 +112,9 @@ jobs:
              # AT_HWCAP2 (added in Linux 3.17). Define it for aws-lc-sys.
              export CFLAGS="$CFLAGS -DAT_HWCAP2=26" &&
              rustup target add aarch64-unknown-linux-gnu
-              # Not `&&`-chained: in dash, errexit does not fire for a
-              # non-final command in an `&&` list, so failures were ignored.
-              #
-              # A wrapper rather than `-C link-arg` because the per-target
-              # rustflags variable does not reach every unit that links, while
-              # the linker variable does. `clang` because GCC silently ignores
-              # `-fuse-ld=lld` unless built with lld support. Two echoes
-              # because printf's newline escape gets rewritten to `;` between
-              # here and the container.
-              echo '#!/bin/sh' > /tmp/aarch64-lld-clang
-              echo 'exec clang --target=aarch64-unknown-linux-gnu --sysroot=/usr/aarch64-unknown-linux-gnu/aarch64-unknown-linux-gnu/sysroot --gcc-toolchain=/usr/aarch64-unknown-linux-gnu -fuse-ld=lld "$@"' >> /tmp/aarch64-lld-clang
-              chmod 0755 /tmp/aarch64-lld-clang
-              # Fail now, not at the cdylib link ~30 minutes later. Linking at
-              # all also proves lld resolved; clang errors out when it cannot.
-              echo 'int main(void){return 0;}' > /tmp/probe.c
-              /tmp/aarch64-lld-clang /tmp/probe.c -o /tmp/probe
-              readelf -h /tmp/probe | grep AArch64
          - target: aarch64-unknown-linux-musl
            host: ubuntu-2404-8x-x64
            features: ","
-            # Fat LTO took the whole runner down. lld cannot help: it died
-            # inside rustc's LLVM, before any linker was spawned.
-            lto: thin
-            codegen_units: 16
            pre_build: |-
              set -e &&
              sudo apt-get update &&
@@ -143,19 +123,6 @@ jobs:
              export EXTRA_ARGS="-x"
    name: build - ${{ matrix.settings.target }}
    runs-on: ${{ matrix.settings.host }}
-    # On the job, not exported from `pre_build`: `Swatinem/rust-cache` hashes
-    # `CARGO_*` into its cache key before any step runs, so a step-local export
-    # leaves the key unchanged while cargo still rebuilds cold. The ThinLTO
-    # legs had been doing that every run.
-    #
-    # Not `RUSTFLAGS`: setting it, even to "", discards every config-file
-    # rustflag, silently dropping .cargo/config.toml's `target-cpu` and
-    # `target-feature` from the published binaries.
-    env:
-      CARGO_PROFILE_RELEASE_LTO: ${{ matrix.settings.lto || 'fat' }}
-      CARGO_PROFILE_RELEASE_CODEGEN_UNITS: ${{ matrix.settings.codegen_units || '1' }}
-      # Empty elsewhere: a per-target variable is only read for that triple.
-      CARGO_TARGET_AARCH64_UNKNOWN_LINUX_GNU_LINKER: ${{ matrix.settings.linker }}
    defaults:
      run:
        working-directory: nodejs
@@ -168,7 +135,8 @@ jobs:
      - name: Setup node
        uses: actions/setup-node@v6
        with:
-          # pnpm 11 requires Node >= 22.13.
+          # pnpm 11 requires Node >= 22.13; use 24 since 22 hits EOL
+          # in October.
          node-version: 24
          cache: pnpm
          cache-dependency-path: nodejs/pnpm-lock.yaml
@@ -201,15 +169,19 @@ jobs:
          # creating ref). The nightly cadence also keeps entries inside
          # GitHub's 7-day eviction window, which a tag-only trigger would not.
          save-if: ${{ github.ref == 'refs/heads/main' }}
-      # Docker builds can use rust-cache too: the workspace is bind-mounted, so
-      # `target/` lives on the host and rust-cache's prune keeps the entry
-      # small.
+      # Docker builds can use rust-cache too. `target/` already lives on the
+      # host because the whole workspace is bind-mounted into the container, and
+      # rust-cache's prune and save run host-side, so they can manage it -- which
+      # is what keeps the entry to dependency artifacts rather than a multi-GB
+      # copy of everything.
      #
      # Two differences from the native builds. The container's CARGO_HOME is
-      # bind-mounted from `.cargo-cache` rather than ~/.cargo, so that is cached
-      # explicitly. And the key uses the *host* rustc version, not the compiler
-      # that built these artifacts -- safe, since cargo fingerprints the real
-      # one; a base-image bump just costs one cold build.
+      # bind-mounted from `.cargo-cache` rather than the host's ~/.cargo, so that
+      # has to be cached explicitly. And the key is derived from the *host* rustc
+      # version, which is not the compiler that produced these artifacts; that is
+      # safe because cargo fingerprints the real compiler and rebuilds on a
+      # mismatch, it just means a base-image toolchain bump costs one cold build
+      # instead of invalidating the key.
      - name: Cache cargo (docker builds)
        uses: Swatinem/rust-cache@v2
        if: ${{ matrix.settings.docker }}
@@ -238,19 +210,14 @@ jobs:
          # cache step above saves. Previously the registry mounts pointed at
          # `.cargo/...`, a path nothing cached, so the container re-downloaded
          # the whole crate registry on every run.
-          #
-          # `docker run` inherits nothing; `-e NAME` carries the job's `env:` in.
          options: "--user 0:0 -v ${{ github.workspace }}/.cargo-cache/git/db:/usr/local/cargo/git/db \
                    -v ${{ github.workspace }}/.cargo-cache/registry/cache:/usr/local/cargo/registry/cache \
                    -v ${{ github.workspace }}/.cargo-cache/registry/index:/usr/local/cargo/registry/index \
-                    -e CARGO_PROFILE_RELEASE_LTO \
-                    -e CARGO_PROFILE_RELEASE_CODEGEN_UNITS \
-                    -e CARGO_TARGET_AARCH64_UNKNOWN_LINUX_GNU_LINKER \
                    -v ${{ github.workspace }}:/build -w /build/nodejs"
          run: |
            set -e
            ${{ matrix.settings.pre_build }}
-            node_modules/.bin/napi build --platform --release \
+            npx napi build --platform --release \
              --features ${{ matrix.settings.features }} \
              --target ${{ matrix.settings.target }} \
              --dts ../lancedb/native.d.ts \
@@ -270,7 +237,7 @@ jobs:
      - name: Build
        run: |
          ${{ matrix.settings.pre_build }}
-          node_modules/.bin/napi build --platform --release \
+          npx napi build --platform --release \
              --features ${{ matrix.settings.features }} \
              --target ${{ matrix.settings.target }} \
              --dts ../lancedb/native.d.ts \
@@ -289,18 +256,6 @@ jobs:
        if: always()
        run: df -h
        shell: bash
-      - name: Report peak memory
-        if: always() && runner.os == 'Linux'
-        shell: bash
-        run: |
-          peak=$(find /sys/fs/cgroup -name memory.peak -readable \
-                 -exec cat {} + 2>/dev/null | sort -n | tail -1)
-          if [ -n "$peak" ]; then
-            echo "peak memory: $((peak / 1024 / 1024)) MiB"
-          else
-            echo "peak memory: unavailable (no readable cgroup v2 memory.peak)"
-          fi
-          free -g || true
      - name: Upload artifact
        uses: actions/upload-artifact@v7
        with:
@@ -338,7 +293,7 @@ jobs:
          - target: aarch64-unknown-linux-gnu
            host: ubuntu-2404-8x-arm64
        node:
-          - '22'
+          - '20'
    runs-on: ${{ matrix.settings.host }}
    defaults:
      run:
@@ -384,9 +339,9 @@ jobs:
      - name: Move built files
        run: cp dist/native.d.ts dist/native.js dist/*.node lancedb/
      - name: Test bindings
-        # Invoke the installed jest binary directly; the pnpm shim is set up
-        # against the install-phase Node, not the version selected above.
-        run: node_modules/.bin/jest --verbose
+        # Invoke jest directly because pnpm 11 itself requires Node 22+
+        # while the matrix tests on older Node versions.
+        run: npx jest --verbose
  publish:
    name: Publish
    runs-on: ubuntu-latest
@@ -0,0 +1,22 @@
+name: Update package-lock.json
+
+on:
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  publish:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v6
+        with:
+          ref: main
+          persist-credentials: false
+          fetch-depth: 0
+          lfs: true
+      - uses: ./.github/workflows/update_package_lock
+        with:
+          github_token: ${{ secrets.LANCEDB_RELEASE_TOKEN }}
@@ -0,0 +1,22 @@
+name: Update NodeJs package-lock.json
+
+on:
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  publish:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v6
+        with:
+          ref: main
+          persist-credentials: false
+          fetch-depth: 0
+          lfs: true
+      - uses: ./.github/workflows/update_package_lock_nodejs
+        with:
+          github_token: ${{ secrets.LANCEDB_RELEASE_TOKEN }}
@@ -20,10 +20,7 @@ repos:
    hooks:
      - id: local-biome-check
        name: biome check
-        # Use the biome from nodejs/package.json rather than a separately
-        # pinned one: the two drifted apart and disagreed on formatting, so
-        # this hook rejected code that `pnpm lint` accepted.
-        entry: nodejs/node_modules/.bin/biome check --config-path nodejs/biome.json nodejs/
+        entry: npx @biomejs/biome@1.8.3 check --config-path nodejs/biome.json nodejs/
        language: system
        types: [text]
        files: "nodejs/.*"
@@ -38,7 +38,7 @@ Before committing changes, run formatting for every language you touched. At min
 * Rust changes: run `cargo fmt --all`.
 * Python changes: run `ruff format .` and `ruff check .` from the repository root,
  and run targeted tests through `cd python && uv run ...`.
-* TypeScript changes: run the relevant `pnpm` lint, format, build, and docs commands in `nodejs`.
+* TypeScript changes: run the relevant `npm`/`pnpm` lint, format, build, and docs commands in `nodejs`.

 Before creating a PR, the exact value passed to `gh pr create --title` must follow
 Conventional Commits, such as `fix: support nested field paths in native index creation`
@@ -101,12 +101,12 @@ Python bindings changes:
 TypeScript bindings changes:

 1. Add napi-rs method binding on `Table` in `nodejs/src/table.rs`.
-2. Run `pnpm build` to generate TypeScript definitions.
+2. Run `npm run build` to generate TypeScript definitions.
 3. Add typescript method on abstract class `Table` in `nodejs/src/table.ts`.
 4. Add concrete method on `LocalTable` class in `nodejs/src/native_table.ts`.
    * Note: despite the name, this class is also used for remote tables.
 5. Add test in `nodejs/__test__/table.test.ts`.
-6. Run `pnpm run docs` to generate TypeScript documentation.
+6. Run `npm run docs` to generate TypeScript documentation.

 ## Python API reference

@@ -535,9 +535,9 @@ dependencies = [

 [[package]]
 name = "async-trait"
-version = "0.1.92"
+version = "0.1.91"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "82f6aeea286b8eb4dd3431a1be1b59d290ace00f5bfd8e2a159bc2a05e2c1667"
+checksum = "ae36dc4177970ef04fde5178d3e2429882def40e57a451f919c098f72baa6cec"
 dependencies = [
 "proc-macro2",
 "quote",
@@ -1443,9 +1443,9 @@ checksum = "175812e0be2bccb6abe50bb8d566126198344f707e304f45c648fd8f2cc0365e"

 [[package]]
 name = "bytemuck"
-version = "1.25.2"
+version = "1.25.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "95832e849adfb21180ccb6826a99da14e5d266ae5c2e668e1602cf234f153797"
+checksum = "c8efb64bd706a16a1bdde310ae86b351e4d21550d98d056f22f8a7f7a2183fec"
 dependencies = [
 "bytemuck_derive",
 ]
@@ -1597,9 +1597,9 @@ checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724"

 [[package]]
 name = "chacha20"
-version = "0.10.2"
+version = "0.10.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06"
+checksum = "6f8d983286843e49675a4b7a2d174efe136dc93a18d69130dd18198a6c167601"
 dependencies = [
 "cfg-if 1.0.4",
 "cpufeatures 0.3.0",
@@ -3455,8 +3455,8 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"

 [[package]]
 name = "fsst"
-version = "12.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.8#4bcecd313e607713a62625d9d7c231fd70043d14"
+version = "12.0.0-beta.4"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.4#474ae88dd75c437d5d506859b627510978e502e9"
 dependencies = [
 "arrow-array",
 "rand 0.9.5",
@@ -4815,8 +4815,8 @@ checksum = "e037a2e1d8d5fdbd49b16a4ea09d5d6401c1f29eca5ff29d03d3824dba16256a"

 [[package]]
 name = "lance"
-version = "12.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.8#4bcecd313e607713a62625d9d7c231fd70043d14"
+version = "12.0.0-beta.4"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.4#474ae88dd75c437d5d506859b627510978e502e9"
 dependencies = [
 "arc-swap",
 "arrow",
@@ -4888,8 +4888,8 @@ dependencies = [

 [[package]]
 name = "lance-arrow"
-version = "12.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.8#4bcecd313e607713a62625d9d7c231fd70043d14"
+version = "12.0.0-beta.4"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.4#474ae88dd75c437d5d506859b627510978e502e9"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4911,7 +4911,7 @@ dependencies = [
 [[package]]
 name = "lance-arrow-scalar"
 version = "58.0.0"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.8#4bcecd313e607713a62625d9d7c231fd70043d14"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.4#474ae88dd75c437d5d506859b627510978e502e9"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4925,7 +4925,7 @@ dependencies = [
 [[package]]
 name = "lance-arrow-stats"
 version = "58.0.0"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.8#4bcecd313e607713a62625d9d7c231fd70043d14"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.4#474ae88dd75c437d5d506859b627510978e502e9"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -4934,8 +4934,8 @@ dependencies = [

 [[package]]
 name = "lance-bitpacking"
-version = "12.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.8#4bcecd313e607713a62625d9d7c231fd70043d14"
+version = "12.0.0-beta.4"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.4#474ae88dd75c437d5d506859b627510978e502e9"
 dependencies = [
 "arrayref",
 "crunchy",
@@ -4945,8 +4945,8 @@ dependencies = [

 [[package]]
 name = "lance-core"
-version = "12.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.8#4bcecd313e607713a62625d9d7c231fd70043d14"
+version = "12.0.0-beta.4"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.4#474ae88dd75c437d5d506859b627510978e502e9"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4983,8 +4983,8 @@ dependencies = [

 [[package]]
 name = "lance-datafusion"
-version = "12.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.8#4bcecd313e607713a62625d9d7c231fd70043d14"
+version = "12.0.0-beta.4"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.4#474ae88dd75c437d5d506859b627510978e502e9"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5000,7 +5000,6 @@ dependencies = [
 "datafusion-functions",
 "datafusion-physical-expr",
 "futures",
- "half",
 "jsonb",
 "lance-arrow",
 "lance-core",
@@ -5014,8 +5013,8 @@ dependencies = [

 [[package]]
 name = "lance-datagen"
-version = "12.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.8#4bcecd313e607713a62625d9d7c231fd70043d14"
+version = "12.0.0-beta.4"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.4#474ae88dd75c437d5d506859b627510978e502e9"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5032,8 +5031,8 @@ dependencies = [

 [[package]]
 name = "lance-derive"
-version = "12.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.8#4bcecd313e607713a62625d9d7c231fd70043d14"
+version = "12.0.0-beta.4"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.4#474ae88dd75c437d5d506859b627510978e502e9"
 dependencies = [
 "proc-macro2",
 "quote",
@@ -5042,8 +5041,8 @@ dependencies = [

 [[package]]
 name = "lance-encoding"
-version = "12.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.8#4bcecd313e607713a62625d9d7c231fd70043d14"
+version = "12.0.0-beta.4"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.4#474ae88dd75c437d5d506859b627510978e502e9"
 dependencies = [
 "arrow-arith",
 "arrow-array",
@@ -5076,8 +5075,8 @@ dependencies = [

 [[package]]
 name = "lance-file"
-version = "12.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.8#4bcecd313e607713a62625d9d7c231fd70043d14"
+version = "12.0.0-beta.4"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.4#474ae88dd75c437d5d506859b627510978e502e9"
 dependencies = [
 "arrow-arith",
 "arrow-array",
@@ -5108,8 +5107,8 @@ dependencies = [

 [[package]]
 name = "lance-index"
-version = "12.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.8#4bcecd313e607713a62625d9d7c231fd70043d14"
+version = "12.0.0-beta.4"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.4#474ae88dd75c437d5d506859b627510978e502e9"
 dependencies = [
 "arc-swap",
 "arrow",
@@ -5173,8 +5172,8 @@ dependencies = [

 [[package]]
 name = "lance-index-core"
-version = "12.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.8#4bcecd313e607713a62625d9d7c231fd70043d14"
+version = "12.0.0-beta.4"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.4#474ae88dd75c437d5d506859b627510978e502e9"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -5196,8 +5195,8 @@ dependencies = [

 [[package]]
 name = "lance-io"
-version = "12.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.8#4bcecd313e607713a62625d9d7c231fd70043d14"
+version = "12.0.0-beta.4"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.4#474ae88dd75c437d5d506859b627510978e502e9"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5237,8 +5236,8 @@ dependencies = [

 [[package]]
 name = "lance-linalg"
-version = "12.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.8#4bcecd313e607713a62625d9d7c231fd70043d14"
+version = "12.0.0-beta.4"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.4#474ae88dd75c437d5d506859b627510978e502e9"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -5252,8 +5251,8 @@ dependencies = [

 [[package]]
 name = "lance-namespace"
-version = "12.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.8#4bcecd313e607713a62625d9d7c231fd70043d14"
+version = "12.0.0-beta.4"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.4#474ae88dd75c437d5d506859b627510978e502e9"
 dependencies = [
 "arrow",
 "async-trait",
@@ -5265,8 +5264,8 @@ dependencies = [

 [[package]]
 name = "lance-namespace-impls"
-version = "12.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.8#4bcecd313e607713a62625d9d7c231fd70043d14"
+version = "12.0.0-beta.4"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.4#474ae88dd75c437d5d506859b627510978e502e9"
 dependencies = [
 "arrow",
 "arrow-ipc",
@@ -5305,9 +5304,9 @@ dependencies = [

 [[package]]
 name = "lance-namespace-reqwest-client"
-version = "0.11.1"
+version = "0.11.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1d06b1fbb5d41f93bc652b61e2872af92e8a6c5f6b4ce8839a8ecfa05365d359"
+checksum = "0a030196da1c994b63a96a4f0bf5b0cfa459fe6dadc9e962320246ca328da22a"
 dependencies = [
 "reqwest 0.12.28",
 "serde",
@@ -5319,8 +5318,8 @@ dependencies = [

 [[package]]
 name = "lance-select"
-version = "12.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.8#4bcecd313e607713a62625d9d7c231fd70043d14"
+version = "12.0.0-beta.4"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.4#474ae88dd75c437d5d506859b627510978e502e9"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -5334,8 +5333,8 @@ dependencies = [

 [[package]]
 name = "lance-table"
-version = "12.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.8#4bcecd313e607713a62625d9d7c231fd70043d14"
+version = "12.0.0-beta.4"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.4#474ae88dd75c437d5d506859b627510978e502e9"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5375,8 +5374,8 @@ dependencies = [

 [[package]]
 name = "lance-testing"
-version = "12.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.8#4bcecd313e607713a62625d9d7c231fd70043d14"
+version = "12.0.0-beta.4"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.4#474ae88dd75c437d5d506859b627510978e502e9"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -5389,8 +5388,8 @@ dependencies = [

 [[package]]
 name = "lance-tokenizer"
-version = "12.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.8#4bcecd313e607713a62625d9d7c231fd70043d14"
+version = "12.0.0-beta.4"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.4#474ae88dd75c437d5d506859b627510978e502e9"
 dependencies = [
 "frostem",
 "icu_segmenter",
@@ -5403,7 +5402,7 @@ dependencies = [

 [[package]]
 name = "lancedb"
-version = "0.39.0-beta.0"
+version = "0.38.0-beta.11"
 dependencies = [
 "ahash",
 "anyhow",
@@ -5491,7 +5490,7 @@ dependencies = [

 [[package]]
 name = "lancedb-nodejs"
-version = "0.39.0-beta.0"
+version = "0.38.0-beta.11"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -5516,7 +5515,7 @@ dependencies = [

 [[package]]
 name = "lancedb-python"
-version = "0.39.0-beta.0"
+version = "0.38.0-beta.11"
 dependencies = [
 "arrow",
 "async-trait",
@@ -5749,9 +5748,9 @@ dependencies = [

 [[package]]
 name = "log"
-version = "0.4.34"
+version = "0.4.33"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6"
+checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad"

 [[package]]
 name = "loom"
@@ -6002,9 +6001,9 @@ dependencies = [

 [[package]]
 name = "moka"
-version = "0.12.16"
+version = "0.12.15"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4293f18e7567a1caf3c584855554377025c65e0aa445344d04171f5ad63d19b9"
+checksum = "957228ad12042ee839f93c8f257b62b4c0ab5eaae1d4fa60de53b27c9d7c5046"
 dependencies = [
 "async-lock",
 "crossbeam-channel",
@@ -6098,15 +6097,14 @@ dependencies = [

 [[package]]
 name = "napi"
-version = "3.12.2"
+version = "3.11.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "58c5f4d5375213fdb7be2655e152386e82f026f9a5ba36a75556e11359aafe09"
+checksum = "de33522036981030a75c231829566bc63414e08101a6f5ff4ac6cef19c8e0941"
 dependencies = [
 "bitflags 2.11.1",
 "chrono",
 "ctor 1.0.12",
 "futures",
- "libc",
 "napi-build",
 "napi-sys",
 "nohash-hasher",
@@ -6118,15 +6116,15 @@ dependencies = [

 [[package]]
 name = "napi-build"
-version = "2.4.1"
+version = "2.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "60fdf9b392c50e7c4170fa633bd909490ed7835cea4c046776d1a4dd8d2ae0ab"
+checksum = "5282704fbe8d49b0cf8b08e3f33233416a528658f205c7e5ace63b582de0b11c"

 [[package]]
 name = "napi-derive"
-version = "3.6.3"
+version = "3.6.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0fa55ea69990c90b888e9e77044410e304ce7f35de599dc6d0b5c1923d2e59af"
+checksum = "4d5c9c02556ea6dc99dffd36c1ce60141411657438501a125b675776d011ce92"
 dependencies = [
 "convert_case",
 "ctor 1.0.12",
@@ -6138,9 +6136,9 @@ dependencies = [

 [[package]]
 name = "napi-derive-backend"
-version = "6.1.2"
+version = "6.1.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "df4056ac7c18e4438ccf0edaed4340ca0d269278c8ec19284f7b23cb039fd0ae"
+checksum = "d60b5d773ad46c698c8cc2cd9fde0b283d39cbb7f71c04bee633c7bdba4423bd"
 dependencies = [
 "convert_case",
 "proc-macro2",
@@ -8603,9 +8601,9 @@ dependencies = [

 [[package]]
 name = "roaring"
-version = "0.11.5"
+version = "0.11.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "18bd8a37d17a58532776dcdf6041ce64929adca78e8489d5cacbafe99229d3e1"
+checksum = "1dedc5658c6ecb3bdb5ef5f3295bb9253f42dcf3fd1402c03f6b1f7659c3c4a9"
 dependencies = [
 "bytemuck",
 "byteorder",
@@ -9065,9 +9063,9 @@ dependencies = [

 [[package]]
 name = "serde_with"
-version = "3.22.0"
+version = "3.21.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ee78f1fbe43ac4a0e47aadb3dbd357b69eb0d3793e948624cd03dd2750ab1c0a"
+checksum = "76a5c54c7310e7b8b9577c286d7e399ddd876c3e12b3ed917a8aabc4b96e9e8c"
 dependencies = [
 "base64 0.22.1",
 "bs58",
@@ -9075,7 +9073,6 @@ dependencies = [
 "hex",
 "indexmap 1.9.3",
 "indexmap 2.14.0",
- "jiff",
 "schemars 0.9.0",
 "schemars 1.2.1",
 "serde_core",
@@ -9086,9 +9083,9 @@ dependencies = [

 [[package]]
 name = "serde_with_macros"
-version = "3.22.0"
+version = "3.21.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8705578779c2b6bd90d84d66eb2e206b708b1a4d7b9f17641b293545bf1c7e46"
+checksum = "84d57bc0c8b9a17920c178daa6bb924850d54a9c97ab45194bb8c17ad66bb660"
 dependencies = [
 "darling 0.23.0",
 "proc-macro2",
@@ -10455,9 +10452,9 @@ checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821"

 [[package]]
 name = "uuid"
-version = "1.26.0"
+version = "1.24.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b5772d71c9be8a8a6ac2117d949c5b224c1b72241bb611d9a3012edcf8af7812"
+checksum = "bf3923a6f5c4c6382e0b653c4117f48d631ea17f38ed86e2a828e6f7412f5239"
 dependencies = [
 "getrandom 0.4.2",
 "js-sys",
@@ -13,20 +13,20 @@ categories = ["database-implementations"]
 rust-version = "1.91.0"

 [workspace.dependencies]
-lance = { "version" = "=12.0.0-beta.8", default-features = false, "tag" = "v12.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
-lance-core = { "version" = "=12.0.0-beta.8", "tag" = "v12.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
-lance-datagen = { "version" = "=12.0.0-beta.8", "tag" = "v12.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
-lance-file = { "version" = "=12.0.0-beta.8", "tag" = "v12.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
-lance-io = { "version" = "=12.0.0-beta.8", default-features = false, "tag" = "v12.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
-lance-index = { "version" = "=12.0.0-beta.8", "tag" = "v12.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
-lance-linalg = { "version" = "=12.0.0-beta.8", "tag" = "v12.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace = { "version" = "=12.0.0-beta.8", "tag" = "v12.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace-impls = { "version" = "=12.0.0-beta.8", default-features = false, "tag" = "v12.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
-lance-table = { "version" = "=12.0.0-beta.8", "tag" = "v12.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
-lance-testing = { "version" = "=12.0.0-beta.8", "tag" = "v12.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
-lance-datafusion = { "version" = "=12.0.0-beta.8", "tag" = "v12.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
-lance-encoding = { "version" = "=12.0.0-beta.8", "tag" = "v12.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
-lance-arrow = { "version" = "=12.0.0-beta.8", "tag" = "v12.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance = { "version" = "=12.0.0-beta.4", default-features = false, "tag" = "v12.0.0-beta.4", "git" = "https://github.com/lance-format/lance.git" }
+lance-core = { "version" = "=12.0.0-beta.4", "tag" = "v12.0.0-beta.4", "git" = "https://github.com/lance-format/lance.git" }
+lance-datagen = { "version" = "=12.0.0-beta.4", "tag" = "v12.0.0-beta.4", "git" = "https://github.com/lance-format/lance.git" }
+lance-file = { "version" = "=12.0.0-beta.4", "tag" = "v12.0.0-beta.4", "git" = "https://github.com/lance-format/lance.git" }
+lance-io = { "version" = "=12.0.0-beta.4", default-features = false, "tag" = "v12.0.0-beta.4", "git" = "https://github.com/lance-format/lance.git" }
+lance-index = { "version" = "=12.0.0-beta.4", "tag" = "v12.0.0-beta.4", "git" = "https://github.com/lance-format/lance.git" }
+lance-linalg = { "version" = "=12.0.0-beta.4", "tag" = "v12.0.0-beta.4", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace = { "version" = "=12.0.0-beta.4", "tag" = "v12.0.0-beta.4", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace-impls = { "version" = "=12.0.0-beta.4", default-features = false, "tag" = "v12.0.0-beta.4", "git" = "https://github.com/lance-format/lance.git" }
+lance-table = { "version" = "=12.0.0-beta.4", "tag" = "v12.0.0-beta.4", "git" = "https://github.com/lance-format/lance.git" }
+lance-testing = { "version" = "=12.0.0-beta.4", "tag" = "v12.0.0-beta.4", "git" = "https://github.com/lance-format/lance.git" }
+lance-datafusion = { "version" = "=12.0.0-beta.4", "tag" = "v12.0.0-beta.4", "git" = "https://github.com/lance-format/lance.git" }
+lance-encoding = { "version" = "=12.0.0-beta.4", "tag" = "v12.0.0-beta.4", "git" = "https://github.com/lance-format/lance.git" }
+lance-arrow = { "version" = "=12.0.0-beta.4", "tag" = "v12.0.0-beta.4", "git" = "https://github.com/lance-format/lance.git" }
 lancedb = { path = "rust/lancedb", default-features = false }
 ahash = "0.8"
 # Note that this one does not include pyarrow
@@ -5,5 +5,5 @@ licenses:
 	cd python && cargo about generate ../about.hbs -o RUST_THIRD_PARTY_LICENSES.html -c ../about.toml
 	cd python && uv sync --all-extras && uv tool run pip-licenses --python .venv/bin/python --format=markdown --with-urls --output-file=PYTHON_THIRD_PARTY_LICENSES.md
 	cd nodejs && cargo about generate ../about.hbs -o RUST_THIRD_PARTY_LICENSES.html -c ../about.toml
-	cd nodejs && pnpm dlx license-checker@25 --markdown --out NODEJS_THIRD_PARTY_LICENSES.md
+	cd nodejs && npx license-checker --markdown --out NODEJS_THIRD_PARTY_LICENSES.md
 	cd java && ./mvnw license:aggregate-add-third-party -q
@@ -12,12 +12,16 @@ done
 # This updates the lockfile without building
 cargo metadata --quiet > /dev/null

+pushd nodejs || exit 1
+npm install --package-lock-only --silent
+popd
+
 if git diff --quiet --exit-code; then
  echo "No lockfile changes to commit; skipping amend."
 elif $AMEND; then
-  git add Cargo.lock
+  git add Cargo.lock nodejs/package-lock.json
  git commit --amend --no-edit
 else
-  git add Cargo.lock
+  git add Cargo.lock nodejs/package-lock.json
  git commit -m "Update lockfiles"
 fi
@@ -131,13 +131,18 @@ allow = [
    "BSD-3-Clause",
    "ISC",
    "Unicode-3.0",
+    "Unicode-DFS-2016",
    "Zlib",
    "CC0-1.0",
    "MPL-2.0",
    "BSL-1.0",
+    "OpenSSL",
    # 0BSD ("BSD Zero Clause") is effectively public domain — no attribution
    # required. Pulled in by `mock_instant`.
    "0BSD",
+    # bzip2-1.0.6 is the permissive upstream bzip2 license (BSD-like). Pulled
+    # in by `libbz2-rs-sys`, the pure-Rust bzip2 implementation.
+    "bzip2-1.0.6",
    # CDLA-Permissive-2.0 is a permissive data license used by `webpki-roots`
    # for the Mozilla CA root bundle. Data-only, distribution-compatible.
    "CDLA-Permissive-2.0",
@@ -145,7 +150,12 @@ allow = [
 confidence-threshold = 0.8
 # Per-crate license exceptions: allow a license for a specific crate only,
 # rather than globally via the `allow` list above.
-exceptions = []
+exceptions = [
+    # CDDL-1.0 (copyleft) is pulled in only as a dev/profiling dependency via
+    # `inferno` -> `pprof` -> `lance-testing`; it is a test dependency that we
+    # do not distribute, so scope the allowance to `inferno` alone.
+    { allow = ["CDDL-1.0"], crate = "inferno" },
+]
 # Crates whose license cannot be determined from Cargo metadata but whose
 # license we've manually confirmed from upstream. Keep this list minimal.
 [[licenses.clarify]]
@@ -47,24 +47,22 @@ pytest -vv python/tests/docs

 ### Checking typescript examples

-The examples depend on `@lancedb/lancedb` at `file:../dist`, so the package must be
-built before running the tests. This uses pnpm; see the
-[Typescript contributing guide](../nodejs/CONTRIBUTING.md) for the toolchain setup.
+The `@lancedb/lancedb` package must be built before running the tests:

 ```shell
 pushd nodejs
-pnpm install
-pnpm build
+npm ci
+npm run build
 popd
 ```

-Then you can run the examples by going to the `nodejs/examples` directory, which is a
-separate pnpm package with its own lockfile:
+Then you can run the examples by going to the `nodejs/examples` directory and
+running the tests like a normal npm package:

 ```shell
 pushd nodejs/examples
-pnpm install
-pnpm test
+npm ci
+npm test
 popd
 ```

@@ -86,7 +84,6 @@ The new files should be checked into the repository.

 ```shell
 pushd nodejs
-# `pnpm docs` would invoke pnpm's built-in `docs` command, not the script.
-pnpm run docs
+npm run docs
 popd
 ```
@@ -0,0 +1,135 @@
+{
+  "name": "lancedb-docs-test",
+  "version": "1.0.0",
+  "lockfileVersion": 3,
+  "requires": true,
+  "packages": {
+    "": {
+      "name": "lancedb-docs-test",
+      "version": "1.0.0",
+      "license": "Apache 2",
+      "dependencies": {
+        "apache-arrow": "file:../node/node_modules/apache-arrow",
+        "vectordb": "file:../node"
+      },
+      "devDependencies": {
+        "@types/node": "^20.11.8",
+        "typescript": "^5.3.3"
+      }
+    },
+    "../node": {
+      "name": "vectordb",
+      "version": "0.21.2-beta.0",
+      "cpu": [
+        "x64",
+        "arm64"
+      ],
+      "license": "Apache-2.0",
+      "os": [
+        "darwin",
+        "linux",
+        "win32"
+      ],
+      "dependencies": {
+        "@neon-rs/load": "^0.0.74",
+        "axios": "^1.4.0"
+      },
+      "devDependencies": {
+        "@neon-rs/cli": "^0.0.160",
+        "@types/chai": "^4.3.4",
+        "@types/chai-as-promised": "^7.1.5",
+        "@types/mocha": "^10.0.1",
+        "@types/node": "^18.16.2",
+        "@types/sinon": "^10.0.15",
+        "@types/temp": "^0.9.1",
+        "@types/uuid": "^9.0.3",
+        "@typescript-eslint/eslint-plugin": "^5.59.1",
+        "apache-arrow-old": "npm:apache-arrow@13.0.0",
+        "cargo-cp-artifact": "^0.1",
+        "chai": "^4.3.7",
+        "chai-as-promised": "^7.1.1",
+        "eslint": "^8.39.0",
+        "eslint-config-standard-with-typescript": "^34.0.1",
+        "eslint-plugin-import": "^2.26.0",
+        "eslint-plugin-n": "^15.7.0",
+        "eslint-plugin-promise": "^6.1.1",
+        "mocha": "^10.2.0",
+        "openai": "^4.24.1",
+        "sinon": "^15.1.0",
+        "temp": "^0.9.4",
+        "ts-node": "^10.9.1",
+        "ts-node-dev": "^2.0.0",
+        "typedoc": "^0.24.7",
+        "typedoc-plugin-markdown": "^3.15.3",
+        "typescript": "^5.1.0",
+        "uuid": "^9.0.0"
+      },
+      "optionalDependencies": {
+        "@lancedb/vectordb-darwin-arm64": "0.21.2-beta.0",
+        "@lancedb/vectordb-darwin-x64": "0.21.2-beta.0",
+        "@lancedb/vectordb-linux-arm64-gnu": "0.21.2-beta.0",
+        "@lancedb/vectordb-linux-x64-gnu": "0.21.2-beta.0",
+        "@lancedb/vectordb-win32-x64-msvc": "0.21.2-beta.0"
+      },
+      "peerDependencies": {
+        "@apache-arrow/ts": "^14.0.2",
+        "apache-arrow": "^14.0.2"
+      }
+    },
+    "../node/node_modules/apache-arrow": {
+      "version": "14.0.2",
+      "license": "Apache-2.0",
+      "dependencies": {
+        "@types/command-line-args": "5.2.0",
+        "@types/command-line-usage": "5.0.2",
+        "@types/node": "20.3.0",
+        "@types/pad-left": "2.1.1",
+        "command-line-args": "5.2.1",
+        "command-line-usage": "7.0.1",
+        "flatbuffers": "23.5.26",
+        "json-bignum": "^0.0.3",
+        "pad-left": "^2.1.0",
+        "tslib": "^2.5.3"
+      },
+      "bin": {
+        "arrow2csv": "bin/arrow2csv.js"
+      }
+    },
+    "node_modules/@types/node": {
+      "version": "20.11.8",
+      "resolved": "https://registry.npmjs.org/@types/node/-/node-20.11.8.tgz",
+      "integrity": "sha512-i7omyekpPTNdv4Jb/Rgqg0RU8YqLcNsI12quKSDkRXNfx7Wxdm6HhK1awT3xTgEkgxPn3bvnSpiEAc7a7Lpyow==",
+      "dev": true,
+      "dependencies": {
+        "undici-types": "~5.26.4"
+      }
+    },
+    "node_modules/apache-arrow": {
+      "resolved": "../node/node_modules/apache-arrow",
+      "link": true
+    },
+    "node_modules/typescript": {
+      "version": "5.3.3",
+      "resolved": "https://registry.npmjs.org/typescript/-/typescript-5.3.3.tgz",
+      "integrity": "sha512-pXWcraxM0uxAS+tN0AG/BF2TyqmHO014Z070UsJ+pFvYuRSq8KH8DmWpnbXe0pEPDHXZV3FcAbJkijJ5oNEnWw==",
+      "dev": true,
+      "bin": {
+        "tsc": "bin/tsc",
+        "tsserver": "bin/tsserver"
+      },
+      "engines": {
+        "node": ">=14.17"
+      }
+    },
+    "node_modules/undici-types": {
+      "version": "5.26.5",
+      "resolved": "https://registry.npmjs.org/undici-types/-/undici-types-5.26.5.tgz",
+      "integrity": "sha512-JlCMO+ehdEIKqlFxk6IfVoAUVmgz7cU7zD/h9XZ0qzeosSHmUJVOzSQvvYSYWXkFXC+IfLKSIffhv0sVZup6pA==",
+      "dev": true
+    },
+    "node_modules/vectordb": {
+      "resolved": "../node",
+      "link": true
+    }
+  }
+}
@@ -0,0 +1,20 @@
+{
+  "name": "lancedb-docs-test",
+  "version": "1.0.0",
+  "description": "auto-generated tests from doc",
+  "author": "dev@lancedb.com",
+  "license": "Apache 2",
+  "dependencies": {
+    "apache-arrow": "file:../node/node_modules/apache-arrow",
+    "vectordb": "file:../node"
+  },
+  "scripts": {
+    "build": "tsc -b && cd ../node && npm run build-release",
+    "example": "npm run build && node",
+    "test": "npm run build && ls dist/*.js | xargs -n 1 node"
+  },
+  "devDependencies": {
+    "@types/node": "^20.11.8",
+    "typescript": "^5.3.3"
+  }
+}
@@ -14,7 +14,7 @@ Add the following dependency to your `pom.xml`:
 <dependency>
    <groupId>com.lancedb</groupId>
    <artifactId>lancedb-core</artifactId>
-    <version>0.39.0-beta.0</version>
+    <version>0.38.0-beta.11</version>
 </dependency>
 ```

@@ -50,16 +50,6 @@ projections: [string, string][];

 ***

-### sourceNamespace
-
-```ts
-sourceNamespace: string[];
-```
-
-Namespace holding the source table; empty is the root namespace.
-
-***
-
 ### sourceTable

 ```ts
@@ -223,14 +223,10 @@ tokens = list(
 Blob columns store large binary values out of line so they can be read lazily
 instead of being materialized with the rest of the row.

-`lancedb.BlobType` is `lance.blob.BlobType` when pylance is installed. Without
-pylance, LanceDB uses a matching `lance.blob.v2` extension type so blob columns
-still work. Queries return descriptors. Call
-[`fetch_blob_files`][lancedb.table.Table.fetch_blob_files] for lazy reads or
-[`fetch_blobs`][lancedb.table.Table.fetch_blobs] for eager bytes.
-
 ::: lancedb.blob

+::: lancedb.BlobType
+
 ::: lancedb._blob.BlobFile
    options:
      show_root_full_path: false
@@ -0,0 +1,17 @@
+{
+  "include": [
+    "src/*.ts",
+  ],
+  "compilerOptions": {
+    "target": "es2022",
+    "module": "nodenext",
+    "declaration": true,
+    "outDir": "./dist",
+    "strict": true,
+    "allowJs": true,
+    "resolveJsonModule": true,
+  },
+  "exclude": [
+    "./dist/*",
+  ]
+}
@@ -8,7 +8,7 @@
    <parent>
      <groupId>com.lancedb</groupId>
      <artifactId>lancedb-parent</artifactId>
-      <version>0.39.0-beta.0</version>
+      <version>0.38.0-beta.11</version>
      <relativePath>../pom.xml</relativePath>
    </parent>

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

    <groupId>com.lancedb</groupId>
    <artifactId>lancedb-parent</artifactId>
-    <version>0.39.0-beta.0</version>
+    <version>0.38.0-beta.11</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>12.0.0-beta.8</lance-core.version>
+        <lance-core.version>12.0.0-beta.4</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.39.0-beta.0"
+version = "0.38.0-beta.11"
 publish = false
 license.workspace = true
 description.workspace = true
@@ -48,28 +48,6 @@ describe("materialized views", () => {
    expect(definitionFromMetadata(safe, "v").limit).toBe(42);
  });

-  it("reads the namespaced select kind and refuses unknown kinds", () => {
-    // "namespaced_select" is the namespaced form of "select": same shape, a
-    // separate kind so readers that predate it refuse instead of resolving
-    // the source at the root.
-    const namespaced = new Map([
-      [
-        DEFINITION_META_KEY,
-        '{"kind":"namespaced_select","source_table":"people","source_namespace":["ns"]}',
-      ],
-    ]);
-    const definition = definitionFromMetadata(namespaced, "v");
-    expect(definition.sourceTable).toBe("people");
-    expect(definition.sourceNamespace).toEqual(["ns"]);
-
-    const unknown = new Map([
-      [DEFINITION_META_KEY, '{"kind":"select_v3","source_table":"people"}'],
-    ]);
-    expect(() => definitionFromMetadata(unknown, "v")).toThrow(
-      /cannot refresh/,
-    );
-  });
-
  it("creates, refreshes and queries a view", async () => {
    const view = await db.createMaterializedView("adults", "people", {
      select: ["name", ["shout", "upper(name)"]],
@@ -5,8 +5,8 @@ import packageJson = require("../package.json");

 describe("package metadata", () => {
  it("requires Node.js type declarations compatible with the runtime", () => {
-    expect(packageJson.engines.node).toBe(">= 22");
-    expect(packageJson.peerDependencies["@types/node"]).toBe(">=22");
+    expect(packageJson.engines.node).toBe(">= 18");
+    expect(packageJson.peerDependencies["@types/node"]).toBe(">=18");
    expect(packageJson.peerDependenciesMeta["@types/node"]).toEqual({
      optional: true,
    });
@@ -3,7 +3,6 @@

 import * as http from "http";
 import { RequestListener } from "http";
-import packageJson = require("../package.json");
 import {
  ClientConfig,
  Connection,
@@ -71,13 +70,7 @@ async function withMockDatabase(
  try {
    await callback(db);
  } finally {
-    // `close()` alone leaves the port bound until keep-alive sockets drain, so
-    // a single failing test would cascade into EADDRINUSE for every test after
-    // it. Destroy the connections and wait for the port to actually be free.
-    await new Promise<void>((resolve) => {
-      server.closeAllConnections();
-      server.close(() => resolve());
-    });
+    server.close();
  }
 }

@@ -138,7 +131,7 @@ describe("remote connection", () => {
      (req, res) => {
        expect(req.headers["x-api-key"]).toEqual("fake");
        expect(req.headers["user-agent"]).toEqual(
-          `LanceDB-Node-Client/${packageJson.version}`,
+          `LanceDB-Node-Client/${process.env.npm_package_version}`,
        );

        const body = JSON.stringify({ tables: [] });
@@ -8,8 +8,7 @@
    "//1": "--experimental-vm-modules is needed to run jest with sentence-transformers",
    "//2": "--testEnvironment is needed to run jest with sentence-transformers",
    "//3": "See: https://github.com/huggingface/transformers.js/issues/57",
-    "//4": "jest is invoked by its JS entry, not node_modules/.bin/jest: under pnpm that path is a shell shim, which `node` cannot execute",
-    "test": "node --experimental-vm-modules node_modules/jest/bin/jest.js --testEnvironment jest-environment-node-single-context --verbose",
+    "test": "node --experimental-vm-modules node_modules/.bin/jest --testEnvironment jest-environment-node-single-context --verbose",
    "lint": "biome check *.ts && biome format *.ts",
    "lint-ci": "biome ci .",
    "lint-fix": "biome check --write *.ts && pnpm format",
@@ -19,8 +19,6 @@ export interface MaterializedViewDefinition {
  limit?: number;
  /** Source columns the projections and filter read. */
  inputs: string[];
-  /** Namespace holding the source table; empty is the root namespace. */
-  sourceNamespace: string[];
 }

 /**
@@ -80,8 +78,7 @@ export function definitionFromMetadata(
  }
  // biome-ignore lint/suspicious/noExplicitAny: raw JSON
  const value: any = JSON.parse(raw);
-  // "namespaced_select" keeps older readers from resolving the source at root.
-  if (value.kind !== "select" && value.kind !== "namespaced_select") {
+  if (value.kind !== "select") {
    throw new Error(
      `materialized view '${name}' is defined by '${value.kind}', which this ` +
        "version of lancedb cannot refresh",
@@ -106,7 +103,6 @@ export function definitionFromMetadata(
    filter: value.filter ?? undefined,
    limit,
    inputs: value.inputs ?? [],
-    sourceNamespace: value.source_namespace ?? [],
  };
 }

@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-darwin-arm64",
-	"version": "0.39.0-beta.0",
+	"version": "0.38.0-beta.11",
 	"os": ["darwin"],
 	"cpu": ["arm64"],
 	"main": "lancedb.darwin-arm64.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-gnu",
-	"version": "0.39.0-beta.0",
+	"version": "0.38.0-beta.11",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-gnu.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-musl",
-	"version": "0.39.0-beta.0",
+	"version": "0.38.0-beta.11",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-musl.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-gnu",
-	"version": "0.39.0-beta.0",
+	"version": "0.38.0-beta.11",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-gnu.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-musl",
-	"version": "0.39.0-beta.0",
+	"version": "0.38.0-beta.11",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-musl.node",
@@ -1,6 +1,6 @@
 {
  "name": "@lancedb/lancedb-win32-arm64-msvc",
-  "version": "0.39.0-beta.0",
+  "version": "0.38.0-beta.11",
  "os": [
    "win32"
  ],
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-win32-x64-msvc",
-	"version": "0.39.0-beta.0",
+	"version": "0.38.0-beta.11",
 	"os": ["win32"],
 	"cpu": ["x64"],
 	"main": "lancedb.win32-x64-msvc.node",
@@ -11,7 +11,7 @@
    "ann"
  ],
  "private": false,
-  "version": "0.39.0-beta.0",
+  "version": "0.38.0-beta.11",
  "main": "dist/index.js",
  "exports": {
    ".": "./dist/index.js",
@@ -67,7 +67,7 @@
    "timeout": "3m"
  },
  "engines": {
-    "node": ">= 22"
+    "node": ">= 18"
  },
  "packageManager": "pnpm@11.1.1",
  "cpu": ["x64", "arm64"],
@@ -101,7 +101,7 @@
    "openai": "4.29.2"
  },
  "peerDependencies": {
-    "@types/node": ">=22",
+    "@types/node": ">=18",
    "apache-arrow": ">=15.0.0 <=18.1.0"
  },
  "peerDependenciesMeta": {
@@ -664,11 +664,7 @@ impl JsFullTextQuery {
 }

 fn parse_fts_query(query: Object) -> napi::Result<FullTextSearchQuery> {
-    // `&JsFullTextQuery` recovers a native class reference through napi's borrow-tracked
-    // path, which is only usable from generated `#[napi]` argument conversion. This is a
-    // manual lookup on a nested `Object` property instead, so use `ClassInstance`, which
-    // unwraps the class without requiring a borrow scope.
-    if let Ok(Some(query)) = query.get::<ClassInstance<JsFullTextQuery>>("query") {
+    if let Ok(Some(query)) = query.get::<&JsFullTextQuery>("query") {
        Ok(FullTextSearchQuery::new_query(query.inner.clone()))
    } else if let Ok(Some(query_text)) = query.get::<String>("query") {
        let mut query_text = query_text;
@@ -1,6 +1,6 @@
 [package]
 name = "lancedb-python"
-version = "0.39.0-beta.0"
+version = "0.38.0-beta.11"
 publish = false
 edition.workspace = true
 description = "Python bindings for LanceDB"
@@ -6,7 +6,7 @@ import importlib.metadata
 import os
 from concurrent.futures import ThreadPoolExecutor
 from datetime import timedelta
-from typing import Dict, Optional, Union, Any, List, Iterable, TYPE_CHECKING
+from typing import Dict, Optional, Union, Any, List, Iterable

 __version__ = importlib.metadata.version("lancedb")

@@ -20,7 +20,7 @@ from .db import AsyncConnection, DBConnection, LanceDBConnection
 from .remote import ClientConfig
 from .remote.db import RemoteDBConnection
 from .expr import Expr, col, lit, func
-from .schema import blob, vector
+from .schema import blob, vector, BlobType
 from .job import AsyncJob, Job
 from .functions import (
    FunctionArtifactRequest as FunctionArtifactRequest,
@@ -49,19 +49,6 @@ from .namespace import (
 )


-if TYPE_CHECKING:
-    from lance.blob import BlobType as BlobType
-
-
-def __getattr__(name: str):
-    if name == "BlobType":
-        from .schema import BlobType
-
-        globals()["BlobType"] = BlobType
-        return BlobType
-    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
-
-
 def _check_s3_bucket_with_dots(
    uri: str, storage_options: Optional[Dict[str, str]]
 ) -> None:
@@ -12,7 +12,7 @@ from typing import TYPE_CHECKING, Optional, Union
 import pyarrow as pa

 from .expr import Expr
-from .schema import row_addressable_blob_v2_paths
+from .schema import blob_v2_column_paths
 from .types import BlobMode, QueryProjection, QueryProjectionSpec

 if TYPE_CHECKING:
@@ -119,7 +119,7 @@ def blob_v2_projection_sources(
    schema: pa.Schema,
    projection: QueryProjection,
 ) -> dict[str, str]:
-    blob_columns = row_addressable_blob_v2_paths(schema)
+    blob_columns = blob_v2_column_paths(schema)
    if not blob_columns:
        return {}
    columns = set(blob_columns)
@@ -140,9 +140,7 @@ def v2_projection_needs_row_id(
 ) -> bool:
    if with_row_id:
        return False
-    return projection_includes_blob_column(
-        projection, row_addressable_blob_v2_paths(schema)
-    )
+    return projection_includes_blob_column(projection, blob_v2_column_paths(schema))


 def blob_auto_row_id_for_scan(
@@ -150,7 +150,6 @@ class Connection(object):
    def job(self, job_id: str) -> Job: ...
    async def create_function_async(self, request_json: str) -> Job: ...
    async def get_function(self, name: str, version: str) -> str: ...
-    async def drop_function(self, name: str, version: str) -> bool: ...
    async def list_jobs(self) -> List[JobInfo]: ...
    async def get_job(self, job_id: str) -> Optional[JobDescription]: ...
    async def cancel_job(self, job_id: str) -> bool: ...
@@ -607,7 +606,6 @@ class FullTextQuery:
 class PyQueryRequest:
    limit: Optional[int]
    offset: Optional[int]
-    take_offsets: Optional[List[int]]
    filter: Optional[Union[str, bytes]]
    full_text_search: Optional[FullTextQuery]
    select: Optional[Union[str, List[str]]]
@@ -712,16 +712,6 @@ class DBConnection(EnforceOverrides):
            "Function catalog operations are not supported for this connection type"
        )

-    def drop_function(self, name: str, *, version: str) -> bool:
-        """Drop one exact immutable Function version from the remote catalog.
-
-        Returns True when the version changed to Dropped and False for an
-        idempotent replay. Local connections raise NotImplementedError.
-        """
-        raise NotImplementedError(
-            "Function catalog operations are not supported for this connection type"
-        )
-
    def job(self, job_id: str) -> Job:
        """A [Job][lancedb.job.Job] handle for a server-side job by id.

@@ -1423,10 +1413,6 @@ class LanceDBConnection(DBConnection):
    def get_function(self, name: str, *, version: str) -> FunctionVersion:
        return LOOP.run(self._conn.get_function(name, version=version))

-    @override
-    def drop_function(self, name: str, *, version: str) -> bool:
-        return LOOP.run(self._conn.drop_function(name, version=version))
-
    @override
    def list_jobs(self) -> List[JobInfo]:
        """List server-side jobs across the database's tables."""
@@ -2257,10 +2243,6 @@ class AsyncConnection(object):
        """Open one exact immutable Function version from the remote catalog."""
        return FunctionVersion.from_json(await self._inner.get_function(name, version))

-    async def drop_function(self, name: str, *, version: str) -> bool:
-        """Drop one exact immutable Function version from the remote catalog."""
-        return await self._inner.drop_function(name, version)
-
    async def list_jobs(self) -> List[JobInfo]:
        """List server-side jobs across the database's tables."""
        return await self._inner.list_jobs()
@@ -49,25 +49,11 @@ from pydantic import (
    model_validator,
 )

-from .schema import is_blob_v2_field as _is_blob_v2_field
-
 _Int32 = conint(strict=True, ge=-(2**31), le=2**31 - 1)
 _UInt32 = conint(strict=True, ge=0, le=2**32 - 1)
 _UInt64 = conint(strict=True, ge=0, le=2**64 - 1)


-def _validate_gpu_wire_marker(value: Any) -> bool:
-    if value is not True:
-        raise ValueError("runtime.gpu must be true")
-    return True
-
-
-def _normalize_gpu_marker(value: bool) -> Optional[bool]:
-    if not isinstance(value, bool):
-        raise ValueError("gpu must be a boolean")
-    return True if value else None
-
-
 class _FrozenDict(dict):
    def _immutable(self, *args, **kwargs):
        raise TypeError("remote canonical values are immutable")
@@ -253,23 +239,6 @@ class PythonRuntimeSpec(_RemoteValue):
    python_version: Optional[str] = None
    environment: Optional[PythonEnvironmentSpec] = None
    env: Optional[Mapping[str, str]] = None
-    gpu: Optional[bool] = None
-
-    @model_validator(mode="before")
-    @classmethod
-    def _discard_unknown_runtime_payload(cls, value):
-        if isinstance(value, Mapping):
-            kind = value.get("kind")
-            if isinstance(kind, str) and kind not in {"python", "python_v2"}:
-                return {"kind": kind}
-        return value
-
-    @field_validator("gpu", mode="before")
-    @classmethod
-    def _validate_gpu_marker(cls, value):
-        if value is None:
-            return None
-        return _validate_gpu_wire_marker(value)

    @model_validator(mode="after")
    def _validate_runtime_kind(self):
@@ -278,28 +247,18 @@ class PythonRuntimeSpec(_RemoteValue):
                raise ValueError("python runtime requires python_version")
            if self.environment is None:
                raise ValueError("python runtime requires environment")
-            if self.gpu is not None:
-                raise ValueError("python runtime with gpu requires kind='python_v2'")
-        elif self.kind == "python_v2":
-            if self.python_version is None:
-                raise ValueError("python_v2 runtime requires python_version")
-            if self.environment is None:
-                raise ValueError("python_v2 runtime requires environment")
-            if self.gpu is None:
-                raise ValueError("python_v2 runtime requires gpu")
        else:
            object.__setattr__(self, "python_version", None)
            object.__setattr__(self, "environment", None)
            object.__setattr__(self, "env", None)
-            object.__setattr__(self, "gpu", None)
        return self


 class FunctionVersion(_RemoteValue):
    """An exact immutable Function version returned by Enterprise.

-    The GPU execution requirement is part of this identity. CPU and memory sizing,
-    priority, concurrency, and retry policy belong to the execution platform.
+    Scheduling resources, priority, concurrency, and retry policy belong to
+    the submitting Job and are not part of this identity.
    """

    name: str
@@ -520,7 +479,6 @@ class RefreshColumnResult(_RemoteValue):


 _FUNCTION_NAME = re.compile(r"^[A-Za-z_][A-Za-z0-9_.-]*$")
-_FUNCTION_BLOB_V2_TYPE = "blob_v2"


 _GRAMMAR_PRIMITIVES = (
@@ -537,7 +495,6 @@ _GRAMMAR_PRIMITIVES = (
    (pa.float32(), "float32"),
    (pa.float64(), "float64"),
    (pa.string(), "utf8"),
-    (pa.large_string(), "large_utf8"),
    (pa.binary(), "binary"),
    (pa.date32(), "date32"),
    (pa.date64(), "date64"),
@@ -545,177 +502,31 @@ _GRAMMAR_PRIMITIVES = (


 def _canonical_arrow_type(data_type: pa.DataType) -> str:
-    """The compact Function grammar, or canonical exact JSON for nested types."""
-    grammar = _grammar_arrow_type(data_type)
-    if grammar is not None:
-        return grammar
-    exact = _exact_arrow_type(data_type)
-    return json.dumps(exact, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
-
-
-def _grammar_arrow_type(data_type: pa.DataType) -> Optional[str]:
+    """The server's V1 Function type grammar. Anything outside it is rejected
+    here rather than at registration."""
    for candidate, name in _GRAMMAR_PRIMITIVES:
        if data_type == candidate:
            return name
    if pa.types.is_list(data_type) or pa.types.is_large_list(data_type):
-        item = _grammar_list_item(data_type)
-        if item is None:
-            return None
        prefix = "list" if pa.types.is_list(data_type) else "large_list"
-        return f"{prefix}<{item}>"
+        return f"{prefix}<{_canonical_list_item(data_type)}>"
    if pa.types.is_fixed_size_list(data_type) and data_type.list_size > 0:
-        item = _grammar_list_item(data_type)
-        if item is not None:
-            return f"fixed_size_list<{item}, {data_type.list_size}>"
-    return None
+        return (
+            f"fixed_size_list<{_canonical_list_item(data_type)}, {data_type.list_size}>"
+        )
+    raise TypeError(f"unsupported Arrow type for Function signature: {data_type}")


-def _grammar_list_item(data_type: pa.DataType) -> Optional[str]:
+def _canonical_list_item(data_type: pa.DataType) -> str:
    """The grammar names only the item type; it always means a non-nullable
-    child called `item`, so other child properties require exact JSON."""
+    child called `item`, so any other child metadata cannot be represented."""
    child = data_type.value_field
    if child.name != "item" or child.nullable or child.metadata:
-        return None
-    return _grammar_arrow_type(child.type)
-
-
-def _validate_exact_arrow_field(field: pa.Field) -> None:
-    if not field.name:
        raise TypeError(
-            "unsupported Arrow type for Function signature: field names "
-            "must not be empty"
+            "unsupported Arrow type for Function signature: list items must be a "
+            f"non-nullable field named 'item', got {child}"
        )
-    if _is_blob_v2_field(field):
-        if not _has_supported_blob_v2_layout(field):
-            raise TypeError(
-                "unsupported Arrow type for Function signature: lance.blob.v2 "
-                f"requires a supported Blob storage layout, got {field}"
-            )
-    elif field.metadata:
-        raise TypeError(
-            "unsupported Arrow type for Function signature: field metadata "
-            f"is not supported, got {field}"
-        )
-
-
-def _has_supported_blob_v2_layout(field: pa.Field) -> bool:
-    data_type = field.type
-    if isinstance(data_type, pa.ExtensionType):
-        data_type = data_type.storage_type
-    if not pa.types.is_struct(data_type):
-        return False
-
-    fields = tuple(data_type)
-
-    def matches(spec, compare_nullable) -> bool:
-        return len(fields) == len(spec) and all(
-            actual.name == name
-            and actual.type == expected_type
-            and (not check_nullable or actual.nullable == nullable)
-            for actual, (name, expected_type, nullable), check_nullable in zip(
-                fields, spec, compare_nullable
-            )
-        )
-
-    logical_minimal = (
-        ("data", pa.large_binary(), True),
-        ("uri", pa.utf8(), True),
-    )
-    logical_full = logical_minimal + (
-        ("position", pa.uint64(), True),
-        ("size", pa.uint64(), True),
-    )
-    prepared = (
-        ("kind", pa.uint8(), True),
-        ("data", pa.large_binary(), True),
-        ("uri", pa.utf8(), True),
-        ("blob_id", pa.uint32(), True),
-        ("blob_size", pa.uint64(), True),
-        ("position", pa.uint64(), True),
-    )
-    descriptor = (
-        ("kind", pa.uint8(), False),
-        ("position", pa.uint64(), False),
-        ("size", pa.uint64(), False),
-        ("blob_id", pa.uint32(), False),
-        ("blob_uri", pa.utf8(), False),
-    )
-    return (
-        matches(logical_minimal, (True, True))
-        or matches(logical_full, (True, True, False, False))
-        or matches(prepared, (True,) * len(prepared))
-        or matches(descriptor, (False,) * len(descriptor))
-    )
-
-
-def _canonical_arrow_field(field: pa.Field) -> str:
-    _validate_exact_arrow_field(field)
-    if _is_blob_v2_field(field):
-        return _FUNCTION_BLOB_V2_TYPE
-    return _canonical_arrow_type(field.type)
-
-
-def _exact_arrow_field(field: pa.Field) -> dict[str, Any]:
-    _validate_exact_arrow_field(field)
-    if _is_blob_v2_field(field):
-        raise TypeError(
-            "unsupported Arrow type for Function signature: nested Blob v2 "
-            "fields are not supported; declare Blob parameters or named result "
-            "fields directly"
-        )
-    value = {
-        "name": field.name,
-        "nullable": field.nullable,
-        "type": _exact_arrow_type(field.type),
-    }
-    return value
-
-
-def _exact_arrow_type(data_type: pa.DataType) -> dict[str, Any]:
-    for candidate, name in _GRAMMAR_PRIMITIVES:
-        if data_type == candidate:
-            return {"type": name}
-    if pa.types.is_struct(data_type):
-        fields = list(data_type)
-        names = [field.name for field in fields]
-        if not fields or len(set(names)) != len(names):
-            raise TypeError(
-                "unsupported Arrow type for Function signature: structs must have "
-                "non-empty, uniquely named fields"
-            )
-        return {
-            "type": "struct",
-            "fields": [_exact_arrow_field(field) for field in fields],
-        }
-    if (
-        pa.types.is_list(data_type)
-        or pa.types.is_large_list(data_type)
-        or pa.types.is_fixed_size_list(data_type)
-    ):
-        if pa.types.is_fixed_size_list(data_type):
-            if data_type.value_field.name != "item":
-                raise TypeError(
-                    "unsupported Arrow type for Function signature: fixed-size list "
-                    "items must be named 'item'"
-                )
-            if data_type.list_size <= 0:
-                raise TypeError(
-                    f"unsupported Arrow type for Function signature: {data_type}"
-                )
-        value: dict[str, Any] = {
-            "type": (
-                "list"
-                if pa.types.is_list(data_type)
-                else "large_list"
-                if pa.types.is_large_list(data_type)
-                else "fixed_size_list"
-            ),
-            "fields": [_exact_arrow_field(data_type.value_field)],
-        }
-        if pa.types.is_fixed_size_list(data_type):
-            value["length"] = data_type.list_size
-        return value
-    raise TypeError(f"unsupported Arrow type for Function signature: {data_type}")
+    return _canonical_arrow_type(child.type)


 def _list_of(item: pa.DataType) -> pa.DataType:
@@ -789,15 +600,8 @@ def _callable_parameters(function: Callable[..., Any]) -> tuple[inspect.Paramete

 def _function_output(output: pa.DataType | pa.Field | pa.Schema) -> FunctionOutput:
    if isinstance(output, pa.Schema):
-        if output.metadata:
-            raise TypeError("Function output schema metadata is not supported")
        fields = tuple(output)
-    elif (
-        isinstance(output, pa.Field)
-        and not _is_blob_v2_field(output)
-        and pa.types.is_struct(output.type)
-    ):
-        _validate_exact_arrow_field(output)
+    elif isinstance(output, pa.Field) and pa.types.is_struct(output.type):
        if output.nullable:
            raise ValueError("Function output must be non-nullable")
        fields = tuple(output.type)
@@ -813,12 +617,11 @@ def _function_output(output: pa.DataType | pa.Field | pa.Schema) -> FunctionOutp
            raise TypeError(
                "output_schema must be a PyArrow DataType, Field, or Schema"
            )
-        _validate_exact_arrow_field(field)
        if field.nullable:
            raise ValueError("Function output must be non-nullable")
        return FunctionOutput(
            kind="scalar",
-            arrow_type=_canonical_arrow_field(field),
+            arrow_type=_canonical_arrow_type(field.type),
            nullable=False,
        )

@@ -826,8 +629,6 @@ def _function_output(output: pa.DataType | pa.Field | pa.Schema) -> FunctionOutp
        raise ValueError("named-struct Function output must contain at least one field")
    if any(field.nullable for field in fields):
        raise ValueError("Function output fields must be non-nullable")
-    for field in fields:
-        _validate_exact_arrow_field(field)
    names = [field.name for field in fields]
    if len(set(names)) != len(names):
        raise ValueError("Function output field names must be unique")
@@ -836,7 +637,7 @@ def _function_output(output: pa.DataType | pa.Field | pa.Schema) -> FunctionOutp
        fields=tuple(
            FunctionResultField(
                name=field.name,
-                arrow_type=_canonical_arrow_field(field),
+                arrow_type=_canonical_arrow_type(field.type),
                nullable=False,
            )
            for field in fields
@@ -856,10 +657,6 @@ def _infer_signature(
    if input_schema is not None:
        if not isinstance(input_schema, pa.Schema):
            raise TypeError("input_schema must be a PyArrow Schema")
-        if input_schema.metadata:
-            raise TypeError("Function input schema metadata is not supported")
-        for field in input_schema:
-            _validate_exact_arrow_field(field)
        expected = tuple(parameter.name for parameter in parameters)
        actual = tuple(input_schema.names)
        if actual != expected:
@@ -870,7 +667,7 @@ def _infer_signature(
        inputs = tuple(
            FunctionParameter(
                name=field.name,
-                arrow_type=_canonical_arrow_field(field),
+                arrow_type=_canonical_arrow_type(field.type),
                nullable=field.nullable,
            )
            for field in input_schema
@@ -893,9 +690,7 @@ def _infer_signature(
        inputs.append(
            FunctionParameter(
                name=parameter.name,
-                arrow_type=_canonical_arrow_field(
-                    pa.field(parameter.name, data_type, nullable=nullable)
-                ),
+                arrow_type=_canonical_arrow_type(data_type),
                nullable=nullable,
            )
        )
@@ -1115,7 +910,6 @@ class UdfDefinition:
        pip: tuple[str, ...],
        env: Mapping[str, str],
        python_version: Optional[str],
-        gpu: bool = False,
        conda: tuple[str, ...] = (),
        conda_channels: tuple[str, ...] = (),
    ):
@@ -1144,14 +938,12 @@ class UdfDefinition:
        signature = _infer_signature(function, input_schema, output_schema)
        source = _package_source(function)
        digest = f"sha256:{hashlib.sha256(source).hexdigest()}"
-        gpu_marker = _normalize_gpu_marker(gpu)
        runtime = PythonRuntimeSpec(
-            kind="python_v2" if gpu_marker is not None else "python",
+            kind="python",
            python_version=python_version
            or f"{sys.version_info.major}.{sys.version_info.minor}",
            environment=environment_spec,
            env=environment,
-            gpu=gpu_marker,
        )
        self._function = function
        self._request = FunctionRegistrationRequest(
@@ -1197,7 +989,6 @@ def udf(
    pip: tuple[str, ...] | list[str] = (),
    env: Optional[Mapping[str, str]] = None,
    python_version: Optional[str] = None,
-    gpu: bool = False,
    conda: tuple[str, ...] | list[str] = (),
    conda_channels: tuple[str, ...] | list[str] = (),
 ) -> Callable[[Callable[..., Any]], UdfDefinition]: ...
@@ -1212,7 +1003,6 @@ def udf(
    pip: tuple[str, ...] | list[str] = (),
    env: Optional[Mapping[str, str]] = None,
    python_version: Optional[str] = None,
-    gpu: bool = False,
    conda: tuple[str, ...] | list[str] = (),
    conda_channels: tuple[str, ...] | list[str] = (),
 ):
@@ -1245,10 +1035,6 @@ def udf(
        Environment variables included in the Function definition.
    python_version : str, optional
        Remote Python major/minor version. Defaults to the client version.
-    gpu : bool, default False
-        Whether every remote execution requires a GPU. The execution platform
-        selects one compatible GPU for each worker. The requirement is part of
-        the immutable Function version.

    The packaged artifact is a snapshot: the function source plus exactly
    the module-level names it references (modules as imports, importable
@@ -1273,11 +1059,6 @@ def udf(
    ...     return value * 2
    >>> score(1.5)
    3.0
-    >>> @udf(pip=["cupy-cuda12x"], gpu=True)
-    ... def gpu_score(value: int) -> int:
-    ...     return value * 2
-    >>> gpu_score.registration_request.runtime.gpu
-    True
    """

    def decorate(target: Callable[..., Any]) -> UdfDefinition:
@@ -1289,7 +1070,6 @@ def udf(
            pip=tuple(pip),
            env={} if env is None else env,
            python_version=python_version,
-            gpu=gpu,
            conda=tuple(conda),
            conda_channels=tuple(conda_channels),
        )
@@ -42,8 +42,6 @@ class MaterializedViewDefinition:
    """Cap on the number of rows the view holds."""
    inputs: List[str] = field(default_factory=list)
    """Source columns the projections and filter read."""
-    source_namespace: List[str] = field(default_factory=list)
-    """Namespace holding the source table; empty is the root namespace."""


 def _definition_from_schema(
@@ -55,8 +53,7 @@ def _definition_from_schema(
        raise ValueError(f"Table '{name}' is not a materialized view")
    value = json.loads(raw)
    kind = value.get("kind")
-    # "namespaced_select" keeps older readers from resolving the source at root.
-    if kind not in ("select", "namespaced_select"):
+    if kind != "select":
        raise NotImplementedError(
            f"materialized view '{name}' is defined by '{kind}', which this "
            "version of lancedb cannot refresh"
@@ -69,7 +66,6 @@ def _definition_from_schema(
        filter=value.get("filter"),
        limit=value.get("limit"),
        inputs=value.get("inputs", []),
-        source_namespace=value.get("source_namespace", []),
    )


@@ -109,7 +109,6 @@ def _query_is_plain_scan(query: Query) -> bool:
    return (
        query.vector is None
        and query.full_text_query is None
-        and query.take_offsets is None
        and not query.postfilter
        and not query.order_by
    )
@@ -805,10 +804,6 @@ class Query(pydantic.BaseModel):
    # offset to start fetching results from
    offset: Optional[int] = None

-    # Dataset offsets whose duplicate occurrences must be restored after lookup.
-    # This is populated when a take query is converted to this serializable form.
-    take_offsets: Optional[List[int]] = None
-
    # if true, will only search the indexed data
    fast_search: Optional[bool] = None

@@ -830,7 +825,6 @@ class Query(pydantic.BaseModel):
        query = cls()
        query.limit = req.limit
        query.offset = req.offset
-        query.take_offsets = req.take_offsets
        query.filter = req.filter
        query.full_text_query = req.full_text_search
        query.columns = req.select
@@ -749,10 +749,6 @@ class RemoteDBConnection(DBConnection):
    def get_function(self, name: str, *, version: str) -> FunctionVersion:
        return LOOP.run(self._conn.get_function(name, version=version))

-    @override
-    def drop_function(self, name: str, *, version: str) -> bool:
-        return LOOP.run(self._conn.drop_function(name, version=version))
-
    @override
    def list_jobs(self) -> List["JobInfo"]:
        """List server-side jobs across the database's tables."""
@@ -67,15 +67,7 @@ from ..query import (
    LanceTakeQueryBuilder,
    LanceVectorQueryBuilder,
 )
-from ..table import (
-    AsyncTable,
-    BlobMode,
-    Branches,
-    IndexStatistics,
-    Query,
-    Table,
-    Tags,
-)
+from ..table import AsyncTable, BlobMode, Branches, IndexStatistics, Query, Table, Tags
 from ..types import BaseTokenizerType


@@ -4,34 +4,30 @@

 """Schema helpers for Lance blob columns."""

-import importlib
-from typing import TYPE_CHECKING
-
 import pyarrow as pa
-import pyarrow.ipc
-
-if TYPE_CHECKING:
-    from lance.blob import BlobType as BlobType

 _BLOB_EXTENSION_NAME = "lance.blob.v2"
 _BLOB_V1_KEY = "lance-encoding:blob"
 _ARROW_EXT_NAME_KEY = "ARROW:extension:name"
-_BLOB_V2_STORAGE_TYPE = pa.struct(
-    [
-        pa.field("data", pa.large_binary(), nullable=True),
-        pa.field("uri", pa.utf8(), nullable=True),
-        pa.field("position", pa.uint64(), nullable=True),
-        pa.field("size", pa.uint64(), nullable=True),
-    ]
-)
-_resolved_blob_type = None


-class _FallbackBlobType(pa.ExtensionType):
-    """lance.blob.v2 extension type used when pylance is not installed."""
+class BlobType(pa.ExtensionType):
+    """PyArrow extension type for a Lance blob v2 column.
+
+    Queries return descriptors; call :meth:`~lancedb.table.Table.fetch_blob_files`
+    for lazy reads or :meth:`~lancedb.table.Table.fetch_blobs` for eager bytes.
+    """

    def __init__(self) -> None:
-        pa.ExtensionType.__init__(self, _BLOB_V2_STORAGE_TYPE, _BLOB_EXTENSION_NAME)
+        storage_type = pa.struct(
+            [
+                pa.field("data", pa.large_binary(), nullable=True),
+                pa.field("uri", pa.utf8(), nullable=True),
+                pa.field("position", pa.uint64(), nullable=True),
+                pa.field("size", pa.uint64(), nullable=True),
+            ]
+        )
+        super().__init__(storage_type, _BLOB_EXTENSION_NAME)

    def __arrow_ext_serialize__(self) -> bytes:
        return b""
@@ -39,16 +35,23 @@ class _FallbackBlobType(pa.ExtensionType):
    @classmethod
    def __arrow_ext_deserialize__(
        cls, storage_type: pa.DataType, serialized: bytes
-    ) -> "_FallbackBlobType":
+    ) -> "BlobType":
        return cls()

    def __reduce__(self):
+        # Ensure pickle round-trips on older pyarrow (apache/arrow#35599).
        return type(self).__arrow_ext_deserialize__, (
            self.storage_type,
            self.__arrow_ext_serialize__(),
        )


+try:
+    pa.register_extension_type(BlobType())  # type: ignore[arg-type]
+except pa.ArrowKeyError:
+    pass
+
+
 def _metadata_value(metadata: dict, key: str):
    return metadata.get(key.encode()) or metadata.get(key)

@@ -89,105 +92,43 @@ def is_blob_like_field(field: pa.Field) -> bool:
    return is_blob_v2_field(field) or _metadata_marks_legacy_blob(field.metadata or {})


-def _collect_blob_paths(schema: pa.Schema, is_blob) -> list[tuple[str, bool]]:
-    """Walk the schema and return (path, has_list_ancestor) for each blob field."""
-    paths: list[tuple[str, bool]] = []
+def _collect_blob_paths(schema: pa.Schema, is_blob) -> list[str]:
+    paths: list[str] = []

-    def walk(fields, prefix: str, has_list_ancestor: bool) -> None:
+    def walk(fields, prefix: str) -> None:
        for field in fields:
            path = f"{prefix}.{field.name}" if prefix else field.name
            if is_blob(field):
-                paths.append((path, has_list_ancestor))
+                paths.append(path)
            elif pa.types.is_struct(field.type):
-                walk(field.type, path, has_list_ancestor)
+                walk(field.type, path)
            elif (
                pa.types.is_list(field.type)
                or pa.types.is_large_list(field.type)
                or pa.types.is_fixed_size_list(field.type)
            ):
-                walk([field.type.value_field], path, True)
+                walk([field.type.value_field], path)

-    walk(schema, "", False)
+    walk(schema, "")
    return paths


 def blob_column_paths(schema: pa.Schema) -> list[str]:
    """Dotted paths of blob-like columns (v2 extension or legacy metadata)."""
-    return [path for path, _ in _collect_blob_paths(schema, is_blob_like_field)]
+    return _collect_blob_paths(schema, is_blob_like_field)


 def blob_v2_column_paths(schema: pa.Schema) -> list[str]:
-    return [path for path, _ in _collect_blob_paths(schema, is_blob_v2_field)]
-
-
-def row_addressable_blob_v2_paths(schema: pa.Schema) -> list[str]:
-    """Blob v2 paths with one blob addressable by table row id.
-
-    ``fetch_blobs`` and the descriptor row-id ride-along address one blob per
-    row, so a blob inside a list container has no row-id slot and no fetch
-    path. Those columns still store and query as raw descriptors.
-    """
-    return [
-        path
-        for path, has_list_ancestor in _collect_blob_paths(schema, is_blob_v2_field)
-        if not has_list_ancestor
-    ]
+    return _collect_blob_paths(schema, is_blob_v2_field)


 def schema_has_blob_field(schema: pa.Schema) -> bool:
    return bool(blob_column_paths(schema))


-def _deserialize_registered_type(extension_type: pa.ExtensionType) -> pa.DataType:
-    """Return the type Arrow reconstructs for this extension name."""
-    schema = pa.schema([pa.field("value", extension_type)])
-    restored = pa.ipc.read_schema(schema.serialize())
-    return restored.field("value").type
-
-
-def _resolve_blob_type():
-    """Return the BlobType class this process should use.
-
-    pylance's class when it owns the lance.blob.v2 registry entry,
-    otherwise LanceDB's fallback. A different registered class is an error.
-    """
-    global _resolved_blob_type
-    if _resolved_blob_type is not None:
-        return _resolved_blob_type
-    try:
-        blob_module = importlib.import_module("lance.blob")
-    except ModuleNotFoundError as err:
-        if err.name not in ("lance", "lance.blob"):
-            raise
-    else:
-        blob_type = getattr(blob_module, "BlobType", None)
-        if blob_type is not None:
-            registered_type = _deserialize_registered_type(blob_type())
-            if type(registered_type) is not blob_type:
-                registered_cls = type(registered_type)
-                raise ValueError(
-                    "lance.blob.v2 is already registered by "
-                    f"{registered_cls.__module__}.{registered_cls.__qualname__}"
-                )
-            _resolved_blob_type = blob_type
-            return blob_type
-    try:
-        pa.register_extension_type(_FallbackBlobType())  # type: ignore[arg-type]
-    except pa.ArrowKeyError as err:
-        raise ValueError(
-            "lance.blob.v2 is already registered by another extension class"
-        ) from err
-    _resolved_blob_type = _FallbackBlobType
-    return _resolved_blob_type
-
-
 def blob(name: str, nullable: bool = True) -> pa.Field:
-    """Create a Lance blob v2 column field.
-
-    When pylance is installed this is ``lance.blob.BlobType``.
-    """
-    blob_type = _resolve_blob_type()
-    return pa.field(name, blob_type(), nullable=nullable)
+    """Create a Lance blob v2 column field."""
+    return pa.field(name, BlobType(), nullable=nullable)


 def vector(dimension: int, value_type: pa.DataType = pa.float32()) -> pa.DataType:
@@ -214,11 +155,3 @@ def vector(dimension: int, value_type: pa.DataType = pa.float32()) -> pa.DataTyp
    ... ])
    """
    return pa.list_(value_type, dimension)
-
-
-def __getattr__(name: str):
-    if name == "BlobType":
-        blob_type = _resolve_blob_type()
-        globals()["BlobType"] = blob_type
-        return blob_type
-    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
@@ -104,12 +104,7 @@ from .util import (
    value_to_sql,
 )
 from .index import lang_mapping
-from .schema import (
-    blob_v2_column_paths,
-    is_blob_v2_field,
-    row_addressable_blob_v2_paths,
-    schema_has_blob_field,
-)
+from .schema import blob_v2_column_paths, schema_has_blob_field


 def _should_push_down_query_table(
@@ -431,7 +426,6 @@ def _cast_to_target_schema(

    def gen():
        for batch in reader:
-            batch = _coerce_blob_write_columns(batch, reordered_schema)
            # Table but not RecordBatch has cast.
            cast_batches = (
                pa.Table.from_batches([batch]).cast(reordered_schema).to_batches()
@@ -444,166 +438,6 @@ def _cast_to_target_schema(
    return pa.RecordBatchReader.from_batches(reordered_schema, gen())


-def _coerce_blob_write_columns(
-    batch: pa.RecordBatch, target_schema: pa.Schema
-) -> pa.RecordBatch:
-    """Materialize blob storage structs before the stream leaves Python.
-
-    merge_insert requires its source reader to already match the table's
-    physical schema. Unlike add and insert, it does not pass through
-    LanceDB's Rust blob coercion, so preserving binary input here would
-    reach Lance as binary and fail the schema check.
-    """
-    columns = []
-    fields = []
-    changed = False
-    for field, column in zip(batch.schema, batch.columns):
-        target_field = target_schema.field(field.name)
-        coerced = _coerce_blob_value(column, target_field)
-        if coerced is not column:
-            column = coerced
-            field = pa.field(
-                field.name,
-                coerced.type,
-                field.nullable,
-                target_field.metadata,
-            )
-            changed = True
-        columns.append(column)
-        fields.append(field)
-    if not changed:
-        return batch
-    return pa.RecordBatch.from_arrays(
-        columns, schema=pa.schema(fields, metadata=batch.schema.metadata)
-    )
-
-
-def _coerce_blob_value(column: pa.Array, target_field: pa.Field) -> pa.Array:
-    if is_blob_v2_field(target_field) and _can_coerce_to_blob(column.type):
-        return _coerce_value_to_blob(column, target_field)
-
-    target_type = target_field.type
-    if pa.types.is_struct(target_type) and pa.types.is_struct(column.type):
-        children = []
-        fields = []
-        changed = False
-        for source_field in column.type:
-            source_column = column.field(source_field.name)
-            nested_target = next(
-                (field for field in target_type if field.name == source_field.name),
-                None,
-            )
-            if nested_target is None:
-                children.append(source_column)
-                fields.append(source_field)
-                continue
-            coerced = _coerce_blob_value(source_column, nested_target)
-            if coerced is not source_column:
-                changed = True
-            child_array, child_type = _physical_array_and_type(coerced)
-            children.append(child_array)
-            fields.append(
-                pa.field(
-                    source_field.name,
-                    child_type,
-                    source_field.nullable,
-                    nested_target.metadata,
-                )
-            )
-        if not changed:
-            return column
-        return pa.StructArray.from_arrays(
-            children,
-            fields=fields,
-            mask=column.is_null() if column.null_count else None,
-        )
-
-    if _is_list_like(target_type) and _is_list_like(column.type):
-        return _coerce_blob_list_values(column, target_type.value_field)
-
-    return column
-
-
-def _coerce_blob_list_values(
-    column: pa.Array, target_value_field: pa.Field
-) -> pa.Array:
-    """Coerce blob values inside a list column, preserving offsets and nulls.
-
-    Works on the raw child values window instead of ``pc.list_flatten`` because
-    flatten drops values spanned by null slots, which would misalign offsets.
-    """
-    mask = column.is_null() if column.null_count else None
-    if pa.types.is_fixed_size_list(column.type):
-        list_size = column.type.list_size
-        values = column.values.slice(column.offset * list_size, len(column) * list_size)
-        coerced = _coerce_blob_value(values, target_value_field)
-        if coerced is values:
-            return column
-        physical_values, _ = _physical_array_and_type(coerced)
-        return pa.FixedSizeListArray.from_arrays(physical_values, list_size, mask=mask)
-    offsets = column.offsets
-    first_offset = offsets[0].as_py()
-    values = column.values.slice(
-        first_offset,
-        offsets[-1].as_py() - first_offset,
-    )
-    coerced = _coerce_blob_value(values, target_value_field)
-    if coerced is values:
-        return column
-    physical_values, _ = _physical_array_and_type(coerced)
-    if first_offset:
-        offsets = pc.subtract(offsets, pa.scalar(first_offset, offsets.type))
-    if pa.types.is_large_list(column.type):
-        return pa.LargeListArray.from_arrays(offsets, physical_values, mask=mask)
-    return pa.ListArray.from_arrays(offsets, physical_values, mask=mask)
-
-
-def _coerce_value_to_blob(values: pa.Array, target_field: pa.Field) -> pa.Array:
-    if pa.types.is_null(values.type):
-        data = pa.nulls(len(values), type=pa.large_binary())
-    elif pa.types.is_large_binary(values.type):
-        data = values
-    else:
-        data = values.cast(pa.large_binary())
-    length = len(values)
-    storage_type = target_field.type
-    if isinstance(storage_type, pa.ExtensionType):
-        storage_type = storage_type.storage_type
-    storage_fields = list(storage_type)
-    children = []
-    for storage_field in storage_fields:
-        if storage_field.name == "data":
-            children.append(data)
-        else:
-            children.append(pa.nulls(length, type=storage_field.type))
-    storage = pa.StructArray.from_arrays(
-        children,
-        fields=storage_fields,
-        mask=values.is_null() if values.null_count else None,
-    )
-    if isinstance(target_field.type, pa.ExtensionType):
-        return pa.ExtensionArray.from_storage(target_field.type, storage)
-    return storage
-
-
-def _physical_array_and_type(array: pa.Array) -> tuple[pa.Array, pa.DataType]:
-    if isinstance(array.type, pa.ExtensionType):
-        return array.storage, array.type.storage_type
-    return array, array.type
-
-
-def _can_coerce_to_blob(data_type: pa.DataType) -> bool:
-    return _is_binary_like(data_type) or pa.types.is_null(data_type)
-
-
-def _is_binary_like(data_type: pa.DataType) -> bool:
-    return (
-        pa.types.is_binary(data_type)
-        or pa.types.is_large_binary(data_type)
-        or pa.types.is_binary_view(data_type)
-    )
-
-
 def _field_extension_name(field: pa.Field) -> Optional[str]:
    extension_name = getattr(field.type, "extension_name", None)
    if extension_name is not None:
@@ -630,73 +464,65 @@ def _align_field_types(
        target_field = next((f for f in target_fields if f.name == field.name), None)
        if target_field is None:
            raise ValueError(f"Field '{field.name}' not found in target schema")
-        new_fields.append(_align_field(field, target_field))
+        # Preserve arrow.json input until it reaches Lance. LanceDB exposes stored
+        # JSON columns as lance.json (JSONB-backed LargeBinary), but casting the
+        # input to that storage type here merely relabels the raw JSON bytes as
+        # JSONB. Lance must see arrow.json so it can perform the JSONB encoding.
+        if (
+            _field_extension_name(field) == "arrow.json"
+            and _field_extension_name(target_field) == "lance.json"
+        ):
+            new_fields.append(field)
+            continue
+        if pa.types.is_struct(target_field.type):
+            if pa.types.is_struct(field.type):
+                new_type = pa.struct(
+                    _align_field_types(
+                        field.type.fields,
+                        target_field.type.fields,
+                    )
+                )
+            else:
+                new_type = target_field.type
+        elif pa.types.is_list(target_field.type):
+            if _is_list_like(field.type):
+                new_type = pa.list_(
+                    _align_field_types(
+                        [field.type.value_field],
+                        [target_field.type.value_field],
+                    )[0]
+                )
+            else:
+                new_type = target_field.type
+        elif pa.types.is_large_list(target_field.type):
+            if _is_list_like(field.type):
+                new_type = pa.large_list(
+                    _align_field_types(
+                        [field.type.value_field],
+                        [target_field.type.value_field],
+                    )[0]
+                )
+            else:
+                new_type = target_field.type
+        elif pa.types.is_fixed_size_list(target_field.type):
+            if _is_list_like(field.type):
+                new_type = pa.list_(
+                    _align_field_types(
+                        [field.type.value_field],
+                        [target_field.type.value_field],
+                    )[0],
+                    target_field.type.list_size,
+                )
+            else:
+                new_type = target_field.type
+        else:
+            new_type = target_field.type
+        new_fields.append(
+            pa.field(field.name, new_type, field.nullable, target_field.metadata)
+        )
    return new_fields


-def _align_list_value_field(
-    value_field: pa.Field, target_value_field: pa.Field
-) -> pa.Field:
-    # A list has exactly one child, so the inferred child name ("item") aligns
-    # positionally and adopts the table's child name; pa.Table.cast renames it.
-    return _align_field(value_field, target_value_field).with_name(
-        target_value_field.name
-    )
-
-
-def _align_field(field: pa.Field, target_field: pa.Field) -> pa.Field:
-    # Preserve arrow.json input until it reaches Lance. LanceDB exposes stored
-    # JSON columns as lance.json (JSONB-backed LargeBinary), but casting the
-    # input to that storage type here merely relabels the raw JSON bytes as
-    # JSONB. Lance must see arrow.json so it can perform the JSONB encoding.
-    if (
-        _field_extension_name(field) == "arrow.json"
-        and _field_extension_name(target_field) == "lance.json"
-    ):
-        return field
-    if pa.types.is_struct(target_field.type):
-        if pa.types.is_struct(field.type):
-            new_type = pa.struct(
-                _align_field_types(
-                    field.type.fields,
-                    target_field.type.fields,
-                )
-            )
-        else:
-            new_type = target_field.type
-    elif pa.types.is_list(target_field.type):
-        if _is_list_like(field.type):
-            new_type = pa.list_(
-                _align_list_value_field(
-                    field.type.value_field, target_field.type.value_field
-                )
-            )
-        else:
-            new_type = target_field.type
-    elif pa.types.is_large_list(target_field.type):
-        if _is_list_like(field.type):
-            new_type = pa.large_list(
-                _align_list_value_field(
-                    field.type.value_field, target_field.type.value_field
-                )
-            )
-        else:
-            new_type = target_field.type
-    elif pa.types.is_fixed_size_list(target_field.type):
-        if _is_list_like(field.type):
-            new_type = pa.list_(
-                _align_list_value_field(
-                    field.type.value_field, target_field.type.value_field
-                ),
-                target_field.type.list_size,
-            )
-        else:
-            new_type = target_field.type
-    else:
-        new_type = target_field.type
-    return pa.field(field.name, new_type, field.nullable, target_field.metadata)
-
-
 def _infer_subschema(
    schema: List[pa.Field],
    reference_fields: List[pa.Field],
@@ -763,7 +589,7 @@ def sanitize_create_table(
        schema = data.schema
    else:
        if schema is not None:
-            data = pa.Table.from_batches([], schema=schema)
+            data = pa.Table.from_pylist([], schema)
    if schema is None:
        if data is None:
            raise ValueError("Either data or schema must be provided")
@@ -1678,9 +1504,9 @@ class Table(ABC):
        Offsets are mostly useful for sampling as the set of all valid offsets is easily
        known in advance to be [0, len(table)).

-        No guarantees are made regarding the order in which results are returned.
-        Repeated offsets produce repeated rows, which makes this method suitable for
-        sampling with replacement.
+        No guarantees are made regarding the order in which results are returned.  If
+        you desire an output order that matches the order of the given offsets, you will
+        need to add the row offset column to the output and align it yourself.

        Parameters
        ----------
@@ -2165,11 +1991,9 @@ class Table(ABC):
            Function columns are supported only on LanceDB Cloud and
            Enterprise.
        computed: Dict[str, str], optional
-            A mapping from output column names to SQL expressions derives each
-            output field from its expression. A direct projection of a Blob v2
-            field inherits Blob v2 semantics; other expressions derive their
-            ordinary Arrow type. Mapping order is declaration and dependency
-            order.
+            A map of column name to a SQL expression defining the column. The
+            column's type and inputs are derived from the expression, so no
+            data type is supplied.

            Unlike ``transforms``, the expression is stored rather than
            evaluated now: the column is committed with no values, and rows get
@@ -2874,7 +2698,7 @@ class LanceTable(Table):
            arrow_tbl = self.to_arrow()
            if blob_mode == "descriptions":
                arrow_tbl = strip_auto_row_ids(
-                    arrow_tbl, row_addressable_blob_v2_paths(self.schema)
+                    arrow_tbl, blob_v2_column_paths(self.schema)
                )
            return arrow_tbl.to_pandas(**kwargs)

@@ -4090,7 +3914,6 @@ class LanceTable(Table):
            )
            and not self._route_pushdown_to_rust
            and self.current_branch() is None
-            and query.take_offsets is None
        ):
            from lancedb.namespace import _execute_server_side_query

@@ -5279,9 +5102,7 @@ class AsyncTable:
        if blob_mode == "descriptions" or not schema_has_blob_field(schema):
            arrow_tbl = await self.to_arrow()
            if blob_mode == "descriptions":
-                arrow_tbl = strip_auto_row_ids(
-                    arrow_tbl, row_addressable_blob_v2_paths(schema)
-                )
+                arrow_tbl = strip_auto_row_ids(arrow_tbl, blob_v2_column_paths(schema))
            return arrow_tbl.to_pandas(**kwargs)

        if blob_mode == "lazy" and get_uri_scheme(await self.uri()) == "memory":
@@ -5984,23 +5805,7 @@ class AsyncTable:

    def _sync_query_to_async(
        self, query: Query
-    ) -> (
-        AsyncHybridQuery
-        | AsyncFTSQuery
-        | AsyncVectorQuery
-        | AsyncQuery
-        | AsyncTakeQuery
-    ):
-        if query.take_offsets is not None:
-            take_query = self.take_offsets(query.take_offsets)
-            if query.columns:
-                take_query = take_query.select(query.columns)
-            if query.use_lsm is not None:
-                take_query = take_query.use_lsm(query.use_lsm)
-            if query.with_row_id:
-                take_query = take_query.with_row_id()
-            return take_query
-
+    ) -> AsyncHybridQuery | AsyncFTSQuery | AsyncVectorQuery | AsyncQuery:
        async_query = self.query()
        if query.limit is not None:
            async_query = async_query.limit(query.limit)
@@ -6065,7 +5870,6 @@ class AsyncTable:
                self._namespace_client, self._pushdown_operations
            )
            and not self._route_pushdown_to_rust
-            and query.take_offsets is None
        ):
            from lancedb.namespace import _execute_server_side_query

@@ -6288,11 +6092,8 @@ class AsyncTable:
            Function columns are supported only on LanceDB Cloud and
            Enterprise.
        computed: Dict[str, str], optional
-            A mapping from output column names to SQL expressions derives each
-            output field from its expression. A direct projection of a Blob v2
-            field inherits Blob v2 semantics; other expressions derive their
-            ordinary Arrow type. Mapping order is declaration and dependency
-            order.
+            A map of column name to a SQL expression defining the column. The
+            column's type and inputs are derived from the expression.

            Unlike ``transforms``, the expression is stored rather than
            evaluated now: the column is committed with no values, and rows get
@@ -6563,9 +6364,6 @@ class AsyncTable:
        Offsets are mostly useful for sampling as the set of all valid offsets is easily
        known in advance to be [0, len(table)).

-        No guarantees are made regarding the order in which results are returned.
-        Repeated offsets produce repeated rows.
-
        Parameters
        ----------
        offsets: list[int]
@@ -2,15 +2,10 @@
 # SPDX-FileCopyrightText: Copyright The LanceDB Authors

 import io
-import subprocess
-import sys
-import textwrap

-import lance
 import pyarrow as pa
 import pyarrow.compute as pc
 import pytest
-from lance.blob import BlobType as LanceBlobType

 import lancedb
 from lancedb._blob import (
@@ -23,20 +18,6 @@ from lancedb.index import FTS
 from lancedb.schema import blob_column_paths, blob_v2_column_paths


-_HIDE_LANCE_BLOB = """\
-import importlib.abc
-import sys
-
-class _MissingLanceBlob(importlib.abc.MetaPathFinder):
-    def find_spec(self, fullname, path, target=None):
-        if fullname == "lance.blob" or fullname.startswith("lance.blob."):
-            raise ModuleNotFoundError(fullname, name="lance.blob")
-
-sys.modules.pop("lance.blob", None)
-sys.meta_path.insert(0, _MissingLanceBlob())
-"""
-
-
 def _blob_table(name, rows):
    db = lancedb.connect("memory:///")
    schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("image")])
@@ -70,181 +51,6 @@ def test_blob_factory_declares_v2_field():
    field = lancedb.blob("image")
    assert isinstance(field.type, pa.ExtensionType)
    assert field.type.extension_name == "lance.blob.v2"
-    assert lancedb.BlobType is LanceBlobType
-    assert type(field.type) is LanceBlobType
-
-
-def test_blob_type_works_without_pylance():
-    script = _HIDE_LANCE_BLOB + textwrap.dedent(
-        """\
-        import lancedb
-        import pyarrow as pa
-
-        field = lancedb.blob("image")
-        if not isinstance(field.type, pa.ExtensionType):
-            raise SystemExit("expected an extension type")
-        if field.type.extension_name != "lance.blob.v2":
-            raise SystemExit(field.type.extension_name)
-        if lancedb.BlobType is not type(field.type):
-            raise SystemExit("BlobType is not the field type class")
-        if lancedb.BlobType.__module__ != "lancedb.schema":
-            raise SystemExit(lancedb.BlobType.__module__)
-
-        db = lancedb.connect("memory:///")
-        table = db.create_table(
-            "images",
-            schema=pa.schema([pa.field("id", pa.int64()), field]),
-        )
-        table.add([{"id": 1, "image": b"hello"}])
-        result = (
-            table.merge_insert("id")
-            .when_matched_update_all()
-            .when_not_matched_insert_all()
-            .execute([{"id": 1, "image": b"updated"}, {"id": 2, "image": b"inserted"}])
-        )
-        if result.num_updated_rows != 1 or result.num_inserted_rows != 1:
-            raise SystemExit(
-                f"merge_insert rows updated={result.num_updated_rows} "
-                f"inserted={result.num_inserted_rows}"
-            )
-        """
-    )
-    result = subprocess.run(
-        [sys.executable, "-c", script],
-        capture_output=True,
-        text=True,
-        check=False,
-    )
-    assert result.returncode == 0, result.stderr
-
-
-def test_blob_resolves_pylance_type_without_eager_import():
-    script = textwrap.dedent(
-        """\
-        import sys
-        import lancedb
-
-        if "lance.blob" in sys.modules:
-            raise SystemExit("import lancedb imported lance.blob")
-        field = lancedb.blob("image")
-        from lance.blob import BlobType
-
-        if type(field.type) is not BlobType:
-            raise SystemExit(f"{type(field.type)} is not {BlobType}")
-        import lance
-
-        image = lance.blob_array([b"x"])
-        if type(image.type) is not BlobType:
-            raise SystemExit("blob_array used a different class")
-        if type(image.type) is not type(field.type):
-            raise SystemExit("field and array classes differ")
-        """
-    )
-    result = subprocess.run(
-        [sys.executable, "-c", script],
-        capture_output=True,
-        text=True,
-        check=False,
-    )
-    assert result.returncode == 0, result.stderr
-
-
-def test_blob_fallback_fails_if_name_already_registered():
-    script = _HIDE_LANCE_BLOB + textwrap.dedent(
-        """\
-        import pyarrow as pa
-
-        class OtherBlobType(pa.ExtensionType):
-            def __init__(self):
-                super().__init__(
-                    pa.struct([pa.field("data", pa.large_binary())]),
-                    "lance.blob.v2",
-                )
-
-            def __arrow_ext_serialize__(self):
-                return b""
-
-            @classmethod
-            def __arrow_ext_deserialize__(cls, storage_type, serialized):
-                return cls()
-
-        pa.register_extension_type(OtherBlobType())
-        import lancedb
-
-        try:
-            lancedb.blob("image")
-        except ValueError as err:
-            if "already registered" not in str(err):
-                raise SystemExit(err)
-        else:
-            raise SystemExit("expected ValueError")
-        """
-    )
-    result = subprocess.run(
-        [sys.executable, "-c", script],
-        capture_output=True,
-        text=True,
-        check=False,
-    )
-    assert result.returncode == 0, result.stderr
-
-
-def test_blob_type_rejects_competing_registration_with_pylance():
-    script = textwrap.dedent(
-        """\
-        import pyarrow as pa
-        import pyarrow.ipc
-
-        class OtherBlobType(pa.ExtensionType):
-            def __init__(self):
-                super().__init__(
-                    pa.struct(
-                        [
-                            pa.field("data", pa.large_binary()),
-                            pa.field("uri", pa.utf8()),
-                            pa.field("position", pa.uint64()),
-                            pa.field("size", pa.uint64()),
-                        ]
-                    ),
-                    "lance.blob.v2",
-                )
-
-            def __arrow_ext_serialize__(self):
-                return b""
-
-            @classmethod
-            def __arrow_ext_deserialize__(cls, storage_type, serialized):
-                return cls()
-
-        pa.register_extension_type(OtherBlobType())
-
-        from lance.blob import BlobType
-
-        if BlobType is OtherBlobType:
-            raise SystemExit("pylance BlobType was replaced")
-        schema = pa.schema([pa.field("value", BlobType())])
-        restored = pa.ipc.read_schema(schema.serialize())
-        if type(restored.field("value").type) is not OtherBlobType:
-            raise SystemExit(type(restored.field("value").type))
-
-        import lancedb
-
-        try:
-            lancedb.blob("image")
-        except ValueError as err:
-            if "__main__.OtherBlobType" not in str(err):
-                raise SystemExit(err)
-        else:
-            raise SystemExit("expected ValueError")
-        """
-    )
-    result = subprocess.run(
-        [sys.executable, "-c", script],
-        capture_output=True,
-        text=True,
-        check=False,
-    )
-    assert result.returncode == 0, result.stderr


 def test_blob_v2_column_paths_include_list_children():
@@ -397,292 +203,6 @@ def test_fetch_blobs_round_trip():
    assert [blobs[0].as_py(), blobs[1].as_py()] == [b"alpha", b"beta"]


-def test_merge_insert_writes_python_bytes():
-    table = _blob_table("merge_bytes", [{"id": 1, "image": b"before"}])
-    result = (
-        table.merge_insert("id")
-        .when_matched_update_all()
-        .when_not_matched_insert_all()
-        .execute([{"id": 1, "image": b"updated"}, {"id": 2, "image": b"inserted"}])
-    )
-    assert result.num_updated_rows == 1
-    assert result.num_inserted_rows == 1
-    by_id = _row_ids_by_id(table)
-    blobs = table.fetch_blobs("image", [by_id[1], by_id[2]])
-    assert blobs.to_pylist() == [b"updated", b"inserted"]
-
-
-def test_merge_insert_bytes_after_reopen_without_touching_blob_type(tmp_path):
-    db = lancedb.connect(tmp_path)
-    schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("image")])
-    table = db.create_table("images", schema=schema)
-    table.add([{"id": 1, "image": b"hello"}])
-
-    script = textwrap.dedent(
-        f"""\
-        import lancedb
-
-        db = lancedb.connect({str(tmp_path)!r})
-        table = db.open_table("images")
-        image_type = table.schema.field("image").type
-        if type(image_type).__name__ != "StructType":
-            raise SystemExit(f"expected StructType, got {{type(image_type)}}")
-        result = (
-            table.merge_insert("id")
-            .when_matched_update_all()
-            .when_not_matched_insert_all()
-            .execute(
-                [{{"id": 1, "image": b"updated"}}, {{"id": 2, "image": b"inserted"}}]
-            )
-        )
-        if result.num_updated_rows != 1 or result.num_inserted_rows != 1:
-            raise SystemExit(
-                f"rows updated={{result.num_updated_rows}} "
-                f"inserted={{result.num_inserted_rows}}"
-            )
-        hits = table.search().with_row_id(True).limit(10).to_arrow()
-        by_id = dict(zip(hits["id"].to_pylist(), hits["_rowid"].to_pylist()))
-        blobs = table.fetch_blobs("image", [by_id[1], by_id[2]])
-        if blobs.to_pylist() != [b"updated", b"inserted"]:
-            raise SystemExit(blobs.to_pylist())
-        """
-    )
-    result = subprocess.run(
-        [sys.executable, "-c", script],
-        capture_output=True,
-        text=True,
-        check=False,
-    )
-    assert result.returncode == 0, result.stderr
-
-
-def test_merge_insert_bytes_after_reopen_without_pylance(tmp_path):
-    db = lancedb.connect(tmp_path)
-    schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("image")])
-    table = db.create_table("images", schema=schema)
-    table.add([{"id": 1, "image": b"hello"}])
-
-    script = _HIDE_LANCE_BLOB + textwrap.dedent(
-        f"""\
-        import lancedb
-
-        db = lancedb.connect({str(tmp_path)!r})
-        table = db.open_table("images")
-        image_type = table.schema.field("image").type
-        if type(image_type).__name__ != "StructType":
-            raise SystemExit(f"expected StructType, got {{type(image_type)}}")
-        result = (
-            table.merge_insert("id")
-            .when_matched_update_all()
-            .when_not_matched_insert_all()
-            .execute(
-                [{{"id": 1, "image": b"updated"}}, {{"id": 2, "image": b"inserted"}}]
-            )
-        )
-        if result.num_updated_rows != 1 or result.num_inserted_rows != 1:
-            raise SystemExit(
-                f"rows updated={{result.num_updated_rows}} "
-                f"inserted={{result.num_inserted_rows}}"
-            )
-        hits = table.search().with_row_id(True).limit(10).to_arrow()
-        by_id = dict(zip(hits["id"].to_pylist(), hits["_rowid"].to_pylist()))
-        blobs = table.fetch_blobs("image", [by_id[1], by_id[2]])
-        if blobs.to_pylist() != [b"updated", b"inserted"]:
-            raise SystemExit(blobs.to_pylist())
-        """
-    )
-    result = subprocess.run(
-        [sys.executable, "-c", script],
-        capture_output=True,
-        text=True,
-        check=False,
-    )
-    assert result.returncode == 0, result.stderr
-
-
-def test_merge_insert_blob_array_into_reopened_unregistered_table(tmp_path):
-    db = lancedb.connect(tmp_path)
-    schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("image")])
-    table = db.create_table("images", schema=schema)
-    table.add([{"id": 1, "image": b"before"}])
-
-    script = textwrap.dedent(
-        f"""\
-        import pyarrow as pa
-        import lancedb
-
-        db = lancedb.connect({str(tmp_path)!r})
-        table = db.open_table("images")
-        image_type = table.schema.field("image").type
-        if type(image_type).__name__ != "StructType":
-            raise SystemExit(
-                f"expected StructType before lance import, got {{type(image_type)}}"
-            )
-
-        import lance
-
-        updates = pa.Table.from_arrays(
-            [
-                pa.array([1, 2], type=pa.int64()),
-                lance.blob_array([b"updated", b"inserted"]),
-            ],
-            names=["id", "image"],
-        )
-        result = (
-            table.merge_insert("id")
-            .when_matched_update_all()
-            .when_not_matched_insert_all()
-            .execute(updates)
-        )
-        if result.num_updated_rows != 1 or result.num_inserted_rows != 1:
-            raise SystemExit(
-                f"rows updated={{result.num_updated_rows}} "
-                f"inserted={{result.num_inserted_rows}}"
-            )
-        hits = table.search().with_row_id(True).limit(10).to_arrow()
-        by_id = dict(zip(hits["id"].to_pylist(), hits["_rowid"].to_pylist()))
-        blobs = table.fetch_blobs("image", [by_id[1], by_id[2]])
-        if blobs.to_pylist() != [b"updated", b"inserted"]:
-            raise SystemExit(blobs.to_pylist())
-        """
-    )
-    result = subprocess.run(
-        [sys.executable, "-c", script],
-        capture_output=True,
-        text=True,
-        check=False,
-    )
-    assert result.returncode == 0, result.stderr
-
-
-def test_add_all_null_blob_column():
-    db = lancedb.connect("memory:///")
-    schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("image")])
-    table = db.create_table("all_null", schema=schema)
-    table.add([{"id": 1, "image": None}, {"id": 2, "image": None}])
-    by_id = _row_ids_by_id(table)
-    blobs = table.fetch_blobs("image", [by_id[1], by_id[2]])
-    assert blobs.to_pylist() == [None, None]
-
-
-def test_create_table_nested_blob_schema_without_rows():
-    db = lancedb.connect("memory:///")
-    schema = pa.schema(
-        [
-            pa.field("id", pa.int64()),
-            pa.field("info", pa.struct([lancedb.blob("blob")])),
-            pa.field("images", pa.list_(lancedb.blob("image"))),
-        ]
-    )
-    table = db.create_table("nested_empty", schema=schema)
-    assert table.count_rows() == 0
-
-
-def test_merge_insert_nested_blob_dicts():
-    db = lancedb.connect("memory:///")
-    info = pa.StructArray.from_arrays(
-        [
-            pa.array(["first"], type=pa.string()),
-            _blob_array("blob", [b"before"]),
-        ],
-        names=["name", "blob"],
-    )
-    data = pa.Table.from_arrays(
-        [pa.array([1], type=pa.int64()), info],
-        names=["id", "info"],
-    )
-    table = db.create_table("nested_merge", data=data)
-    result = (
-        table.merge_insert("id")
-        .when_matched_update_all()
-        .execute([{"id": 1, "info": {"name": "first", "blob": b"after"}}])
-    )
-    assert result.num_updated_rows == 1
-    by_id = _row_ids_by_id(table)
-    blobs = table.fetch_blobs("info.blob", [by_id[1]])
-    assert blobs.to_pylist() == [b"after"]
-
-
-def _list_blob_table(name):
-    db = lancedb.connect("memory:///")
-    blob_field = lancedb.blob("image")
-    images = pa.ListArray.from_arrays(
-        pa.array([0, 1], type=pa.int32()), _blob_array("image", [b"before"])
-    )
-    data = pa.Table.from_arrays(
-        [pa.array([1], type=pa.int64()), images],
-        schema=pa.schema(
-            [pa.field("id", pa.int64()), pa.field("images", pa.list_(blob_field))]
-        ),
-    )
-    return db.create_table(name, data=data)
-
-
-def test_merge_insert_list_blob_dicts():
-    table = _list_blob_table("list_merge")
-    result = (
-        table.merge_insert("id")
-        .when_matched_update_all()
-        .when_not_matched_insert_all()
-        .execute([{"id": 1, "images": [b"one", b"two"]}, {"id": 2, "images": None}])
-    )
-    assert result.num_updated_rows == 1
-    assert result.num_inserted_rows == 1
-    hits = table.search().limit(10).to_arrow()
-    sizes = {
-        row["id"]: None if row["images"] is None else [d["size"] for d in row["images"]]
-        for row in hits.to_pylist()
-    }
-    assert sizes == {1: [3, 3], 2: None}
-
-
-def test_list_blob_column_queries_as_raw_descriptors():
-    table = _list_blob_table("list_query")
-    hits = table.search().limit(10).to_arrow()
-    element = hits.schema.field("images").type.value_type
-    assert pa.types.is_struct(element)
-    assert "_lance_row_id" not in element.names
-    with pytest.raises(ValueError, match="expected struct before segment"):
-        table.fetch_blobs("images.image", [0])
-
-
-def test_row_addressable_paths_exclude_list_children():
-    from lancedb.schema import row_addressable_blob_v2_paths
-
-    schema = pa.schema(
-        [
-            pa.field("id", pa.int64()),
-            pa.field("info", pa.struct([lancedb.blob("blob")])),
-            pa.field("images", pa.list_(lancedb.blob("image"))),
-        ]
-    )
-    assert blob_v2_column_paths(schema) == ["info.blob", "images.image"]
-    assert row_addressable_blob_v2_paths(schema) == ["info.blob"]
-
-
-def test_merge_insert_writes_pylance_blob_array():
-    table = _blob_table("merge_pylance", [{"id": 1, "image": b"before"}])
-    image = lance.blob_array([b"updated", b"inserted"])
-    assert type(image.type) is LanceBlobType
-    assert type(image.type) is type(lancedb.BlobType())
-    updates = pa.Table.from_arrays(
-        [pa.array([1, 2], type=pa.int64()), image], names=["id", "image"]
-    )
-
-    result = (
-        table.merge_insert("id")
-        .when_matched_update_all()
-        .when_not_matched_insert_all()
-        .execute(updates)
-    )
-
-    assert result.num_updated_rows == 1
-    assert result.num_inserted_rows == 1
-    by_id = _row_ids_by_id(table)
-    blobs = table.fetch_blobs("image", [by_id[1], by_id[2]])
-    assert blobs.to_pylist() == [b"updated", b"inserted"]
-
-
 def test_fetch_blobs_accepts_query_result():
    table = _blob_table("from_result", [{"id": 1, "image": b"gamma"}])
    hits = table.search().limit(10).to_arrow()
@@ -12,8 +12,6 @@ from datetime import date
 import http.server
 import json
 from pathlib import Path
-import subprocess
-import sys
 import threading
 from typing import Optional

@@ -21,13 +19,7 @@ import pyarrow as pa
 import pytest

 import lancedb
-from lancedb.functions import (
-    PythonRuntimeSpec,
-    UdfDefinition,
-    _canonical_arrow_type,
-    _GRAMMAR_PRIMITIVES,
-    udf,
-)
+from lancedb.functions import UdfDefinition, udf

 THRESHOLD = 20
 _CACHE = None
@@ -69,80 +61,6 @@ def test_scalar_udf_matches_shared_registration_golden_and_remains_callable():
    }


-def _main_udf_source(
-    *, threshold: int = 20, input_annotation: str = "int", comparison: str = ">="
-) -> str:
-    return (
-        "from __future__ import annotations\n"
-        "from lancedb.functions import udf\n"
-        f"THRESHOLD = {threshold}\n"
-        "\n"
-        "@udf\n"
-        f"def label(value: {input_annotation}) -> str:\n"
-        f"    return 'big' if value {comparison} THRESHOLD else 'small'\n"
-        "\n"
-        "assert label.__module__ == '__main__'\n"
-        "print(label.registration_request.to_canonical_json())\n"
-    )
-
-
-def _run_main_udf(path: Path, source: str) -> dict:
-    path.parent.mkdir(parents=True, exist_ok=True)
-    path.write_text(source)
-    result = subprocess.run(
-        [sys.executable, str(path)],
-        check=True,
-        capture_output=True,
-        text=True,
-    )
-    return json.loads(result.stdout)
-
-
-def test_main_udf_registration_identity_is_stable_across_processes_and_paths(
-    tmp_path,
-):
-    source = _main_udf_source()
-    original_path = tmp_path / "original" / "job.py"
-    moved_path = tmp_path / "moved" / "renamed_job.py"
-
-    original_runs = [_run_main_udf(original_path, source) for _ in range(2)]
-    moved_run = _run_main_udf(moved_path, source)
-
-    assert len({run["artifact"]["digest"] for run in [*original_runs, moved_run]}) == 1
-    assert all(
-        run["signature"] == original_runs[0]["signature"]
-        for run in [original_runs[1], moved_run]
-    )
-    assert original_runs[0] == original_runs[1] == moved_run
-
-    body_change = _run_main_udf(
-        tmp_path / "changes" / "body.py", _main_udf_source(comparison=">")
-    )
-    global_change = _run_main_udf(
-        tmp_path / "changes" / "global.py", _main_udf_source(threshold=21)
-    )
-    annotation_change = _run_main_udf(
-        tmp_path / "changes" / "annotation.py",
-        _main_udf_source(input_annotation="float"),
-    )
-
-    baseline = original_runs[0]
-    assert baseline["signature"] == body_change["signature"]
-    assert baseline["signature"] == global_change["signature"]
-    assert baseline["signature"] != annotation_change["signature"]
-    assert (
-        len(
-            {
-                baseline["artifact"]["digest"],
-                body_change["artifact"]["digest"],
-                global_change["artifact"]["digest"],
-                annotation_change["artifact"]["digest"],
-            }
-        )
-        == 4
-    )
-
-
 def _run_packaged(definition, *args):
    """Execute the shipped artifact in a fresh namespace, as a worker would."""
    source = base64.b64decode(definition.registration_request.artifact.content.data)
@@ -171,58 +89,6 @@ def test_udf_conda_environment():
        udf(name="channels", conda_channels=["conda-forge"])(lambda value: value)


-def test_udf_gpu_marker_uses_gpu_runtime():
-    @udf(pip=["cupy-cuda12x"], gpu=True)
-    def double_on_gpu(value: int) -> int:
-        return value * 2
-
-    request = json.loads(double_on_gpu.registration_request.to_canonical_json())
-    assert request["runtime"]["kind"] == "python_v2"
-    assert request["runtime"]["gpu"] is True
-
-    @udf(pip=["pyarrow"])
-    def cpu_function(value: int) -> int:
-        return value
-
-    cpu_runtime = json.loads(cpu_function.registration_request.to_canonical_json())[
-        "runtime"
-    ]
-    assert cpu_runtime["kind"] == "python"
-    assert "gpu" not in cpu_runtime
-
-    def identity(value: int) -> int:
-        return value
-
-    for invalid in [None, 0, 1, -1, 1.5, "", "true", "1", "H100"]:
-        with pytest.raises(ValueError, match="gpu must be a boolean"):
-            udf(name="invalid_gpu", gpu=invalid)(identity)
-
-    base_runtime = {
-        "kind": "python_v2",
-        "python_version": "3.12",
-        "environment": {"kind": "pip"},
-    }
-    runtime = PythonRuntimeSpec.model_validate({**base_runtime, "gpu": True})
-    assert runtime.gpu is True
-    for invalid in [False, 1, 0, "", "true", "1", "H100"]:
-        with pytest.raises(ValueError, match="runtime.gpu must be true"):
-            PythonRuntimeSpec.model_validate({**base_runtime, "gpu": invalid})
-
-
-def test_unknown_runtime_discards_payload_before_known_field_validation():
-    for payload in [
-        {"kind": "python_v3", "gpu": {"model": "H100"}},
-        {"kind": "python_v3", "resources": []},
-        {
-            "kind": "python_v3",
-            "environment": {"kind": []},
-            "python_version": 3.15,
-        },
-    ]:
-        runtime = PythonRuntimeSpec.model_validate(payload)
-        assert runtime.to_canonical_json() == '{"kind":"python_v3"}'
-
-
 def test_udf_packages_attribute_access_and_body_imports():
    @udf
    def word_norm(body: str) -> float:
@@ -302,7 +168,9 @@ def test_udf_resolves_module_globals_before_builtins(tmp_path):
        udf(module.uses_callable_shadow)


-def test_canonical_arrow_type_prefers_the_compact_grammar():
+def test_canonical_arrow_type_is_exactly_the_grammar():
+    from lancedb.functions import _GRAMMAR_PRIMITIVES, _canonical_arrow_type
+
    golden = json.loads(
        (
            Path(__file__).parents[3]
@@ -313,19 +181,14 @@ def test_canonical_arrow_type_prefers_the_compact_grammar():
        case["arrow_type"] for case in golden["valid"] if "<" not in case["arrow_type"]
    ]
    assert [name for _, name in _GRAMMAR_PRIMITIVES] == primitives
-    assert _canonical_arrow_type(pa.list_(pa.field("item", pa.float32(), False))) == (
-        "list<float32>"
-    )
-    assert (
-        _canonical_arrow_type(pa.large_list(pa.field("item", pa.float32(), False)))
-        == "large_list<float32>"
-    )
    for outside in [
        pa.timestamp("us"),
        pa.decimal128(10, 2),
+        pa.large_string(),
        pa.large_binary(),
        pa.binary(4),
        pa.duration("s"),
+        pa.struct([pa.field("a", pa.int32())]),
        pa.list_(pa.float32(), 0),
        pa.list_(pa.timestamp("us")),
    ]:
@@ -515,27 +378,14 @@ def test_udf_recursion_versus_a_rebound_module_name(tmp_path):
        udf(raw_fact)


-def test_canonical_arrow_type_uses_exact_json_for_list_child_properties():
-    nullable = pa.list_(pa.float32())
-    assert json.loads(_canonical_arrow_type(nullable)) == {
-        "type": "list",
-        "fields": [
-            {
-                "name": "item",
-                "nullable": True,
-                "type": {"type": "float32"},
-            }
-        ],
-    }
-    named = pa.list_(pa.field("custom", pa.float32(), nullable=False))
-    assert json.loads(_canonical_arrow_type(named))["fields"][0]["name"] == "custom"
+def test_canonical_arrow_type_rejects_unrepresentable_list_children():
+    from lancedb.functions import _canonical_arrow_type
+
    for outside in [
+        pa.list_(pa.float32()),  # pyarrow default: nullable child
+        pa.list_(pa.field("custom", pa.float32(), nullable=False)),
        pa.list_(pa.field("item", pa.float32(), nullable=False, metadata={"k": "v"})),
        pa.list_(pa.field("item", pa.float32(), nullable=False), 0),
-        pa.list_(
-            pa.field("item", pa.float32(), nullable=False, metadata={"k": "v"}), 3
-        ),
-        pa.list_(pa.field("custom", pa.float32(), nullable=False), 3),
    ]:
        with pytest.raises(TypeError, match="unsupported Arrow type"):
            _canonical_arrow_type(outside)
@@ -545,29 +395,6 @@ def test_canonical_arrow_type_uses_exact_json_for_list_child_properties():
        )
        == "fixed_size_list<float32, 3>"
    )
-    fixed = json.loads(_canonical_arrow_type(pa.list_(pa.float32(), 3)))
-    assert fixed == {
-        "type": "fixed_size_list",
-        "fields": [
-            {
-                "name": "item",
-                "nullable": True,
-                "type": {"type": "float32"},
-            }
-        ],
-        "length": 3,
-    }
-    large = json.loads(_canonical_arrow_type(pa.large_list(pa.float32())))
-    assert large["type"] == "large_list"
-    assert large["fields"][0]["nullable"] is True
-
-    for invalid_struct in [
-        pa.struct([]),
-        pa.struct([pa.field("a", pa.int32()), pa.field("a", pa.int64())]),
-        pa.struct([pa.field("", pa.int32())]),
-    ]:
-        with pytest.raises(TypeError, match="unsupported Arrow type"):
-            _canonical_arrow_type(invalid_struct)


 def _calls_missing(value: int) -> int:
@@ -605,7 +432,6 @@ def _arrow_type_from_golden(spec: dict) -> pa.DataType:
        "null": pa.null(),
        "bool": pa.bool_(),
        "utf8": pa.string(),
-        "large_utf8": pa.large_string(),
        "binary": pa.binary(),
        "float16": pa.float16(),
        "float32": pa.float32(),
@@ -622,6 +448,8 @@ def test_arrow_type_grammar_matches_the_shared_golden():
            / "rust/lancedb/tests/fixtures/first_class_functions/v1/arrow_types.json"
        ).read_text()
    )
+    from lancedb.functions import _canonical_arrow_type
+
    emitted = {
        case["arrow_type"]: _canonical_arrow_type(_arrow_type_from_golden(case["json"]))
        for case in golden["valid"]
@@ -654,250 +482,6 @@ def test_explicit_arrow_schema_is_deterministic():
    assert signature.output.nullable is False


-def test_blob_fields_use_the_scalar_function_semantic_type():
-    @udf(
-        input_schema=pa.schema([lancedb.blob("image", nullable=False)]),
-        output_schema=lancedb.blob("result", nullable=False),
-    )
-    def copy_blob(image):
-        return image
-
-    signature = copy_blob.registration_request.signature
-    assert signature.inputs[0].arrow_type == "blob_v2"
-    assert signature.output.kind == "scalar"
-    assert signature.output.arrow_type == "blob_v2"
-
-
-def test_named_struct_function_can_include_a_blob_result_field():
-    @udf(
-        input_schema=pa.schema([lancedb.blob("image", nullable=False)]),
-        output_schema=pa.schema(
-            [
-                lancedb.blob("thumbnail", nullable=False),
-                pa.field("width", pa.int32(), nullable=False),
-            ]
-        ),
-    )
-    def inspect_blob(image):
-        return {"thumbnail": image, "width": 1}
-
-    output = inspect_blob.registration_request.signature.output
-    assert output.kind == "named_struct"
-    assert [(field.name, field.arrow_type) for field in output.fields] == [
-        ("thumbnail", "blob_v2"),
-        ("width", "int32"),
-    ]
-
-
-def test_metadata_marked_blob_field_uses_the_semantic_type():
-    extension = lancedb.blob("image", nullable=False).type
-    storage = (
-        extension.storage_type if isinstance(extension, pa.ExtensionType) else extension
-    )
-    metadata_blob = pa.field(
-        "image",
-        storage,
-        nullable=False,
-        metadata={"ARROW:extension:name": "lance.blob.v2"},
-    )
-
-    @udf(
-        input_schema=pa.schema([metadata_blob]),
-        output_schema=pa.field("size", pa.int64(), nullable=False),
-    )
-    def blob_size(image):
-        return len(image)
-
-    assert blob_size.registration_request.signature.inputs[0].arrow_type == "blob_v2"
-
-
-def test_blob_marker_rejects_invalid_storage_layout():
-    malformed = pa.field(
-        "image",
-        pa.int64(),
-        nullable=False,
-        metadata={"ARROW:extension:name": "lance.blob.v2"},
-    )
-
-    with pytest.raises(TypeError, match="requires a supported Blob storage layout"):
-
-        @udf(
-            input_schema=pa.schema([malformed]),
-            output_schema=pa.field("size", pa.int64(), nullable=False),
-        )
-        def blob_size(image):
-            return len(image)
-
-
-def test_nested_blob_signature_field_has_a_clear_error():
-    nested = pa.field(
-        "value",
-        pa.struct([lancedb.blob("image", nullable=False)]),
-        nullable=False,
-    )
-    with pytest.raises(TypeError, match="nested Blob v2 fields are not supported"):
-
-        @udf(
-            input_schema=pa.schema([nested]),
-            output_schema=pa.field("size", pa.int64(), nullable=False),
-        )
-        def blob_size(value):
-            return len(value["image"])
-
-
-def test_nested_non_blob_extension_is_not_silently_unwrapped():
-    class TestExtension(pa.ExtensionType):
-        def __init__(self):
-            super().__init__(pa.int64(), "test.function.extension")
-
-        def __arrow_ext_serialize__(self):
-            return b""
-
-        @classmethod
-        def __arrow_ext_deserialize__(cls, storage_type, serialized):
-            return cls()
-
-    nested = pa.field(
-        "value",
-        pa.struct([pa.field("extended", TestExtension(), nullable=False)]),
-        nullable=False,
-    )
-    with pytest.raises(TypeError, match="unsupported Arrow type"):
-
-        @udf(
-            input_schema=pa.schema([nested]),
-            output_schema=pa.field("result", pa.int64(), nullable=False),
-        )
-        def extension_value(value):
-            return value["extended"]
-
-
-def test_explicit_large_utf8_schemas_use_the_canonical_function_name():
-    input_schema = pa.schema([pa.field("text", pa.large_string(), nullable=True)])
-    output_schema = pa.field("result", pa.large_string(), nullable=False)
-
-    @udf(input_schema=input_schema, output_schema=output_schema)
-    def preserve(text):
-        return text
-
-    signature = preserve.registration_request.signature
-    assert signature.inputs[0].arrow_type == "large_utf8"
-    assert signature.inputs[0].nullable is True
-    assert signature.output.arrow_type == "large_utf8"
-    assert signature.output.nullable is False
-
-    nested = pa.struct([pa.field("text", pa.large_string(), nullable=True)])
-    assert json.loads(_canonical_arrow_type(nested)) == {
-        "type": "struct",
-        "fields": [
-            {
-                "name": "text",
-                "nullable": True,
-                "type": {"type": "large_utf8"},
-            }
-        ],
-    }
-
-
-def test_nested_struct_output_uses_canonical_exact_json():
-    token = pa.struct(
-        [
-            pa.field("position", pa.int32(), nullable=False),
-            pa.field("value", pa.string(), nullable=False),
-            pa.field("length", pa.int32(), nullable=False),
-        ]
-    )
-    analysis = pa.struct(
-        [
-            pa.field("normalized_text", pa.string(), nullable=False),
-            pa.field("has_content", pa.bool_(), nullable=False),
-            pa.field(
-                "metrics",
-                pa.struct(
-                    [
-                        pa.field("character_count", pa.int64(), nullable=False),
-                        pa.field("word_count", pa.int32(), nullable=False),
-                        pa.field("average_word_length", pa.float64(), nullable=False),
-                    ]
-                ),
-                nullable=False,
-            ),
-            pa.field(
-                "diagnostics",
-                pa.struct(
-                    [
-                        pa.field("status", pa.string(), nullable=False),
-                        pa.field(
-                            "normalization",
-                            pa.struct(
-                                [
-                                    pa.field("changed", pa.bool_(), nullable=False),
-                                    pa.field(
-                                        "original_length", pa.int64(), nullable=False
-                                    ),
-                                ]
-                            ),
-                            nullable=False,
-                        ),
-                    ]
-                ),
-                nullable=False,
-            ),
-            pa.field(
-                "token_preview",
-                pa.list_(pa.field("item", token, nullable=False)),
-                nullable=False,
-            ),
-        ]
-    )
-
-    @udf(
-        input_schema=pa.schema([pa.field("text", pa.string(), nullable=False)]),
-        output_schema=pa.field("analysis", analysis, nullable=False),
-    )
-    def analyze(text):
-        return {"normalized_text": text}
-
-    output = analyze.registration_request.signature.output
-    assert output.kind == "named_struct"
-    assert [field.name for field in output.fields] == [
-        "normalized_text",
-        "has_content",
-        "metrics",
-        "diagnostics",
-        "token_preview",
-    ]
-    metrics = json.loads(output.fields[2].arrow_type)
-    assert metrics == {
-        "type": "struct",
-        "fields": [
-            {
-                "name": "character_count",
-                "nullable": False,
-                "type": {"type": "int64"},
-            },
-            {
-                "name": "word_count",
-                "nullable": False,
-                "type": {"type": "int32"},
-            },
-            {
-                "name": "average_word_length",
-                "nullable": False,
-                "type": {"type": "float64"},
-            },
-        ],
-    }
-    preview = json.loads(output.fields[4].arrow_type)
-    assert preview["type"] == "list"
-    assert preview["fields"][0]["type"]["type"] == "struct"
-    assert [field["name"] for field in preview["fields"][0]["type"]["fields"]] == [
-        "position",
-        "value",
-        "length",
-    ]
-
-
 def test_annotation_and_explicit_schema_validation_fail_closed():
    with pytest.raises(TypeError, match="missing Function annotations"):

@@ -941,72 +525,6 @@ def test_annotation_and_explicit_schema_validation_fail_closed():
        def nullable_explicit(value):
            return value

-    for invalid_field in [
-        pa.field("", pa.int32(), nullable=False),
-        pa.field("result", pa.int32(), nullable=False, metadata={"k": "v"}),
-    ]:
-        with pytest.raises(TypeError, match="unsupported Arrow type"):
-
-            @udf(
-                input_schema=pa.schema([pa.field("value", pa.int64())]),
-                output_schema=pa.schema([invalid_field]),
-            )
-            def invalid_explicit_field(value):
-                return value
-
-    with pytest.raises(TypeError, match="unsupported Arrow type"):
-
-        @udf(
-            input_schema=pa.schema(
-                [pa.field("value", pa.int64(), metadata={"k": "v"})]
-            ),
-            output_schema=pa.int64(),
-        )
-        def input_field_metadata(value):
-            return value
-
-    with pytest.raises(TypeError, match="unsupported Arrow type"):
-
-        @udf(
-            input_schema=pa.schema([pa.field("value", pa.int64())]),
-            output_schema=pa.field(
-                "result", pa.int64(), nullable=False, metadata={"k": "v"}
-            ),
-        )
-        def scalar_output_field_metadata(value):
-            return value
-
-    struct_type = pa.struct([pa.field("value", pa.int64(), nullable=False)])
-    with pytest.raises(TypeError, match="unsupported Arrow type"):
-
-        @udf(
-            input_schema=pa.schema([pa.field("value", pa.int64())]),
-            output_schema=pa.field(
-                "result", struct_type, nullable=False, metadata={"k": "v"}
-            ),
-        )
-        def struct_output_field_metadata(value):
-            return {"value": value}
-
-    for input_schema, output_schema in [
-        (
-            pa.schema([pa.field("value", pa.int64())], metadata={"k": "v"}),
-            pa.int64(),
-        ),
-        (
-            pa.schema([pa.field("value", pa.int64())]),
-            pa.schema(
-                [pa.field("result", pa.int64(), nullable=False)],
-                metadata={"k": "v"},
-            ),
-        ),
-    ]:
-        with pytest.raises(TypeError, match="schema metadata"):
-
-            @udf(input_schema=input_schema, output_schema=output_schema)
-            def schema_metadata(value):
-                return value
-

 def test_local_function_catalog_operations_are_not_supported(tmp_path):
    db = lancedb.connect(tmp_path)
@@ -1017,8 +535,6 @@ def test_local_function_catalog_operations_are_not_supported(tmp_path):
        db.create_function_async(normalize_score)
    with pytest.raises(NotImplementedError, match=message):
        db.get_function("normalize_score", version="fv_exact")
-    with pytest.raises(NotImplementedError, match=message):
-        db.drop_function("normalize_score", version="fv_exact")


@contextlib.contextmanager
@@ -1064,12 +580,6 @@ def _mock_remote_function_catalog():
                    "version": "fv_exact",
                }
                response = state["version"]
-            elif self.path == "/v1/functions/drop":
-                assert body == {
-                    "name": "normalize_score",
-                    "version": "fv_exact",
-                }
-                response = {"dropped": True}
            else:
                status = 404
                response = {"error": "not found"}
@@ -1128,40 +638,3 @@ def test_blocking_remote_registration_returns_function_version():
        "/v1/functions/create",
        "/v1/jobs/describe",
    ]
-
-
-def test_remote_drop_function_sends_exact_version():
-    with _mock_remote_function_catalog() as (host, state):
-        db = lancedb.connect(
-            "db://dev",
-            api_key="fake",
-            host_override=host,
-            client_config={"retry_config": {"retries": 0}},
-        )
-        assert db.drop_function("normalize_score", version="fv_exact") is True
-
-    assert state["requests"] == [
-        (
-            "/v1/functions/drop",
-            {"name": "normalize_score", "version": "fv_exact"},
-        )
-    ]
-
-
-@pytest.mark.asyncio
-async def test_async_remote_drop_function_sends_exact_version():
-    with _mock_remote_function_catalog() as (host, state):
-        db = await lancedb.connect_async(
-            "db://dev",
-            api_key="fake",
-            host_override=host,
-            client_config={"retry_config": {"retries": 0}},
-        )
-        assert await db.drop_function("normalize_score", version="fv_exact") is True
-
-    assert state["requests"] == [
-        (
-            "/v1/functions/drop",
-            {"name": "normalize_score", "version": "fv_exact"},
-        )
-    ]
@@ -266,38 +266,3 @@ async def test_async_namespace_connection_materialized_views(tmp_path):
            handle._route_pushdown_to_rust == through_namespace._route_pushdown_to_rust
        )
        assert handle._namespace_path == through_namespace._namespace_path
-
-
-def test_namespaced_select_kind_is_read_and_unknown_kinds_are_refused():
-    import json
-
-    import pyarrow as pa
-
-    from lancedb.materialized_view import _definition_from_schema
-
-    def schema_with(definition: dict) -> pa.Schema:
-        return pa.schema([pa.field("id", pa.int32())]).with_metadata(
-            {b"mv.definition": json.dumps(definition).encode()}
-        )
-
-    # "namespaced_select" is the namespaced form of "select": same shape,
-    # a separate kind so readers that predate it refuse instead of
-    # resolving the source at the root.
-    definition = _definition_from_schema(
-        schema_with(
-            {
-                "kind": "namespaced_select",
-                "source_table": "people",
-                "source_namespace": ["ns"],
-                "projections": [{"output": "name", "expression": "name"}],
-            }
-        ),
-        "v",
-    )
-    assert definition.source_table == "people"
-    assert definition.source_namespace == ["ns"]
-
-    with pytest.raises(NotImplementedError, match="cannot refresh"):
-        _definition_from_schema(
-            schema_with({"kind": "select_v3", "source_table": "people"}), "v"
-        )
@@ -1923,21 +1923,6 @@ def test_take_queries(tmp_path):
        17,
    ]

-    # Duplicate offsets are occurrences, not set members. Ordering is unspecified.
-    assert sorted(table.take_offsets([5, 2, 5, 17]).to_pandas()["idx"].to_list()) == [
-        2,
-        5,
-        5,
-        17,
-    ]
-
-    # Converting a take builder to its serializable query representation must
-    # retain occurrence metadata and execute with the same multiplicity.
-    query = table.take_offsets([5, 2, 5, 17]).select(["idx"]).to_query_object()
-    assert query.take_offsets == [5, 2, 5, 17]
-    converted = table._execute_query(query).read_all()
-    assert sorted(converted["idx"].to_pylist()) == [2, 5, 5, 17]
-
    # Take by row id
    assert list(
        sorted(table.take_row_ids([5, 2, 17]).to_pandas()["idx"].to_list())
@@ -479,49 +479,24 @@ def test_remote_permutation_is_picklable():
                match = re.search(
                    r"_rowoffset\s+in\s+\((.*?)\)", body["filter"], re.IGNORECASE
                )
-                offsets = list(
-                    dict.fromkeys(int(o.strip()) for o in match.group(1).split(","))
-                )
+                offsets = [int(o.strip()) for o in match.group(1).split(",")]
            else:
                offsets = list(range(len(rows)))
-            columns = body.get("columns") or ["a"]
-            table = pa.table(
-                {
-                    column: (
-                        [rows[offset] for offset in offsets]
-                        if column == "a"
-                        else offsets
-                    )
-                    for column in columns
-                }
-            )
+            table = pa.table({"a": [rows[offset] for offset in offsets]})

            request.send_response(200)
            request.send_header("Content-Type", "application/vnd.apache.arrow.file")
            request.end_headers()
            with pa.ipc.new_file(request.wfile, schema=table.schema) as writer:
-                writer.write_table(table, max_chunksize=2)
+                writer.write_table(table)
        else:
            request.send_response(404)
            request.end_headers()

    with mock_lancedb_connection(handler) as db:
-        table = db.open_table("test")
-        assert table.take_offsets([0, 2, 0, 4]).to_list() == [
-            {"a": 0},
-            {"a": 0},
-            {"a": 2},
-            {"a": 4},
-        ]
-
-        permutation = Permutation.identity(table)
+        permutation = Permutation.identity(db.open_table("test"))
        restored = pickle.loads(pickle.dumps(permutation))
-        assert restored.__getitems__([0, 2, 0, 4]) == [
-            {"a": 0},
-            {"a": 2},
-            {"a": 0},
-            {"a": 4},
-        ]
+        assert restored.__getitems__([0, 2, 4]) == [{"a": 0}, {"a": 2}, {"a": 4}]


 def test_create_table_exist_ok():
@@ -4087,29 +4087,6 @@ def test_computed_column_rejects_transforms_and_computed_together(tmp_path):
        table.add_columns({"a": "x + 1"}, computed={"b": "x * 2"})


-def test_computed_column_blob_projection_inherits_semantics(tmp_path):
-    schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("image")])
-    db = lancedb.connect(tmp_path)
-    table = db.create_table("computed_column_blob", schema=schema)
-    table.add(
-        [
-            {"id": 1, "image": b"hello"},
-            {"id": 2, "image": b""},
-            {"id": 3, "image": None},
-        ]
-    )
-
-    table.add_columns(computed={"image_copy": "image", "second_copy": "image_copy"})
-    assert table.refresh_column("image_copy").rows_filled == 2
-    assert table.refresh_column("second_copy").rows_filled == 2
-    assert table.blob_columns() == ["image", "image_copy", "second_copy"]
-
-    hits = table.search().with_row_id(True).limit(10).to_arrow()
-    rows = sorted(zip(hits["id"].to_pylist(), hits["_rowid"].to_pylist()))
-    copied = table.fetch_blobs("second_copy", [row_id for _, row_id in rows])
-    assert copied.to_pylist() == [b"hello", b"", None]
-
-
@pytest.mark.asyncio
 async def test_computed_column_async(tmp_path):
    db = await lancedb.connect_async(tmp_path)
@@ -7,7 +7,6 @@ import pathlib
 from typing import Optional

 import lance
-from lance.blob import BlobType as LanceBlobType
 from lancedb.conftest import MockTextEmbeddingFunction
 from lancedb.embeddings.base import EmbeddingFunctionConfig
 from lancedb.embeddings.registry import EmbeddingFunctionRegistry
@@ -908,165 +907,6 @@ def test_cast_to_target_schema():
    assert output == expected


-def test_cast_to_target_schema_coerces_binary_to_blob_v2():
-    data = pa.table({"image": pa.array([b"hello", None], type=pa.binary())})
-    target = pa.schema([lancedb.blob("image")])
-
-    output = _cast_to_target_schema(data.to_reader(), target).read_all()
-
-    image = output["image"].chunk(0)
-    assert type(image.type) is lancedb.BlobType
-    assert image.storage.to_pylist() == [
-        {"data": b"hello", "uri": None, "position": None, "size": None},
-        None,
-    ]
-
-
-def test_cast_to_target_schema_coerces_binary_to_metadata_blob_struct():
-    storage = lancedb.blob("image").type.storage_type
-    target = pa.schema(
-        [
-            pa.field(
-                "image",
-                storage,
-                metadata={
-                    b"ARROW:extension:name": b"lance.blob.v2",
-                    b"ARROW:extension:metadata": b"",
-                },
-            )
-        ]
-    )
-    data = pa.table({"image": pa.array([b"hello", None], type=pa.binary())})
-
-    output = _cast_to_target_schema(data.to_reader(), target).read_all()
-
-    image = output["image"].chunk(0)
-    assert not isinstance(image.type, pa.ExtensionType)
-    assert image.to_pylist() == [
-        {"data": b"hello", "uri": None, "position": None, "size": None},
-        None,
-    ]
-
-
-def test_cast_to_target_schema_coerces_nested_binary_blob():
-    data = pa.table(
-        {
-            "info": pa.array(
-                [{"blob": b"hello"}, {"blob": None}],
-                type=pa.struct([pa.field("blob", pa.binary())]),
-            )
-        }
-    )
-    target = pa.schema([pa.field("info", pa.struct([lancedb.blob("blob")]))])
-
-    output = _cast_to_target_schema(data.to_reader(), target).read_all()
-
-    blob = output["info"].chunk(0).field("blob")
-    assert type(blob.type) is lancedb.BlobType
-    assert blob.storage.to_pylist() == [
-        {"data": b"hello", "uri": None, "position": None, "size": None},
-        None,
-    ]
-
-
-def test_cast_to_target_schema_coerces_list_binary_blob_with_inferred_child_name():
-    data = pa.table(
-        {"images": pa.array([[b"a", b"b"], None], type=pa.list_(pa.binary()))}
-    )
-    target = pa.schema([pa.field("images", pa.list_(lancedb.blob("image")))])
-
-    output = _cast_to_target_schema(data.to_reader(), target).read_all()
-
-    images = output["images"].chunk(0)
-    assert images.type.value_field.name == "image"
-    assert type(images.type.value_type) is lancedb.BlobType
-    assert images.to_pylist()[1] is None
-    assert images.values.storage.to_pylist() == [
-        {"data": b"a", "uri": None, "position": None, "size": None},
-        {"data": b"b", "uri": None, "position": None, "size": None},
-    ]
-
-
-def test_list_blob_coercion_preserves_null_slots_with_nonzero_extent():
-    child = pa.field("image", pa.binary())
-    source = pa.ListArray.from_arrays(
-        pa.array([0, 2, 4], type=pa.int32()),
-        pa.array([b"a", b"b", b"dead", b"beef"], type=pa.binary()),
-        mask=pa.array([False, True]),
-    ).cast(pa.list_(child))
-    target = pa.schema([pa.field("images", pa.list_(lancedb.blob("image")))])
-
-    output = _cast_to_target_schema(
-        pa.table({"images": source}).to_reader(), target
-    ).read_all()
-
-    images = output["images"].chunk(0)
-    assert images.to_pylist()[1] is None
-    assert [b["data"] for b in images.to_pylist()[0]] == [b"a", b"b"]
-
-
-def test_fixed_size_list_blob_coercion_keeps_null_rows():
-    child = pa.field("frame", pa.binary())
-    source = (
-        pa.FixedSizeListArray.from_arrays(
-            pa.array([b"a", b"b", b"c", b"d"], type=pa.binary()), 2
-        )
-        .take(pa.array([0, None], type=pa.int32()))
-        .cast(pa.list_(child, 2))
-    )
-    target = pa.schema([pa.field("frames", pa.list_(lancedb.blob("frame"), 2))])
-
-    output = _cast_to_target_schema(
-        pa.table({"frames": source}).to_reader(), target
-    ).read_all()
-
-    frames = output["frames"].chunk(0)
-    assert frames.to_pylist()[1] is None
-    assert [b["data"] for b in frames.to_pylist()[0]] == [b"a", b"b"]
-
-
-def test_cast_to_target_schema_accepts_pylance_blob_v2():
-    target_type = lancedb.BlobType()
-    source = lance.blob_array([b"hello", None])
-    assert type(source.type) is LanceBlobType
-    assert type(source.type) is type(target_type)
-    data = pa.table({"image": source})
-    target = pa.schema([pa.field("image", target_type)])
-
-    output = _cast_to_target_schema(data.to_reader(), target).read_all()
-
-    image = output["image"].chunk(0)
-    assert type(image.type) is LanceBlobType
-    assert image.type == target_type
-    assert image.storage.to_pylist() == [
-        {"data": b"hello", "uri": None, "position": None, "size": None},
-        None,
-    ]
-
-
-def test_cast_to_target_schema_rejects_different_blob_v2_class():
-    class OtherBlobType(pa.ExtensionType):
-        def __init__(self):
-            super().__init__(lancedb.BlobType().storage_type, "lance.blob.v2")
-
-        def __arrow_ext_serialize__(self) -> bytes:
-            return b""
-
-        @classmethod
-        def __arrow_ext_deserialize__(
-            cls, storage_type: pa.DataType, serialized: bytes
-        ) -> "OtherBlobType":
-            return cls()
-
-    storage = lance.blob_array([b"hello"]).storage
-    source = pa.ExtensionArray.from_storage(OtherBlobType(), storage)
-    data = pa.table({"image": source})
-    target = pa.schema([lancedb.blob("image")])
-
-    with pytest.raises(pa.ArrowTypeError, match="different extension type"):
-        _cast_to_target_schema(data.to_reader(), target).read_all()
-
-
 def test_sanitize_data_stream():
    # Make sure we don't collect the whole stream when running sanitize_data
    schema = pa.schema({"a": pa.int32()})
@@ -629,17 +629,6 @@ impl Connection {
        })
    }

-    pub fn drop_function(
-        self_: PyRef<'_, Self>,
-        name: String,
-        version: String,
-    ) -> PyResult<Bound<'_, PyAny>> {
-        let inner = self_.get_inner()?.clone();
-        future_into_py(self_.py(), async move {
-            inner.drop_function(name, version).await.infer_error()
-        })
-    }
-
    pub fn list_jobs(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
        let inner = self_.get_inner()?.clone();
        future_into_py(self_.py(), async move {
@@ -323,7 +323,6 @@ impl<'py> IntoPyObject<'py> for PyQueryVectors {
 pub struct PyQueryRequest {
    pub limit: Option<usize>,
    pub offset: Option<usize>,
-    pub take_offsets: Option<Vec<u64>>,
    pub filter: Option<PyQueryFilter>,
    pub full_text_search: Option<PyLanceDB<FtsQuery>>,
    pub select: PySelect,
@@ -354,7 +353,6 @@ impl From<AnyQuery> for PyQueryRequest {
            AnyQuery::Query(query_request) => Self {
                limit: query_request.limit,
                offset: query_request.offset,
-                take_offsets: query_request.take_offsets,
                filter: query_request.filter.map(PyQueryFilter),
                full_text_search: query_request
                    .full_text_search
@@ -383,7 +381,6 @@ impl From<AnyQuery> for PyQueryRequest {
            AnyQuery::VectorQuery(vector_query) => Self {
                limit: vector_query.base.limit,
                offset: vector_query.base.offset,
-                take_offsets: vector_query.base.take_offsets,
                filter: vector_query.base.filter.map(PyQueryFilter),
                full_text_search: None,
                select_source_columns: PySelect::source_columns(&vector_query.base.select),
@@ -1,6 +1,6 @@
 [package]
 name = "lancedb"
-version = "0.39.0-beta.0"
+version = "0.38.0-beta.11"
 edition.workspace = true
 description = "LanceDB: A serverless, low-latency vector database for AI applications"
 license.workspace = true
@@ -523,21 +523,6 @@ impl Connection {
            .await
    }

-    /// Drop one exact immutable Function version from the remote catalog.
-    ///
-    /// Returns `true` when the server appended a Dropped transition and
-    /// `false` for an idempotent replay. Local databases return
-    /// [`Error::NotSupported`].
-    pub async fn drop_function(
-        &self,
-        name: impl AsRef<str>,
-        version: impl AsRef<str>,
-    ) -> Result<bool> {
-        self.internal
-            .drop_function(name.as_ref(), version.as_ref())
-            .await
-    }
-
    /// Rename a table in the database.
    ///
    /// This is only supported in LanceDB Cloud.
@@ -307,10 +307,6 @@ pub trait Database:
    ) -> Result<crate::function::FunctionVersion> {
        function_catalog_not_supported()
    }
-    /// Drop one exact immutable Function version from the remote catalog.
-    async fn drop_function(&self, _name: &str, _version: &str) -> Result<bool> {
-        function_catalog_not_supported()
-    }
    /// A [`crate::job::Job`] handle for a server-side job by id, suitable for
    /// waiting on or cancelling the job. The handle is constructed without a
    /// server round trip; an unknown id surfaces when the handle is used.
@@ -13,7 +13,7 @@ use lance::dataset::{ReadParams, WriteMode, builder::DatasetBuilder};
 use lance::io::{ObjectStore, ObjectStoreParams, WrappingObjectStore};
 use lance_datafusion::utils::StreamingWriteSource;
 use lance_file::version::LanceFileVersion;
-use lance_io::object_store::{ReadDirOptions, StorageOptionsAccessor, StorageOptionsProvider};
+use lance_io::object_store::{StorageOptionsAccessor, StorageOptionsProvider};
 use lance_table::io::commit::commit_handler_from_url;
 use object_store::local::LocalFileSystem;
 use snafu::ResultExt;
@@ -281,22 +281,6 @@ impl std::fmt::Display for ListingDatabase {
 }

 const LANCE_EXTENSION: &str = "lance";
-
-/// The table a listed child of the database names, or `None` if the child is not a table.
-///
-/// A table is the directory `<name>.lance`; a loose file or any other directory under the
-/// database prefix belongs to something else. `dir_suffix` is `.lance`, built once by the
-/// caller rather than per child.
-/// The table a listed child directory holds, or `None` if it is not a table at all.
-///
-/// Only directories are considered, so a loose object named like a table is not one.
-fn table_name(location: &object_store::path::Path, dir_suffix: &str) -> Option<String> {
-    location
-        .filename()?
-        .strip_suffix(dir_suffix)
-        .map(String::from)
-        .filter(|name| !name.is_empty())
-}
 const ENGINE: &str = "engine";
 const MIRRORED_STORE: &str = "mirroredStore";

@@ -960,72 +944,51 @@ impl Database for ListingDatabase {
        Ok(f)
    }

-    /// List the tables in the database, a page at a time.
-    ///
-    /// The page_token is opaque, unlike the `start_after` parameter of [`Self::table_names()`].
-    ///
-    /// When there are no more results, the returned page_token will be None.
-    ///
-    /// `limit` is the maximum number of tables to return in the response. But it is possible
-    /// for the response to contain fewer than `limit` tables, even when there are more tables
-    /// to return. Clients should check the returned page_token to determine if there are
-    /// more results, rather than relying on the number of tables returned.
-    ///
-    /// The order that results are returned in not guaranteed to be stable across calls,
-    /// so clients should not rely on it.
    async fn list_tables(&self, request: ListTablesRequest) -> Result<ListTablesResponse> {
        if request.id.as_ref().map(|v| !v.is_empty()).unwrap_or(false) {
            return self.namespace_database().list_tables(request).await;
        }
-        let limit = request.limit.map(|limit| limit.max(0) as usize);
-        let dir_suffix = format!(".{LANCE_EXTENSION}");
-        let mut tables = Vec::new();
-        let mut page_token = request.page_token.filter(|token| !token.is_empty());
+        let mut f = self
+            .object_store
+            .read_dir(self.base_path.clone())
+            .await?
+            .iter()
+            .map(Path::new)
+            .filter(|path| {
+                let is_lance = path
+                    .extension()
+                    .and_then(|e| e.to_str())
+                    .map(|e| e == LANCE_EXTENSION);
+                is_lance.unwrap_or(false)
+            })
+            .filter_map(|p| p.file_stem().and_then(|s| s.to_str().map(String::from)))
+            .collect::<Vec<String>>();
+        f.sort();

-        // A page of nothing: the store rejects a limit of zero, and no table was handed over
-        // for a token to resume after.
-        if limit == Some(0) {
-            return Ok(ListTablesResponse {
-                context: None,
-                tables,
-                page_token: None,
-            });
+        // Handle pagination with page_token
+        if let Some(ref page_token) = request.page_token {
+            let index = f
+                .iter()
+                .position(|name| name.as_str() > page_token.as_str())
+                .unwrap_or(f.len());
+            f.drain(0..index);
        }

-        loop {
-            // Ask only for what the page still has room for, so a database holding more
-            // than one page costs one request per page rather than one per table.
-            let listing = self
-                .object_store
-                .read_dir_page(
-                    self.base_path.clone(),
-                    ReadDirOptions {
-                        page_token: page_token.take(),
-                        limit: limit.map(|limit| limit - tables.len()),
-                    },
-                )
-                .await?;
-            page_token = listing.page_token;
-            // Only child directories can be tables, and the store already separates them
-            // out, so the objects in the page are not looked at.
-            tables.extend(
-                listing
-                    .result
-                    .common_prefixes
-                    .iter()
-                    .filter_map(|location| table_name(location, &dir_suffix)),
-            );
-            // Children that are not tables leave the page short of the limit, so keep
-            // going until the page is full or the database runs out.
-            if page_token.is_none() || limit.is_none_or(|limit| tables.len() >= limit) {
-                break;
+        // Determine if there's a next page. The token is the last name of this page,
+        // not the first of the next one: the next page resumes strictly after the
+        // token, so naming the next page's first entry would skip it.
+        let next_page_token = match request.limit {
+            Some(limit) if f.len() > limit as usize => {
+                f.truncate(limit as usize);
+                f.last().cloned()
            }
-        }
+            _ => None,
+        };

        Ok(ListTablesResponse {
            context: None,
-            tables,
-            page_token,
+            tables: f,
+            page_token: next_page_token,
        })
    }

@@ -1521,182 +1484,6 @@ mod tests {
    use tokio::sync::Barrier;
    use tokio::time::timeout;

-    async fn create_tables(db: &ListingDatabase, names: &[&str]) {
-        let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
-        for name in names {
-            db.create_table(CreateTableRequest {
-                name: name.to_string(),
-                namespace_path: vec![],
-                data: Box::new(RecordBatch::new_empty(schema.clone())) as Box<dyn Scannable>,
-                mode: CreateTableMode::Create,
-                write_options: Default::default(),
-                location: None,
-                namespace_client: None,
-            })
-            .await
-            .unwrap();
-        }
-    }
-
-    /// Every table in the database, taken `limit` at a time, which is how a caller walks a
-    /// listing: the token ends the walk, never a short page.
-    async fn walk(db: &ListingDatabase, limit: Option<i32>) -> Vec<String> {
-        let mut seen = Vec::new();
-        let mut page_token = None;
-        loop {
-            let page = db
-                .list_tables(ListTablesRequest {
-                    limit,
-                    page_token,
-                    ..Default::default()
-                })
-                .await
-                .unwrap();
-            seen.extend(page.tables);
-            page_token = page.page_token;
-            if page_token.is_none() {
-                return seen;
-            }
-            assert!(
-                seen.len() < 100,
-                "the walk is serving tables more than once"
-            );
-        }
-    }
-
-    /// Paging with the returned token has to visit every table exactly once, whatever the
-    /// page size, with nothing lost or repeated at a boundary.
-    #[rstest::rstest]
-    #[tokio::test]
-    async fn test_list_tables_pages_over_every_table_once(#[values(1, 2, 3, 5, 10)] limit: i32) {
-        let (_tempdir, db) = setup_database().await;
-        create_tables(&db, &["a", "b", "c", "d", "e"]).await;
-
-        assert_eq!(walk(&db, Some(limit)).await, vec!["a", "b", "c", "d", "e"]);
-    }
-
-    /// The token is opaque: it is whatever resumes the store the database sits on, not a
-    /// table name. Callers hand it back and nothing else.
-    ///
-    /// Nothing validates a token, so one invented by a caller is read as a position rather
-    /// than refused — which is why the token has to come back from a previous page.
-    #[tokio::test]
-    async fn test_the_page_token_is_not_a_table_name() {
-        let (_tempdir, db) = setup_database().await;
-        create_tables(&db, &["a", "b", "c"]).await;
-
-        let page = db
-            .list_tables(ListTablesRequest {
-                limit: Some(1),
-                ..Default::default()
-            })
-            .await
-            .unwrap();
-
-        assert_eq!(page.tables, vec!["a"]);
-        let token = page.page_token.expect("two tables are still to come");
-        assert_ne!(token, "a");
-
-        // Handing it back is the only thing a caller does with it, and it resumes.
-        let rest = db
-            .list_tables(ListTablesRequest {
-                page_token: Some(token),
-                ..Default::default()
-            })
-            .await
-            .unwrap();
-        assert_eq!(rest.tables, vec!["b", "c"]);
-    }
-
-    /// A limit the listing does not fill leaves no token behind, so a caller paging by token
-    /// stops without asking for an empty page.
-    #[tokio::test]
-    async fn test_a_listing_that_runs_out_has_no_token() {
-        let (_tempdir, db) = setup_database().await;
-        create_tables(&db, &["a", "b"]).await;
-
-        let page = db
-            .list_tables(ListTablesRequest {
-                limit: Some(10),
-                ..Default::default()
-            })
-            .await
-            .unwrap();
-
-        assert_eq!(page.tables, vec!["a", "b"]);
-        assert_eq!(page.page_token, None);
-    }
-
-    /// An empty page token means "from the start", which is how a client looping on a token
-    /// spells its first request.
-    #[tokio::test]
-    async fn test_an_empty_page_token_lists_from_the_start() {
-        let (_tempdir, db) = setup_database().await;
-        create_tables(&db, &["a", "b"]).await;
-
-        let page = db
-            .list_tables(ListTablesRequest {
-                page_token: Some(String::new()),
-                ..Default::default()
-            })
-            .await
-            .unwrap();
-
-        assert_eq!(page.tables, vec!["a", "b"]);
-    }
-
-    /// Listing follows the order the object store lists directories in, so a name that
-    /// extends another comes first: the `-` of `users-archive.lance` sorts below the `.` of
-    /// `users.lance`. Pagination pushes its cursor into the list request, so it cannot report
-    /// an order other than the one it resumes in.
-    #[tokio::test]
-    async fn test_listing_order_follows_the_store_not_the_table_name() {
-        let (_tempdir, db) = setup_database().await;
-        create_tables(&db, &["users", "users-archive", "users.old"]).await;
-
-        assert_eq!(
-            walk(&db, None).await,
-            vec!["users-archive", "users", "users.old"]
-        );
-        // And paging reports the same order, so a walk sees each table once.
-        assert_eq!(
-            walk(&db, Some(1)).await,
-            vec!["users-archive", "users", "users.old"]
-        );
-    }
-
-    /// Only directories named `<name>.lance` are tables; loose files and other directories
-    /// under the database prefix are not. A page spent on them is filled from the next one,
-    /// so a page holding only non-tables does not read as an empty database.
-    #[tokio::test]
-    async fn test_listing_ignores_non_table_children() {
-        let (tempdir, db) = setup_database().await;
-        create_tables(&db, &["real"]).await;
-        std::fs::write(tempdir.path().join("aaa-loose.lance"), b"not a table").unwrap();
-        create_dir_all(tempdir.path().join("aaa-scratch")).unwrap();
-
-        let page = db
-            .list_tables(ListTablesRequest {
-                limit: Some(1),
-                ..Default::default()
-            })
-            .await
-            .unwrap();
-
-        assert_eq!(page.tables, vec!["real"]);
-    }
-
-    #[tokio::test]
-    async fn listing_ignores_empty_table_name() {
-        let (tempdir, db) = setup_database().await;
-        create_dir_all(tempdir.path().join(".lance")).unwrap();
-        let page = db.list_tables(ListTablesRequest::default()).await.unwrap();
-        assert!(
-            page.tables.is_empty(),
-            "invalid empty table name was listed"
-        );
-    }
-
    async fn setup_database() -> (tempfile::TempDir, ListingDatabase) {
        let tempdir = tempdir().unwrap();
        let uri = tempdir.path().to_str().unwrap();
@@ -539,7 +539,9 @@ impl Database for LanceNamespaceDatabase {
        self.namespace
            .drop_table(drop_request)
            .await
-            .map_err(|e| map_namespace_lance_error(e, name))?;
+            .map_err(|e| Error::Runtime {
+                message: format!("Failed to drop table: {}", e),
+            })?;

        Ok(())
    }
@@ -1493,15 +1495,6 @@ mod tests {
            .expect("Failed to list tables");
        assert!(!table_names_after.contains(&"drop_test".to_string()));

-        let error = conn
-            .drop_table("drop_test", &["test_ns".into()])
-            .await
-            .expect_err("dropping a missing table should fail");
-        assert!(
-            matches!(error, Error::TableNotFound { ref name, .. } if name == "drop_test"),
-            "expected TableNotFound, got: {error:?}"
-        );
-
        // Verify: Cannot open dropped table
        let open_result = conn.open_table("drop_test").execute().await;
        assert!(open_result.is_err());
@@ -31,7 +31,7 @@ use lance::io::RecordBatchStream;
 use lance_arrow::RecordBatchExt;
 use lance_core::ROW_ID;
 use lance_core::error::LanceOptionExt;
-use std::collections::{HashMap, HashSet};
+use std::collections::HashMap;
 use std::sync::Arc;

 /// Reads a permutation of a source table based on row IDs stored in a separate table
@@ -234,14 +234,7 @@ impl PermutationReader {
            .expect_ok()?
            .values();

-        let mut unique_row_ids = HashSet::with_capacity(num_rows);
-        let in_list: Vec<Expr> = row_ids
-            .iter()
-            .copied()
-            .filter(|row_id| unique_row_ids.insert(*row_id))
-            .map(lit)
-            .collect();
-        let num_unique_row_ids = unique_row_ids.len();
+        let in_list: Vec<Expr> = row_ids.iter().map(|id| lit(*id)).collect();

        let base_query = QueryRequest {
            filter: Some(QueryFilter::Datafusion(col(ROW_ID).in_list(in_list, false))),
@@ -254,7 +247,7 @@ impl PermutationReader {
            .query(
                &AnyQuery::Query(base_query),
                QueryExecutionOptions {
-                    max_batch_length: num_unique_row_ids as u32,
+                    max_batch_length: num_rows as u32,
                    ..Default::default()
                },
            )
@@ -269,9 +262,9 @@ impl PermutationReader {
            });
        }

-        if batches.iter().map(|b| b.num_rows()).sum::<usize>() != num_unique_row_ids {
+        if batches.iter().map(|b| b.num_rows()).sum::<usize>() != num_rows {
            return Err(Error::InvalidInput {
-                message: "Base table returned a different number of rows than the number of unique row IDs"
+                message: "Base table returned different number of rows than the number of row IDs"
                    .to_string(),
            });
        }
@@ -511,7 +504,6 @@ impl PermutationReader {
            let table = Table::from(self.base_table.clone());
            let batches = table
                .take_offsets(offsets.to_vec())
-                .preserve_order()
                .select(selection.clone())
                .execute()
                .await?
@@ -811,10 +803,10 @@ mod tests {
        .unwrap();

        // Take offsets in reverse order and verify returned rows match that order
-        let offsets = vec![5, 3, 5, 1, 0];
+        let offsets = vec![5, 3, 1, 0];
        let batch = reader.take_offsets(&offsets, Select::All).await.unwrap();

-        assert_eq!(batch.num_rows(), 5);
+        assert_eq!(batch.num_rows(), 4);

        let idx_values = batch
            .column(0)
@@ -828,52 +820,6 @@ mod tests {
        assert_eq!(idx_values, expected);
    }

-    #[tokio::test]
-    async fn test_take_offsets_preserves_repeated_rows_in_permutation() {
-        let base_table = lance_datagen::gen_batch()
-            .col("idx", lance_datagen::array::step::<Int32Type>())
-            .into_mem_table("tbl", RowCount::from(5), BatchCount::from(1))
-            .await;
-        let base_row_ids = collect_column::<UInt64Type>(&base_table, "_rowid").await;
-        let permutation_row_ids = vec![
-            base_row_ids[3],
-            base_row_ids[1],
-            base_row_ids[3],
-            base_row_ids[2],
-        ];
-        let permutation_batch = RecordBatch::try_new(
-            Arc::new(Schema::new(vec![
-                Field::new("row_id", DataType::UInt64, false),
-                Field::new(SPLIT_ID_COLUMN, DataType::UInt64, false),
-            ])),
-            vec![
-                Arc::new(UInt64Array::from(permutation_row_ids)),
-                Arc::new(UInt64Array::from(vec![0; 4])),
-            ],
-        )
-        .unwrap();
-        let permutation_table = virtual_table("row_ids", &permutation_batch).await;
-        let reader = PermutationReader::try_from_tables(
-            base_table.base_table().clone(),
-            permutation_table.base_table().clone(),
-            0,
-        )
-        .await
-        .unwrap();
-
-        let batch = reader
-            .take_offsets(&[0, 1, 2, 3], Select::All)
-            .await
-            .unwrap();
-        let idx_values = batch
-            .column(0)
-            .as_primitive::<Int32Type>()
-            .values()
-            .to_vec();
-
-        assert_eq!(idx_values, vec![3, 1, 3, 2]);
-    }
-
    #[tokio::test]
    async fn test_take_offsets_with_column_selection() {
        let (base_table, row_ids_table, row_ids) = setup_permutation_tables(10).await;
@@ -937,17 +883,17 @@ mod tests {
            .unwrap();

        // With no permutation table, take_offsets uses the base table directly
-        let offsets = vec![0, 2, 0, 4, 6];
+        let offsets = vec![0, 2, 4, 6];
        let batch = reader.take_offsets(&offsets, Select::All).await.unwrap();

-        assert_eq!(batch.num_rows(), 5);
+        assert_eq!(batch.num_rows(), 4);

        let idx_values = batch
            .column(0)
            .as_primitive::<Int32Type>()
            .values()
            .to_vec();
-        assert_eq!(idx_values, vec![0, 2, 0, 4, 6]);
+        assert_eq!(idx_values, vec![0, 2, 4, 6]);
    }

    #[tokio::test]
@@ -15,9 +15,6 @@ use serde_json::Value;

 use crate::{Error, Result};

-/// Semantic Function type for a Blob v2 value.
-pub const FUNCTION_BLOB_V2_TYPE: &str = "blob_v2";
-
 fn invalid_json(error: impl std::fmt::Display) -> Error {
    Error::InvalidInput {
        message: format!("invalid remote Function JSON: {error}"),
@@ -210,33 +207,6 @@ pub enum PythonRuntimeSpec {
        environment: PythonEnvironmentSpec,
        env: BTreeMap<String, String>,
    },
-    /// The GPU-enabled Sophon-managed Python runtime.
-    ///
-    /// # Examples
-    ///
-    /// ```
-    /// use std::collections::BTreeMap;
-    /// use lancedb::function::{PythonEnvironmentSpec, PythonRuntimeSpec};
-    ///
-    /// let runtime = PythonRuntimeSpec::PythonV2 {
-    ///     python_version: "3.12".to_string(),
-    ///     environment: PythonEnvironmentSpec {
-    ///         kind: "pip".to_string(),
-    ///         packages: vec!["cupy-cuda12x".to_string()],
-    ///         channels: Vec::new(),
-    ///         path: None,
-    ///         modules: Vec::new(),
-    ///         image: None,
-    ///     },
-    ///     env: BTreeMap::new(),
-    /// };
-    /// assert!(runtime.requires_gpu());
-    /// ```
-    PythonV2 {
-        python_version: String,
-        environment: PythonEnvironmentSpec,
-        env: BTreeMap<String, String>,
-    },
    /// A runtime kind introduced by a newer server.
    ///
    /// Unknown payload fields are intentionally not retained because the
@@ -249,27 +219,22 @@ impl PythonRuntimeSpec {
    pub fn kind(&self) -> &str {
        match self {
            Self::Python { .. } => "python",
-            Self::PythonV2 { .. } => "python_v2",
            Self::Unrecognized { kind } => kind,
        }
    }

-    /// The Python version for a known Python runtime, or `None` for an unknown kind.
+    /// The Python version for the V1 runtime, or `None` for an unknown kind.
    pub fn python_version(&self) -> Option<&str> {
        match self {
-            Self::Python { python_version, .. } | Self::PythonV2 { python_version, .. } => {
-                Some(python_version)
-            }
+            Self::Python { python_version, .. } => Some(python_version),
            Self::Unrecognized { .. } => None,
        }
    }

-    /// The Python environment for a known Python runtime, or `None` for an unknown kind.
+    /// The Python environment for the V1 runtime, or `None` for an unknown kind.
    pub fn environment(&self) -> Option<&PythonEnvironmentSpec> {
        match self {
-            Self::Python { environment, .. } | Self::PythonV2 { environment, .. } => {
-                Some(environment)
-            }
+            Self::Python { environment, .. } => Some(environment),
            Self::Unrecognized { .. } => None,
        }
    }
@@ -277,73 +242,38 @@ impl PythonRuntimeSpec {
    /// Environment variables, or `None` for an unknown kind.
    pub fn env(&self) -> Option<&BTreeMap<String, String>> {
        match self {
-            Self::Python { env, .. } | Self::PythonV2 { env, .. } => Some(env),
+            Self::Python { env, .. } => Some(env),
            Self::Unrecognized { .. } => None,
        }
    }
-
-    /// Whether the runtime requires a GPU selected by the execution platform.
-    pub fn requires_gpu(&self) -> bool {
-        matches!(self, Self::PythonV2 { .. })
-    }
 }

 #[derive(Deserialize)]
-struct PythonRuntimeV1Wire {
-    python_version: String,
-    environment: PythonEnvironmentSpec,
+struct PythonRuntimeWire {
+    kind: String,
+    #[serde(default)]
+    python_version: Option<String>,
+    #[serde(default)]
+    environment: Option<PythonEnvironmentSpec>,
    #[serde(default)]
    env: BTreeMap<String, String>,
-    #[serde(default)]
-    gpu: Option<Value>,
-}
-
-#[derive(Deserialize)]
-struct PythonRuntimeV2Wire {
-    python_version: String,
-    environment: PythonEnvironmentSpec,
-    #[serde(default)]
-    env: BTreeMap<String, String>,
-    gpu: bool,
 }

 impl<'de> Deserialize<'de> for PythonRuntimeSpec {
    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> std::result::Result<Self, D::Error> {
-        let value = Value::deserialize(deserializer)?;
-        let kind = value
-            .get("kind")
-            .ok_or_else(|| de::Error::missing_field("kind"))?
-            .as_str()
-            .ok_or_else(|| de::Error::custom("runtime.kind must be a string"))?
-            .to_string();
-        match kind.as_str() {
-            "python" => {
-                let wire: PythonRuntimeV1Wire =
-                    serde_json::from_value(value).map_err(de::Error::custom)?;
-                if wire.gpu.is_some() {
-                    return Err(de::Error::custom(
-                        "python runtime with gpu requires kind='python_v2'",
-                    ));
-                }
-                Ok(Self::Python {
-                    python_version: wire.python_version,
-                    environment: wire.environment,
-                    env: wire.env,
-                })
-            }
-            "python_v2" => {
-                let wire: PythonRuntimeV2Wire =
-                    serde_json::from_value(value).map_err(de::Error::custom)?;
-                if !wire.gpu {
-                    return Err(de::Error::custom("runtime.gpu must be true"));
-                }
-                Ok(Self::PythonV2 {
-                    python_version: wire.python_version,
-                    environment: wire.environment,
-                    env: wire.env,
-                })
-            }
-            _ => Ok(Self::Unrecognized { kind }),
+        let wire = PythonRuntimeWire::deserialize(deserializer)?;
+        if wire.kind == "python" {
+            Ok(Self::Python {
+                python_version: wire
+                    .python_version
+                    .ok_or_else(|| de::Error::missing_field("python_version"))?,
+                environment: wire
+                    .environment
+                    .ok_or_else(|| de::Error::missing_field("environment"))?,
+                env: wire.env,
+            })
+        } else {
+            Ok(Self::Unrecognized { kind: wire.kind })
        }
    }
 }
@@ -357,8 +287,6 @@ impl Serialize for PythonRuntimeSpec {
            environment: &'a PythonEnvironmentSpec,
            #[serde(skip_serializing_if = "BTreeMap::is_empty")]
            env: &'a BTreeMap<String, String>,
-            #[serde(skip_serializing_if = "Option::is_none")]
-            gpu: Option<bool>,
        }

        #[derive(Serialize)]
@@ -376,19 +304,6 @@ impl Serialize for PythonRuntimeSpec {
                python_version,
                environment,
                env,
-                gpu: None,
-            }
-            .serialize(serializer),
-            Self::PythonV2 {
-                python_version,
-                environment,
-                env,
-            } => PythonRuntimeRef {
-                kind: "python_v2",
-                python_version,
-                environment,
-                env,
-                gpu: Some(true),
            }
            .serialize(serializer),
            Self::Unrecognized { kind } => UnrecognizedRuntimeRef { kind }.serialize(serializer),
@@ -398,8 +313,8 @@ impl Serialize for PythonRuntimeSpec {

 /// Immutable Function version returned by the Enterprise catalog.
 ///
-/// The GPU execution requirement is part of this identity. CPU and memory sizing,
-/// priority, concurrency, and retry policy belong to the execution platform.
+/// Scheduling resources, priority, concurrency, and retry policy belong to
+/// the submitting Job and are not part of this identity.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct FunctionVersion {
    name: String,
@@ -674,7 +589,7 @@ impl_json!(RefreshColumnResult);

 #[cfg(test)]
 mod conda_environment_tests {
-    use super::{PythonEnvironmentSpec, PythonRuntimeSpec};
+    use super::PythonEnvironmentSpec;

    #[test]
    fn conda_channels_round_trip_and_pip_stays_bare() {
@@ -693,45 +608,4 @@ mod conda_environment_tests {
            serde_json::from_str(r#"{"kind":"pip","packages":["numpy"]}"#).unwrap();
        assert!(!serde_json::to_string(&pip).unwrap().contains("channels"));
    }
-
-    #[test]
-    fn gpu_python_runtime_marker_round_trips_and_validates() {
-        let runtime: PythonRuntimeSpec = serde_json::from_str(
-            r#"{"kind":"python_v2","python_version":"3.12","environment":{"kind":"pip"},"gpu":true}"#,
-        )
-        .unwrap();
-        assert_eq!(runtime.kind(), "python_v2");
-        assert!(runtime.requires_gpu());
-        assert_eq!(
-            super::canonical_json(&runtime).unwrap(),
-            r#"{"environment":{"kind":"pip"},"gpu":true,"kind":"python_v2","python_version":"3.12"}"#
-        );
-
-        for invalid in [
-            r#"{"kind":"python","python_version":"3.12","environment":{"kind":"pip"},"gpu":true}"#,
-            r#"{"kind":"python_v2","python_version":"3.12","environment":{"kind":"pip"}}"#,
-            r#"{"kind":"python_v2","python_version":"3.12","environment":{"kind":"pip"},"gpu":1}"#,
-            r#"{"kind":"python_v2","python_version":"3.12","environment":{"kind":"pip"},"gpu":false}"#,
-            r#"{"kind":"python_v2","python_version":"3.12","environment":{"kind":"pip"},"gpu":"true"}"#,
-            r#"{"kind":"python_v2","python_version":"3.12","environment":{"kind":"pip"},"gpu":"H100"}"#,
-        ] {
-            assert!(serde_json::from_str::<PythonRuntimeSpec>(invalid).is_err());
-        }
-    }
-
-    #[test]
-    fn unknown_runtime_discards_payload_before_known_field_validation() {
-        for encoded in [
-            r#"{"kind":"python_v3","gpu":{"model":"H100"}}"#,
-            r#"{"kind":"python_v3","resources":[]}"#,
-            r#"{"kind":"python_v3","python_version":3.15,"environment":{"kind":[]}}"#,
-        ] {
-            let runtime: PythonRuntimeSpec = serde_json::from_str(encoded).unwrap();
-            assert_eq!(runtime.kind(), "python_v3");
-            assert_eq!(
-                super::canonical_json(&runtime).unwrap(),
-                r#"{"kind":"python_v3"}"#
-            );
-        }
-    }
 }
@@ -74,15 +74,8 @@ const EMBEDDING_FUNCTIONS_META_KEY: &str = "embedding_functions";
 const COLUMN_DEFINITIONS_META_KEY: &str = "lancedb::column_definitions";

 /// Value of the definition's `kind` tag for the projected `select` form.
-/// Reserved for root-namespace sources; see [`NAMESPACED_SELECT_KIND`].
 pub const SELECT_KIND: &str = "select";

-/// The `select` form over a namespaced source: its own kind, because released
-/// readers drop unknown fields and resolve a `select` source at the root, so
-/// this routes them to the [`MaterializedViewKind::Unrecognized`] refusal
-/// instead of a wrong-table refresh.
-pub const NAMESPACED_SELECT_KIND: &str = "namespaced_select";
-
 /// Which view outputs each source column is projected to directly. A column
 /// may be projected more than once, so each carries every name the view gives
 /// it, in projection order.
@@ -102,10 +95,6 @@ pub struct ViewProjection {
 pub struct MaterializedViewDefinition {
    /// Name of the source table, in the same database as the view.
    pub source_table: String,
-    /// Namespace path holding the source table; empty is the root namespace.
-    /// A definition written before namespaced sources reads as root.
-    #[serde(default, skip_serializing_if = "Vec::is_empty")]
-    pub source_namespace: Vec<String>,
    /// The projected output columns, in view schema order.
    pub projections: Vec<ViewProjection>,
    /// SQL predicate selecting the source rows the view holds.
@@ -140,12 +129,7 @@ pub(crate) fn definition_to_metadata(definition: &MaterializedViewDefinition) ->
    let mut value = serde_json::to_value(definition).map_err(|e| Error::Runtime {
        message: format!("failed to serialize view definition: {e}"),
    })?;
-    let kind = if definition.source_namespace.is_empty() {
-        SELECT_KIND
-    } else {
-        NAMESPACED_SELECT_KIND
-    };
-    value["kind"] = serde_json::Value::String(kind.to_string());
+    value["kind"] = serde_json::Value::String(SELECT_KIND.to_string());
    Ok(value.to_string())
 }

@@ -166,21 +150,12 @@ pub fn materialized_view_kind(
        .get("kind")
        .and_then(|k| k.as_str())
        .ok_or_else(|| unreadable(&"missing kind tag"))?;
-    if kind != SELECT_KIND && kind != NAMESPACED_SELECT_KIND {
+    if kind != SELECT_KIND {
        return Ok(Some(MaterializedViewKind::Unrecognized {
            kind: kind.to_string(),
        }));
    }
-    let kind = kind.to_string();
-    let definition: MaterializedViewDefinition =
-        serde_json::from_value(value).map_err(|e| unreadable(&e))?;
-    // No correct writer produces a kind that disagrees with its namespace.
-    if (kind == SELECT_KIND) != definition.source_namespace.is_empty() {
-        return Err(unreadable(&format!(
-            "kind '{kind}' does not match its source namespace {:?}",
-            definition.source_namespace
-        )));
-    }
+    let definition = serde_json::from_value(value).map_err(|e| unreadable(&e))?;
    Ok(Some(MaterializedViewKind::Select(definition)))
 }

@@ -191,7 +166,6 @@ pub fn materialized_view_kind(
 pub(crate) fn plan(
    source_schema: SchemaRef,
    source_table: &str,
-    source_namespace: &[String],
    projections: &[(String, String)],
    filter: Option<&str>,
    limit: Option<u64>,
@@ -345,7 +319,6 @@ pub(crate) fn plan(

    let definition = MaterializedViewDefinition {
        source_table: source_table.to_string(),
-        source_namespace: source_namespace.to_vec(),
        projections: projections
            .into_iter()
            .map(|(output, expression)| ViewProjection { output, expression })
@@ -629,7 +602,7 @@ pub struct PreparedDeclaration {
    definition: MaterializedViewDefinition,
    /// The source's own database: the only place
    /// [`PreparedDeclaration::create`] will put the view, because refresh
-    /// resolves the recorded source coordinate through the view's database.
+    /// resolves the recorded source name through the view's database.
    database: Arc<dyn Database>,
 }

@@ -649,21 +622,10 @@ impl PreparedDeclaration {

    /// Create the view table and verify it, consuming the declaration.
    ///
-    /// The view goes at the root of the source's own database, where refresh
-    /// resolves the recorded source coordinate. Stable row ids are requested
-    /// at both levels and verified rather than trusted; nothing is rolled
-    /// back on failure.
+    /// The view goes in the source's own database, where refresh resolves the
+    /// recorded source name. Stable row ids are requested at both levels and
+    /// verified rather than trusted; nothing is rolled back on failure.
    pub async fn create(self, name: &str) -> Result<MaterializedView> {
-        self.create_in(&[], name).await
-    }
-
-    /// Create the view in `namespace_path`, empty for the root namespace.
-    /// Otherwise [`PreparedDeclaration::create`].
-    pub async fn create_in(
-        self,
-        namespace_path: &[String],
-        name: &str,
-    ) -> Result<MaterializedView> {
        let empty: Vec<std::result::Result<arrow_array::RecordBatch, arrow_schema::ArrowError>> =
            vec![];
        // Minted here, not at preparation: a declaration can be cloned and
@@ -678,7 +640,6 @@ impl PreparedDeclaration {
        let reader: Box<dyn arrow_array::RecordBatchReader + Send> =
            Box::new(arrow_array::RecordBatchIterator::new(empty, schema));
        let mut request = CreateTableRequest::new(name.to_string(), Box::new(reader));
-        request.namespace_path = namespace_path.to_vec();
        let write_params = request
            .write_options
            .lance_write_params
@@ -719,8 +680,8 @@ impl PreparedDeclaration {

 /// Validate a view declaration against its live source and hold what its
 /// creation needs. The declaration is canonicalized through the coordinate a
-/// refresh will resolve -- name and namespace both -- so a handle that does
-/// not resolve back to itself is rejected. Same creation-time checks as
+/// refresh will resolve, so a handle that does not resolve back to itself is
+/// rejected, as is a namespaced source. Same creation-time checks as
 /// [`Connection::create_materialized_view`].
 ///
 /// ```no_run
@@ -749,9 +710,17 @@ pub async fn prepare_declaration(
            message: "materialized views are supported only on local databases".into(),
        });
    };
-    // Refresh resolves the source at exactly this coordinate, so the
-    // definition records the namespace alongside the name.
-    let source_namespace = source.namespace().to_vec();
+    // The definition records the source by bare name; any other source
+    // form would be recorded as a name its refresh cannot resolve.
+    if !source.namespace().is_empty() {
+        return Err(Error::NotSupported {
+            message: format!(
+                "a namespaced source cannot be recorded in a view definition; \
+                 '{}' must be a root-namespace table",
+                source.name()
+            ),
+        });
+    }
    let database = source
        .database_opt()
        .ok_or_else(|| Error::InvalidInput {
@@ -765,7 +734,7 @@ pub async fn prepare_declaration(
    let resolved = database
        .open_table(OpenTableRequest {
            name: source.name().to_string(),
-            namespace_path: source_namespace.clone(),
+            namespace_path: vec![],
            index_cache_size: None,
            lance_read_params: None,
            location: None,
@@ -811,7 +780,6 @@ pub async fn prepare_declaration(
    let (definition, mut fields, lineage) = plan(
        source_schema.clone(),
        resolved.name(),
-        &source_namespace,
        projections,
        filter,
        limit,
@@ -871,9 +839,7 @@ fn ensure_local(connection: &Connection) -> Result<()> {
 pub struct CreateMaterializedViewBuilder {
    connection: Connection,
    name: String,
-    namespace: Vec<String>,
    source: String,
-    source_namespace: Vec<String>,
    projections: Vec<(String, String)>,
    filter: Option<String>,
    limit: Option<u64>,
@@ -884,28 +850,13 @@ impl CreateMaterializedViewBuilder {
        Self {
            connection,
            name,
-            namespace: Vec::new(),
            source,
-            source_namespace: Vec::new(),
            projections: Vec::new(),
            filter: None,
            limit: None,
        }
    }

-    /// The namespace to create the view in. Defaults to the root namespace.
-    pub fn namespace(mut self, namespace_path: Vec<String>) -> Self {
-        self.namespace = namespace_path;
-        self
-    }
-
-    /// The namespace holding the source table; recorded in the definition
-    /// for refresh to resolve. Defaults to the root namespace.
-    pub fn source_namespace(mut self, namespace_path: Vec<String>) -> Self {
-        self.source_namespace = namespace_path;
-        self
-    }
-
    /// The view's columns, as `(name, SQL expression)` pairs. Not calling
    /// this selects every source column, expanded at creation time.
    pub fn select(
@@ -936,12 +887,7 @@ impl CreateMaterializedViewBuilder {
    /// provenance across compaction, and cannot be enabled later.
    pub async fn execute(self) -> Result<MaterializedView> {
        ensure_local(&self.connection)?;
-        let source = self
-            .connection
-            .open_table(&self.source)
-            .namespace(self.source_namespace.clone())
-            .execute()
-            .await?;
+        let source = self.connection.open_table(&self.source).execute().await?;
        let prepared = prepare_declaration(
            &source,
            &self.projections,
@@ -949,7 +895,7 @@ impl CreateMaterializedViewBuilder {
            self.limit,
        )
        .await?;
-        prepared.create_in(&self.namespace, &self.name).await
+        prepared.create(&self.name).await
    }
 }

@@ -1206,7 +1152,6 @@ mod tests {
            view.definition(),
            &MaterializedViewDefinition {
                source_table: "people".into(),
-                source_namespace: Vec::new(),
                projections: vec![
                    ViewProjection {
                        output: "name".into(),
@@ -2138,138 +2083,33 @@ mod tests {
            .await
            .unwrap_err();
        assert!(err.to_string().contains("custom_loc"), "{err}");
-    }

-    /// A view declared over a namespaced source records that namespace, and
-    /// refresh resolves the source through it -- the coordinate round-trips.
-    #[tokio::test]
-    async fn a_namespaced_source_round_trips_through_refresh() {
-        use lance_namespace::models::CreateNamespaceRequest;
-
-        let tmp = tempfile::tempdir().unwrap();
-        let mut properties = std::collections::HashMap::new();
-        properties.insert("root".to_string(), tmp.path().to_str().unwrap().to_string());
-        let conn = crate::connect_namespace("dir", properties)
-            .execute()
-            .await
-            .unwrap();
-        conn.create_namespace(CreateNamespaceRequest {
-            id: Some(vec!["ns".into()]),
-            ..Default::default()
-        })
+        // A namespaced source cannot be recorded in the definition: the
+        // bare name refresh resolves would reach a different table or none.
+        let namespaced = crate::table::NativeTable::create(
+            "memory://ns_src",
+            "ns_src",
+            vec!["ns".to_string()],
+            Box::new(arrow_array::RecordBatchIterator::new(
+                vec![],
+                std::sync::Arc::new(arrow_schema::Schema::new(vec![arrow_schema::Field::new(
+                    "id",
+                    arrow_schema::DataType::Int32,
+                    true,
+                )])),
+            )) as Box<dyn arrow_array::RecordBatchReader + Send>,
+            None,
+            None,
+            None,
+            None,
+            std::collections::HashSet::new(),
+        )
        .await
        .unwrap();
-
-        let batch = record_batch!(
-            ("name", Utf8, ["ada", "grace", "alan"]),
-            ("age", Int32, [36, 85, 41])
-        )
-        .unwrap();
-        conn.create_table("people", batch)
-            .namespace(vec!["ns".to_string()])
-            .write_options(stable_row_ids())
-            .execute()
+        let namespaced = Table::new(std::sync::Arc::new(namespaced), conn.database().clone());
+        let err = prepare_declaration(&namespaced, &[], None, None)
            .await
-            .unwrap();
-
-        // A decoy of the same name at the root: resolving the source at the
-        // wrong namespace materializes one row here instead of three.
-        let decoy = record_batch!(("name", Utf8, ["mallory"]), ("age", Int32, [42])).unwrap();
-        conn.create_table("people", decoy)
-            .write_options(stable_row_ids())
-            .execute()
-            .await
-            .unwrap();
-
-        let view = conn
-            .create_materialized_view("adults", "people")
-            .namespace(vec!["ns".to_string()])
-            .source_namespace(vec!["ns".to_string()])
-            .select([("name", "name")])
-            .only_if("age >= 18")
-            .execute()
-            .await
-            .unwrap();
-
-        assert_eq!(view.definition().source_table, "people");
-        assert_eq!(view.definition().source_namespace, vec!["ns".to_string()]);
-        assert_eq!(view.table().namespace(), &["ns"]);
-
-        // Refresh resolves the source at the recorded namespace, not at root.
-        let result = view.refresh().execute().await.unwrap();
-        assert_eq!(result.rows_written, 3);
-    }
-
-    /// A definition stored before namespaced sources existed carries no
-    /// namespace key and must read as the root namespace.
-    #[test]
-    fn a_definition_without_a_namespace_reads_as_root() {
-        let stored =
-            r#"{"source_table":"people","projections":[{"output":"name","expression":"name"}]}"#;
-        let definition: MaterializedViewDefinition = serde_json::from_str(stored).unwrap();
-        assert!(definition.source_namespace.is_empty());
-    }
-
-    fn definition(source_namespace: Vec<String>) -> MaterializedViewDefinition {
-        MaterializedViewDefinition {
-            source_table: "people".to_string(),
-            source_namespace,
-            projections: vec![ViewProjection {
-                output: "name".to_string(),
-                expression: "name".to_string(),
-            }],
-            filter: None,
-            limit: None,
-            inputs: vec!["name".to_string()],
-        }
-    }
-
-    /// A root definition keeps the pre-namespace `select` form byte-stably;
-    /// a namespaced one moves off `select`, which sends pre-namespace readers
-    /// to the `Unrecognized` refusal instead of a root resolve.
-    #[test]
-    fn a_namespaced_definition_is_refused_by_the_pre_namespace_reader() {
-        let root = definition_to_metadata(&definition(Vec::new())).unwrap();
-        let root: serde_json::Value = serde_json::from_str(&root).unwrap();
-        assert_eq!(root["kind"], "select");
-        assert!(
-            root.get("source_namespace").is_none(),
-            "a root definition must not grow new keys: {root}"
-        );
-
-        let stored = definition_to_metadata(&definition(vec!["ns".to_string()])).unwrap();
-        let value: serde_json::Value = serde_json::from_str(&stored).unwrap();
-        // The pre-namespace discriminator is `kind == "select"`; anything
-        // else lands in its Unrecognized refusal rather than in a root open.
-        assert_eq!(value["kind"], "namespaced_select");
-
-        // The current reader round-trips the coordinate.
-        let metadata = HashMap::from([(DEFINITION_META_KEY.to_string(), stored)]);
-        match materialized_view_kind(&metadata).unwrap() {
-            Some(MaterializedViewKind::Select(read)) => {
-                assert_eq!(read.source_namespace, vec!["ns".to_string()])
-            }
-            other => panic!("expected the namespaced select form, got {other:?}"),
-        }
-    }
-
-    /// A kind that disagrees with its namespace is an error, not a view:
-    /// under `select` it is the shape old readers would resolve at the root.
-    #[test]
-    fn a_kind_namespace_mismatch_is_refused() {
-        for (kind, namespace) in [
-            (SELECT_KIND, vec!["ns".to_string()]),
-            (NAMESPACED_SELECT_KIND, Vec::new()),
-        ] {
-            let mut value = serde_json::to_value(definition(namespace)).unwrap();
-            value["kind"] = serde_json::Value::String(kind.to_string());
-            let metadata = HashMap::from([(DEFINITION_META_KEY.to_string(), value.to_string())]);
-            let err = materialized_view_kind(&metadata).unwrap_err();
-            assert!(
-                err.to_string()
-                    .contains("does not match its source namespace"),
-                "kind '{kind}': {err}"
-            );
-        }
+            .unwrap_err();
+        assert!(err.to_string().contains("namespaced source"), "{err}");
    }
 }
@@ -170,7 +170,6 @@ pub(crate) async fn execute_refresh(
    let (replanned, mut planned_fields, _renames) = super::plan(
        source_schema,
        &definition.source_table,
-        &definition.source_namespace,
        &projections,
        definition.filter.as_deref(),
        definition.limit,
@@ -591,7 +590,7 @@ async fn open_source(view: &Table, definition: &MaterializedViewDefinition) -> R
    let source = database
        .open_table(OpenTableRequest {
            name: definition.source_table.clone(),
-            namespace_path: definition.source_namespace.clone(),
+            namespace_path: Vec::new(),
            index_cache_size: None,
            lance_read_params: None,
            location: None,
@@ -2920,7 +2919,6 @@ mod tests {

        let replacement = crate::materialized_view::MaterializedViewDefinition {
            source_table: "src".into(),
-            source_namespace: Vec::new(),
            projections: vec![
                crate::materialized_view::ViewProjection {
                    output: "x".into(),
@@ -2960,7 +2958,6 @@ mod tests {

        let narrower = crate::materialized_view::MaterializedViewDefinition {
            source_table: "src".into(),
-            source_namespace: Vec::new(),
            projections: vec![crate::materialized_view::ViewProjection {
                output: "x".into(),
                expression: "x".into(),
@@ -1,37 +1,21 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright The LanceDB Authors

-use std::collections::{HashMap, HashSet};
-use std::pin::Pin;
 use std::sync::Arc;
 use std::{future::Future, time::Duration};

 use arrow::compute::concat_batches;
-use arrow_array::{
-    Array, Float16Array, Float32Array, Float64Array, RecordBatch, UInt64Array,
-    cast::AsArray,
-    make_array,
-    types::{Int64Type, UInt64Type},
-};
+use arrow_array::{Array, Float16Array, Float32Array, Float64Array, RecordBatch, make_array};
 use arrow_schema::{DataType, SchemaRef};
-use datafusion_common::{DataFusionError, Result as DataFusionResult};
-use datafusion_execution::TaskContext;
 use datafusion_expr::{Expr, col, lit};
-use datafusion_physical_expr::{EquivalenceProperties, Partitioning};
-use datafusion_physical_plan::{
-    DisplayAs, DisplayFormatType, ExecutionPlan, ExecutionPlanProperties, PlanProperties,
-    coalesce_partitions::CoalescePartitionsExec,
-    execution_plan::{Boundedness, EmissionType},
-    limit::GlobalLimitExec,
-    stream::RecordBatchStreamAdapter,
-};
-use futures::{FutureExt, StreamExt, TryFutureExt, TryStreamExt, stream, try_join};
+use datafusion_physical_plan::ExecutionPlan;
+use futures::{FutureExt, TryFutureExt, TryStreamExt, stream, try_join};
 use half::f16;
 /// Re-export Lance ColumnOrdering type for use in query ordering
 pub use lance::dataset::scanner::ColumnOrdering;
 use lance::dataset::{ROW_ID, scanner::DatasetRecordBatchStream};
 use lance_arrow::RecordBatchExt;
-use lance_datafusion::exec::{execute_plan, format_plan as format_analyzed_plan};
+use lance_datafusion::exec::execute_plan;
 use lance_index::scalar::FullTextSearchQuery;
 use lance_index::scalar::inverted::SCORE_COL;
 use lance_index::vector::DIST_COL;
@@ -841,14 +825,6 @@ pub struct QueryRequest {
    /// Offset of the query.
    pub offset: Option<usize>,

-    /// Dataset offsets whose occurrence multiplicity must be restored after
-    /// executing the physical lookup represented by this request.
-    ///
-    /// This is client-side execution metadata used when a [`TakeQuery`] is
-    /// converted into a request. It is not sent to remote services.
-    #[doc(hidden)]
-    pub take_offsets: Option<Vec<u64>>,
-
    /// Apply filter to the returned rows.
    pub filter: Option<QueryFilter>,

@@ -917,7 +893,6 @@ impl Default for QueryRequest {
        Self {
            limit: None,
            offset: None,
-            take_offsets: None,
            filter: None,
            filter_error: None,
            full_text_search: None,
@@ -1554,302 +1529,6 @@ impl HasQuery for VectorQuery {
    }
 }

-fn take_occurrences(offsets: &[u64]) -> HashMap<u64, usize> {
-    let mut occurrences = HashMap::with_capacity(offsets.len());
-    for offset in offsets {
-        *occurrences.entry(*offset).or_insert(0) += 1;
-    }
-    occurrences
-}
-
-fn restore_take_batch_with_occurrences(
-    batch: RecordBatch,
-    offsets: &[u64],
-    occurrences: &HashMap<u64, usize>,
-    ordering_column: &str,
-    drop_ordering_column: bool,
-    preserve_order: bool,
-) -> Result<RecordBatch> {
-    let actual_offsets = batch
-        .column_by_name(ordering_column)
-        .ok_or_else(|| Error::Schema {
-            message: format!(
-                "take query result did not include ordering column '{ordering_column}'"
-            ),
-        })?;
-    let actual_offsets = match actual_offsets.data_type() {
-        DataType::UInt64 => actual_offsets
-            .as_primitive::<UInt64Type>()
-            .values()
-            .to_vec(),
-        DataType::Int64 => actual_offsets
-            .as_primitive::<Int64Type>()
-            .values()
-            .iter()
-            .map(|offset| {
-                u64::try_from(*offset).map_err(|_| Error::Schema {
-                    message: format!(
-                        "take query ordering column '{ordering_column}' contained a negative offset"
-                    ),
-                })
-            })
-            .collect::<Result<Vec<_>>>()?,
-        data_type => {
-            return Err(Error::Schema {
-                message: format!(
-                    "take query ordering column '{ordering_column}' had unsupported type {data_type}"
-                ),
-            });
-        }
-    };
-
-    let mut desired_order = Vec::with_capacity(offsets.len());
-    if preserve_order {
-        let ordering = actual_offsets
-            .iter()
-            .copied()
-            .enumerate()
-            .map(|(index, offset)| (offset, index as u64))
-            .collect::<HashMap<_, _>>();
-        // Missing offsets retain the filter-based behavior of returning no row.
-        desired_order.extend(
-            offsets
-                .iter()
-                .filter_map(|offset| ordering.get(offset).copied()),
-        );
-    } else {
-        // Public take queries do not guarantee output order. Preserve the lookup's
-        // existing order and only restore the multiplicity of each matching row.
-        for (index, offset) in actual_offsets.iter().enumerate() {
-            if let Some(count) = occurrences.get(offset) {
-                desired_order.extend(std::iter::repeat_n(index as u64, *count));
-            }
-        }
-    }
-
-    let mut ordered_batch = if desired_order.len() == batch.num_rows()
-        && desired_order
-            .iter()
-            .enumerate()
-            .all(|(index, desired)| *desired == index as u64)
-    {
-        batch
-    } else {
-        arrow_select::take::take_record_batch(&batch, &UInt64Array::from(desired_order))?
-    };
-
-    if drop_ordering_column {
-        ordered_batch = ordered_batch.drop_column(ordering_column)?;
-    }
-
-    Ok(ordered_batch)
-}
-
-#[cfg(test)]
-fn restore_take_batch(
-    batch: RecordBatch,
-    offsets: &[u64],
-    ordering_column: &str,
-    drop_ordering_column: bool,
-    preserve_order: bool,
-) -> Result<RecordBatch> {
-    restore_take_batch_with_occurrences(
-        batch,
-        offsets,
-        &take_occurrences(offsets),
-        ordering_column,
-        drop_ordering_column,
-        preserve_order,
-    )
-}
-
-/// Restores the logical offset occurrence sequence above the physical lookup plan.
-///
-/// The lookup plan returns each matching row at most once. For ordinary unordered
-/// takes this operator expands each input batch incrementally and preserves the
-/// lookup's partitioning. The explicitly ordered reader path collects one coalesced
-/// input before restoring requested order. Pagination must remain above this operator
-/// so it applies to occurrences.
-#[derive(Debug)]
-struct TakeRestoreExec {
-    input: Arc<dyn ExecutionPlan>,
-    offsets: Vec<u64>,
-    occurrences: Arc<HashMap<u64, usize>>,
-    ordering_column: String,
-    drop_ordering_column: bool,
-    preserve_order: bool,
-    schema: SchemaRef,
-    properties: Arc<PlanProperties>,
-}
-
-impl TakeRestoreExec {
-    fn try_new(
-        input: Arc<dyn ExecutionPlan>,
-        offsets: Vec<u64>,
-        ordering_column: String,
-        drop_ordering_column: bool,
-        preserve_order: bool,
-    ) -> Result<Self> {
-        let schema = if drop_ordering_column {
-            RecordBatch::new_empty(input.schema())
-                .drop_column(&ordering_column)?
-                .schema()
-        } else {
-            input.schema()
-        };
-        let partition_count = if preserve_order {
-            1
-        } else {
-            input.output_partitioning().partition_count()
-        };
-        let emission_type = if preserve_order {
-            EmissionType::Final
-        } else {
-            EmissionType::Incremental
-        };
-        let properties = Arc::new(PlanProperties::new(
-            EquivalenceProperties::new(schema.clone()),
-            Partitioning::UnknownPartitioning(partition_count),
-            emission_type,
-            Boundedness::Bounded,
-        ));
-
-        Ok(Self {
-            input,
-            occurrences: Arc::new(take_occurrences(&offsets)),
-            offsets,
-            ordering_column,
-            drop_ordering_column,
-            preserve_order,
-            schema,
-            properties,
-        })
-    }
-}
-
-impl DisplayAs for TakeRestoreExec {
-    fn fmt_as(
-        &self,
-        _display_type: DisplayFormatType,
-        formatter: &mut std::fmt::Formatter<'_>,
-    ) -> std::fmt::Result {
-        write!(
-            formatter,
-            "TakeRestoreExec: occurrences={}",
-            self.offsets.len()
-        )
-    }
-}
-
-impl ExecutionPlan for TakeRestoreExec {
-    fn name(&self) -> &str {
-        "TakeRestoreExec"
-    }
-
-    fn properties(&self) -> &Arc<PlanProperties> {
-        &self.properties
-    }
-
-    fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
-        vec![&self.input]
-    }
-
-    fn maintains_input_order(&self) -> Vec<bool> {
-        vec![!self.preserve_order]
-    }
-
-    fn benefits_from_input_partitioning(&self) -> Vec<bool> {
-        vec![false]
-    }
-
-    fn with_new_children(
-        self: Arc<Self>,
-        children: Vec<Arc<dyn ExecutionPlan>>,
-    ) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
-        if children.len() != 1 {
-            return Err(DataFusionError::Internal(format!(
-                "TakeRestoreExec expected one child, got {}",
-                children.len()
-            )));
-        }
-        let child = children.into_iter().next().unwrap();
-        let plan = Self::try_new(
-            child,
-            self.offsets.clone(),
-            self.ordering_column.clone(),
-            self.drop_ordering_column,
-            self.preserve_order,
-        )
-        .map_err(|error| DataFusionError::External(Box::new(error)))?;
-        Ok(Arc::new(plan))
-    }
-
-    fn execute(
-        &self,
-        partition: usize,
-        context: Arc<TaskContext>,
-    ) -> DataFusionResult<datafusion_physical_plan::SendableRecordBatchStream> {
-        let partition_count = self.input.output_partitioning().partition_count();
-        if partition >= partition_count || (self.preserve_order && partition != 0) {
-            return Err(DataFusionError::Internal(format!(
-                "TakeRestoreExec cannot execute partition {partition}; input has {partition_count} partitions"
-            )));
-        }
-
-        let input = self.input.execute(partition, context)?;
-        let output_schema = self.schema.clone();
-        let offsets = self.offsets.clone();
-        let occurrences = self.occurrences.clone();
-        let ordering_column = self.ordering_column.clone();
-        let drop_ordering_column = self.drop_ordering_column;
-        let preserve_order = self.preserve_order;
-        let stream: Pin<Box<dyn futures::Stream<Item = DataFusionResult<RecordBatch>> + Send>> =
-            if preserve_order {
-                let input_schema = input.schema();
-                Box::pin(stream::once(async move {
-                    let batches = input.try_collect::<Vec<_>>().await?;
-                    let batch = if batches.is_empty() {
-                        RecordBatch::new_empty(input_schema.clone())
-                    } else {
-                        concat_batches(&input_schema, &batches)?
-                    };
-                    restore_take_batch_with_occurrences(
-                        batch,
-                        &offsets,
-                        &occurrences,
-                        &ordering_column,
-                        drop_ordering_column,
-                        true,
-                    )
-                    .map_err(|error| DataFusionError::External(Box::new(error)))
-                }))
-            } else {
-                Box::pin(input.map(move |batch| {
-                    batch.and_then(|batch| {
-                        restore_take_batch_with_occurrences(
-                            batch,
-                            &offsets,
-                            &occurrences,
-                            &ordering_column,
-                            drop_ordering_column,
-                            false,
-                        )
-                        .map_err(|error| DataFusionError::External(Box::new(error)))
-                    })
-                }))
-            };
-
-        Ok(Box::pin(RecordBatchStreamAdapter::new(
-            output_schema,
-            stream,
-        )))
-    }
-
-    fn supports_limit_pushdown(&self) -> bool {
-        false
-    }
-}
-
 /// A builder for LanceDB take queries.
 ///
 /// See [`crate::Table::query`] for more details on queries
@@ -1866,8 +1545,6 @@ impl ExecutionPlan for TakeRestoreExec {
 pub struct TakeQuery {
    parent: Arc<dyn BaseTable>,
    request: QueryRequest,
-    offsets: Option<Vec<u64>>,
-    preserve_order: bool,
 }

 impl TakeQuery {
@@ -1875,24 +1552,15 @@ impl TakeQuery {
    ///
    /// See [`crate::Table::take_offsets`] for more details.
    pub fn from_offsets(parent: Arc<dyn BaseTable>, offsets: Vec<u64>) -> Self {
-        let mut seen = HashSet::with_capacity(offsets.len());
-        let in_list: Vec<Expr> = offsets
-            .iter()
-            .copied()
-            .filter(|offset| seen.insert(*offset))
-            .map(lit)
-            .collect();
+        let in_list: Vec<Expr> = offsets.iter().map(|o| lit(*o)).collect();
        Self {
            parent,
            request: QueryRequest {
                filter: Some(QueryFilter::Datafusion(
                    col("_rowoffset").in_list(in_list, false),
                )),
-                take_offsets: Some(offsets.clone()),
                ..Default::default()
            },
-            offsets: Some(offsets),
-            preserve_order: false,
        }
    }

@@ -1907,181 +1575,9 @@ impl TakeQuery {
                filter: Some(QueryFilter::Datafusion(col(ROW_ID).in_list(in_list, false))),
                ..Default::default()
            },
-            offsets: None,
-            preserve_order: false,
        }
    }

-    /// Preserve the requested offset order when restoring duplicate occurrences.
-    ///
-    /// This is reserved for readers whose API explicitly guarantees ordering.
-    pub(crate) fn preserve_order(mut self) -> Self {
-        debug_assert!(self.offsets.is_some());
-        self.preserve_order = true;
-        self
-    }
-
-    async fn request_with_row_offset(
-        parent: &dyn BaseTable,
-        request: &QueryRequest,
-    ) -> Result<(QueryRequest, String, bool)> {
-        const ROW_OFFSET: &str = "_rowoffset";
-        const INTERNAL_ROW_OFFSET: &str = "__lancedb_take_row_offset";
-
-        let mut request = request.clone();
-        // The physical lookup must not recursively restore occurrences. The
-        // wrapper above this request owns that logical operation.
-        request.take_offsets = None;
-        let (ordering_column, drop_ordering_column) = match &mut request.select {
-            Select::All => {
-                let mut columns = parent
-                    .schema()
-                    .await?
-                    .fields()
-                    .iter()
-                    .map(|field| field.name().clone())
-                    .collect::<Vec<_>>();
-                columns.push(ROW_OFFSET.to_string());
-                request.select = Select::Columns(columns);
-                (ROW_OFFSET.to_string(), true)
-            }
-            Select::Columns(columns) => {
-                if columns.iter().any(|column| column == ROW_OFFSET) {
-                    (ROW_OFFSET.to_string(), false)
-                } else {
-                    columns.push(ROW_OFFSET.to_string());
-                    (ROW_OFFSET.to_string(), true)
-                }
-            }
-            Select::Dynamic(columns) => {
-                let mut ordering_column = INTERNAL_ROW_OFFSET.to_string();
-                while columns.iter().any(|(name, _)| name == &ordering_column) {
-                    ordering_column.push('_');
-                }
-                columns.push((ordering_column.clone(), ROW_OFFSET.to_string()));
-                (ordering_column, true)
-            }
-            Select::Expr(columns) => {
-                let mut ordering_column = INTERNAL_ROW_OFFSET.to_string();
-                while columns.iter().any(|(name, _)| name == &ordering_column) {
-                    ordering_column.push('_');
-                }
-                columns.push((ordering_column.clone(), col(ROW_OFFSET)));
-                (ordering_column, true)
-            }
-        };
-
-        Ok((request, ordering_column, drop_ordering_column))
-    }
-
-    async fn prepare_offsets_lookup(
-        parent: &dyn BaseTable,
-        request: &QueryRequest,
-    ) -> Result<(QueryRequest, String, bool, usize, Option<usize>)> {
-        let (mut request, ordering_column, drop_ordering_column) =
-            Self::request_with_row_offset(parent, request).await?;
-        // The lookup operates on distinct physical rows. Pagination is a logical
-        // operation over occurrences and must be applied only after restoration.
-        let output_offset = request.offset.take().unwrap_or_default();
-        let output_limit = request.limit.take();
-
-        Ok((
-            request,
-            ordering_column,
-            drop_ordering_column,
-            output_offset,
-            output_limit,
-        ))
-    }
-
-    fn wrap_offsets_plan(
-        lookup: Arc<dyn ExecutionPlan>,
-        offsets: &[u64],
-        ordering_column: String,
-        drop_ordering_column: bool,
-        output_offset: usize,
-        output_limit: Option<usize>,
-        preserve_order: bool,
-    ) -> Result<Arc<dyn ExecutionPlan>> {
-        let lookup = if preserve_order {
-            Arc::new(CoalescePartitionsExec::new(lookup)) as Arc<dyn ExecutionPlan>
-        } else {
-            lookup
-        };
-        let restored: Arc<dyn ExecutionPlan> = Arc::new(TakeRestoreExec::try_new(
-            lookup,
-            offsets.to_vec(),
-            ordering_column,
-            drop_ordering_column,
-            preserve_order,
-        )?);
-
-        if output_offset > 0 || output_limit.is_some() {
-            Ok(Arc::new(GlobalLimitExec::new(
-                restored,
-                output_offset,
-                output_limit,
-            )))
-        } else {
-            Ok(restored)
-        }
-    }
-
-    fn wrap_offsets_explanation(
-        lookup: &str,
-        occurrence_count: usize,
-        output_offset: usize,
-        output_limit: Option<usize>,
-        preserve_order: bool,
-    ) -> String {
-        fn indent(plan: &str, spaces: usize) -> String {
-            let indentation = " ".repeat(spaces);
-            plan.lines()
-                .map(|line| format!("{indentation}{line}"))
-                .collect::<Vec<_>>()
-                .join("\n")
-        }
-
-        let restored = if preserve_order {
-            format!(
-                "TakeRestoreExec: occurrences={occurrence_count}\n  CoalescePartitionsExec\n{}",
-                indent(lookup, 4)
-            )
-        } else {
-            format!(
-                "TakeRestoreExec: occurrences={occurrence_count}\n{}",
-                indent(lookup, 2)
-            )
-        };
-
-        if output_offset > 0 || output_limit.is_some() {
-            let fetch = output_limit
-                .map(|limit| limit.to_string())
-                .unwrap_or_else(|| "None".to_string());
-            format!(
-                "GlobalLimitExec: skip={output_offset}, fetch={fetch}\n{}",
-                indent(&restored, 2)
-            )
-        } else {
-            restored
-        }
-    }
-
-    async fn create_offsets_plan(
-        &self,
-        offsets: &[u64],
-        options: QueryExecutionOptions,
-    ) -> Result<Arc<dyn ExecutionPlan>> {
-        create_take_offsets_plan(
-            self.parent.as_ref(),
-            &self.request,
-            offsets,
-            options,
-            self.preserve_order,
-        )
-        .await
-    }
-
    /// Convert the `TakeQuery` into a `QueryRequest`.
    pub fn into_request(self) -> QueryRequest {
        self.request
@@ -2126,63 +1622,6 @@ impl TakeQuery {
    }
 }

-pub(crate) async fn create_take_offsets_plan(
-    parent: &dyn BaseTable,
-    request: &QueryRequest,
-    offsets: &[u64],
-    options: QueryExecutionOptions,
-    preserve_order: bool,
-) -> Result<Arc<dyn ExecutionPlan>> {
-    let (request, ordering_column, drop_ordering_column, output_offset, output_limit) =
-        TakeQuery::prepare_offsets_lookup(parent, request).await?;
-    let lookup_options = if preserve_order {
-        options.without_output_batch_length_limit()
-    } else {
-        options
-    };
-    let lookup = parent
-        .create_plan(&AnyQuery::Query(request), lookup_options)
-        .await?;
-
-    TakeQuery::wrap_offsets_plan(
-        lookup,
-        offsets,
-        ordering_column,
-        drop_ordering_column,
-        output_offset,
-        output_limit,
-        preserve_order,
-    )
-}
-
-pub(crate) async fn explain_take_offsets_plan(
-    parent: &dyn BaseTable,
-    request: &QueryRequest,
-    offsets: &[u64],
-    verbose: bool,
-) -> Result<String> {
-    let (request, _, _, output_offset, output_limit) =
-        TakeQuery::prepare_offsets_lookup(parent, request).await?;
-    let lookup = parent
-        .explain_plan(&AnyQuery::Query(request), verbose)
-        .await?;
-    Ok(TakeQuery::wrap_offsets_explanation(
-        &lookup,
-        offsets.len(),
-        output_offset,
-        output_limit,
-        false,
-    ))
-}
-
-pub(crate) async fn prepare_take_offsets_request(
-    parent: &dyn BaseTable,
-    request: &QueryRequest,
-) -> Result<QueryRequest> {
-    let (request, _, _, _, _) = TakeQuery::prepare_offsets_lookup(parent, request).await?;
-    Ok(request)
-}
-
 impl HasQuery for TakeQuery {
    fn mut_query(&mut self) -> &mut QueryRequest {
        &mut self.request
@@ -2191,10 +1630,6 @@ impl HasQuery for TakeQuery {

 impl ExecutableQuery for TakeQuery {
    async fn create_plan(&self, options: QueryExecutionOptions) -> Result<Arc<dyn ExecutionPlan>> {
-        if let Some(offsets) = &self.offsets {
-            return self.create_offsets_plan(offsets, options).await;
-        }
-
        let req = AnyQuery::Query(self.request.clone());
        self.parent.clone().create_plan(&req, options).await
    }
@@ -2203,18 +1638,6 @@ impl ExecutableQuery for TakeQuery {
        &self,
        options: QueryExecutionOptions,
    ) -> Result<SendableRecordBatchStream> {
-        if self.offsets.is_some() {
-            let plan = self.create_plan(options.clone()).await?;
-            let inner = execute_plan(plan, Default::default())?;
-            let inner = MaxBatchLengthStream::new_boxed(inner, options.max_batch_length as usize);
-            let inner = if let Some(timeout) = options.timeout {
-                TimeoutStream::new_boxed(inner, timeout)
-            } else {
-                inner
-            };
-            return Ok(DatasetRecordBatchStream::new(inner).into());
-        }
-
        let query = AnyQuery::Query(self.request.clone());
        Ok(SendableRecordBatchStream::from(
            self.parent.clone().query(&query, options).await?,
@@ -2222,51 +1645,11 @@ impl ExecutableQuery for TakeQuery {
    }

    async fn explain_plan(&self, verbose: bool) -> Result<String> {
-        if let Some(offsets) = &self.offsets {
-            let (request, _, _, output_offset, output_limit) =
-                Self::prepare_offsets_lookup(self.parent.as_ref(), &self.request).await?;
-            // Ask the backend to explain only the distinct-row lookup. This keeps
-            // remote explanation non-executing while still showing the client-side
-            // operators that create_plan and execution place above that lookup.
-            let lookup = self
-                .parent
-                .explain_plan(&AnyQuery::Query(request), verbose)
-                .await?;
-            return Ok(Self::wrap_offsets_explanation(
-                &lookup,
-                offsets.len(),
-                output_offset,
-                output_limit,
-                self.preserve_order,
-            ));
-        }
-
        let query = AnyQuery::Query(self.request.clone());
        self.parent.explain_plan(&query, verbose).await
    }

    async fn analyze_plan_with_options(&self, options: QueryExecutionOptions) -> Result<String> {
-        if self.offsets.is_some() {
-            if self.parent.analyze_plan_is_remote() {
-                let (request, _, _, _, _) =
-                    Self::prepare_offsets_lookup(self.parent.as_ref(), &self.request).await?;
-                // Remote analysis is owned by the service. The current wire
-                // request represents only the distinct-row lookup, so return
-                // the service report unchanged instead of fabricating metrics
-                // for client-side restoration operators.
-                return self
-                    .parent
-                    .analyze_plan(&AnyQuery::Query(request), options)
-                    .await;
-            }
-
-            let plan = self.create_plan(options).await?;
-            execute_plan(plan.clone(), Default::default())?
-                .try_collect::<Vec<_>>()
-                .await?;
-            return Ok(format_analyzed_plan(plan));
-        }
-
        let query = AnyQuery::Query(self.request.clone());
        self.parent.analyze_plan(&query, options).await
    }
@@ -2287,7 +1670,6 @@ mod tests {
        StringArray, cast::AsArray, types::Float32Type,
    };
    use arrow_schema::{DataType, Field as ArrowField, Schema as ArrowSchema};
-    use datafusion_physical_plan::display::DisplayableExecutionPlan;
    use futures::{StreamExt, TryStreamExt};
    use lance_testing::datagen::{BatchGenerator, IncrementingInt32, RandomVector};
    use rand::seq::IndexedRandom;
@@ -3542,218 +2924,6 @@ mod tests {
        assert_eq!(results[0].num_columns(), 1);
    }

-    #[tokio::test]
-    async fn test_take_offsets_preserves_duplicate_multiplicity() {
-        let tmp_dir = tempdir().unwrap();
-        let table = make_test_table(&tmp_dir).await;
-
-        let results = table
-            .take_offsets(vec![5, 1, 5, 17])
-            .select(Select::Columns(vec!["id".to_string()]))
-            .execute_with_options(QueryExecutionOptions {
-                max_batch_length: 2,
-                ..Default::default()
-            })
-            .await
-            .unwrap()
-            .try_collect::<Vec<_>>()
-            .await
-            .unwrap();
-
-        assert_eq!(results.len(), 2);
-        assert!(results.iter().all(|batch| batch.num_columns() == 1));
-        let mut ids = results
-            .iter()
-            .flat_map(|batch| {
-                batch
-                    .column_by_name("id")
-                    .unwrap()
-                    .as_primitive::<Int32Type>()
-                    .values()
-                    .to_vec()
-            })
-            .collect::<Vec<_>>();
-        ids.sort_unstable();
-        assert_eq!(ids, vec![1, 5, 5, 17]);
-    }
-
-    #[tokio::test]
-    async fn test_take_offsets_plan_is_incremental() {
-        let tmp_dir = tempdir().unwrap();
-        let table = make_test_table(&tmp_dir).await;
-
-        let plan = table
-            .take_offsets(vec![5, 1, 17])
-            .create_plan(QueryExecutionOptions {
-                max_batch_length: 1,
-                ..Default::default()
-            })
-            .await
-            .unwrap();
-
-        assert_eq!(plan.properties().emission_type, EmissionType::Incremental);
-        let displayed = DisplayableExecutionPlan::new(plan.as_ref())
-            .indent(false)
-            .to_string();
-        assert!(displayed.contains("TakeRestoreExec"));
-        assert!(!displayed.contains("CoalescePartitionsExec"));
-    }
-
-    #[tokio::test]
-    async fn test_take_into_request_preserves_duplicate_multiplicity() {
-        let tmp_dir = tempdir().unwrap();
-        let table = make_test_table(&tmp_dir).await;
-        let request = table.take_offsets(vec![5, 5]).into_request();
-        assert_eq!(request.take_offsets, Some(vec![5, 5]));
-
-        let batches = table
-            .base_table()
-            .query(&AnyQuery::Query(request), QueryExecutionOptions::default())
-            .await
-            .unwrap()
-            .try_collect::<Vec<_>>()
-            .await
-            .unwrap();
-
-        assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 2);
-    }
-
-    #[test]
-    fn test_restore_take_batch_only_reorders_when_requested() {
-        let batch = RecordBatch::try_from_iter([
-            (
-                "id",
-                Arc::new(Int32Array::from(vec![17, 5, 1])) as Arc<dyn Array>,
-            ),
-            (
-                "_rowoffset",
-                Arc::new(UInt64Array::from(vec![17, 5, 1])) as Arc<dyn Array>,
-            ),
-        ])
-        .unwrap();
-
-        let restored =
-            restore_take_batch(batch.clone(), &[5, 1, 5, 17], "_rowoffset", true, false).unwrap();
-        assert_eq!(
-            restored
-                .column_by_name("id")
-                .unwrap()
-                .as_primitive::<Int32Type>()
-                .values(),
-            &[17, 5, 5, 1]
-        );
-
-        let ordered = restore_take_batch(batch, &[5, 1, 5, 17], "_rowoffset", true, true).unwrap();
-        assert_eq!(
-            ordered
-                .column_by_name("id")
-                .unwrap()
-                .as_primitive::<Int32Type>()
-                .values(),
-            &[5, 1, 5, 17]
-        );
-    }
-
-    #[tokio::test]
-    async fn test_take_offsets_applies_pagination_after_restoration() {
-        let tmp_dir = tempdir().unwrap();
-        let table = make_test_table(&tmp_dir).await;
-
-        let limited = table
-            .take_offsets(vec![0, 1, 0, 2])
-            .select(Select::Columns(vec!["id".to_string()]))
-            .limit(3)
-            .execute()
-            .await
-            .unwrap()
-            .try_collect::<Vec<_>>()
-            .await
-            .unwrap();
-        let limited = concat_batches(&limited[0].schema(), &limited).unwrap();
-        assert_eq!(limited.num_rows(), 3);
-        assert!(
-            limited
-                .column_by_name("id")
-                .unwrap()
-                .as_primitive::<Int32Type>()
-                .values()
-                .iter()
-                .all(|id| [0, 1, 2].contains(id))
-        );
-
-        let offset = table
-            .take_offsets(vec![5, 1, 5, 17])
-            .select(Select::Columns(vec!["id".to_string()]))
-            .offset(1)
-            .execute()
-            .await
-            .unwrap()
-            .try_collect::<Vec<_>>()
-            .await
-            .unwrap();
-        let offset = concat_batches(&offset[0].schema(), &offset).unwrap();
-        assert_eq!(offset.num_rows(), 3);
-        assert!(
-            offset
-                .column_by_name("id")
-                .unwrap()
-                .as_primitive::<Int32Type>()
-                .values()
-                .iter()
-                .all(|id| [1, 5, 17].contains(id))
-        );
-    }
-
-    #[tokio::test]
-    async fn test_take_offsets_create_plan_restores_occurrences() {
-        let tmp_dir = tempdir().unwrap();
-        let table = make_test_table(&tmp_dir).await;
-        let take = table
-            .take_offsets(vec![5, 1, 5, 17])
-            .select(Select::Columns(vec!["id".to_string()]));
-
-        let plan = take
-            .create_plan(QueryExecutionOptions::default())
-            .await
-            .unwrap();
-        assert_eq!(plan.schema().fields().len(), 1);
-        assert_eq!(plan.schema().field(0).name(), "id");
-        let planned = execute_plan(plan, Default::default())
-            .unwrap()
-            .try_collect::<Vec<_>>()
-            .await
-            .unwrap();
-        let planned = concat_batches(&planned[0].schema(), &planned).unwrap();
-        let mut ids = planned
-            .column_by_name("id")
-            .unwrap()
-            .as_primitive::<Int32Type>()
-            .values()
-            .to_vec();
-        ids.sort_unstable();
-        assert_eq!(ids, vec![1, 5, 5, 17]);
-    }
-
-    #[tokio::test]
-    async fn test_take_offsets_plan_introspection_shows_restoration() {
-        let tmp_dir = tempdir().unwrap();
-        let table = make_test_table(&tmp_dir).await;
-        let take = table
-            .take_offsets(vec![0, 1, 0, 2])
-            .select(Select::Columns(vec!["id".to_string()]))
-            .limit(3);
-
-        let explained = take.explain_plan(false).await.unwrap();
-        assert!(explained.contains("GlobalLimitExec"));
-        assert!(explained.contains("TakeRestoreExec"));
-        assert!(!explained.contains("CoalescePartitionsExec"));
-
-        let analyzed = take.analyze_plan().await.unwrap();
-        assert!(analyzed.contains("GlobalLimitExec"));
-        assert!(analyzed.contains("TakeRestoreExec"));
-        assert!(!analyzed.contains("CoalescePartitionsExec"));
-    }
-
    #[tokio::test]
    async fn test_take_row_ids() {
        let tmp_dir = tempdir().unwrap();
@@ -533,11 +533,6 @@ struct RemoteListJobsResponse {
    page_token: Option<String>,
 }

-#[derive(serde::Deserialize)]
-struct RemoteDropFunctionResponse {
-    dropped: bool,
-}
-
 /// Bound on `list_jobs` page walking; a warning is logged when the listing
 /// is truncated at this many pages.
 const MAX_LIST_JOBS_PAGES: usize = 100;
@@ -588,20 +583,6 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
        response.json().await.err_to_http(request_id)
    }

-    async fn drop_function(&self, name: &str, version: &str) -> Result<bool> {
-        let req = self
-            .client
-            .post("/v1/functions/drop")
-            .json(&serde_json::json!({
-                "name": name,
-                "version": version,
-            }));
-        let (request_id, response) = self.client.send(req).await?;
-        let response = self.client.check_response(&request_id, response).await?;
-        let response: RemoteDropFunctionResponse = response.json().await.err_to_http(request_id)?;
-        Ok(response.dropped)
-    }
-
    fn job(&self, job_id: &str) -> Result<crate::job::Job> {
        Ok(crate::job::Job::new(Box::new(super::job::RemoteJob::new(
            self.client.clone(),
@@ -2708,25 +2689,6 @@ mod tests {
        assert_eq!(version.version(), "fv_01K3EXACT");
    }

-    #[tokio::test]
-    async fn test_drop_function_sends_exact_version_and_decodes_replay() {
-        let conn = Connection::new_with_handler(|request| {
-            assert_eq!(request.method(), &reqwest::Method::POST);
-            assert_eq!(request.url().path(), "/v1/functions/drop");
-            let body: serde_json::Value =
-                serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()).unwrap();
-            assert_eq!(
-                body,
-                serde_json::json!({"name": "embed", "version": "fv_01K3EXACT"})
-            );
-            http::Response::builder()
-                .status(200)
-                .body(r#"{"dropped":false}"#)
-                .unwrap()
-        });
-        assert!(!conn.drop_function("embed", "fv_01K3EXACT").await.unwrap());
-    }
-
    #[tokio::test]
    async fn test_conn_job_waits_to_done() {
        let polls = Arc::new(AtomicUsize::new(0));
@@ -40,8 +40,8 @@ use crate::table::{
 use crate::table::{AnyQuery, Filter, Predicate, PreprocessingOutput, TableStatistics};
 use crate::utils::background_cache::BackgroundCache;
 use crate::utils::{
-    MaxBatchLengthStream, TimeoutStream, resolve_arrow_field_path, resolve_arrow_fts_field_path,
-    supported_btree_data_type, supported_vector_data_type,
+    resolve_arrow_field_path, resolve_arrow_fts_field_path, supported_btree_data_type,
+    supported_vector_data_type,
 };
 use crate::{DistanceType, Error};
 use crate::{
@@ -2022,9 +2022,6 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
    fn as_any(&self) -> &dyn std::any::Any {
        self
    }
-    fn analyze_plan_is_remote(&self) -> bool {
-        true
-    }
    fn name(&self) -> &str {
        &self.name
    }
@@ -2597,13 +2594,6 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
        query: &AnyQuery,
        options: QueryExecutionOptions,
    ) -> Result<Arc<dyn ExecutionPlan>> {
-        if let AnyQuery::Query(request) = query
-            && let Some(offsets) = &request.take_offsets
-        {
-            return crate::query::create_take_offsets_plan(self, request, offsets, options, false)
-                .await;
-        }
-
        let streams = self.execute_query(query, &options).await?;
        if streams.len() == 1 {
            let stream = streams.into_iter().next().unwrap();
@@ -2622,27 +2612,6 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
        query: &AnyQuery,
        options: QueryExecutionOptions,
    ) -> Result<DatasetRecordBatchStream> {
-        if let AnyQuery::Query(request) = query
-            && let Some(offsets) = &request.take_offsets
-        {
-            let plan = crate::query::create_take_offsets_plan(
-                self,
-                request,
-                offsets,
-                options.clone(),
-                false,
-            )
-            .await?;
-            let inner = execute_plan(plan, Default::default())?;
-            let inner = MaxBatchLengthStream::new_boxed(inner, options.max_batch_length as usize);
-            let inner = if let Some(timeout) = options.timeout {
-                TimeoutStream::new_boxed(inner, timeout)
-            } else {
-                inner
-            };
-            return Ok(DatasetRecordBatchStream::new(inner));
-        }
-
        let streams = self.execute_query(query, &options).await?;

        if streams.len() == 1 {
@@ -2680,12 +2649,6 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
    }

    async fn explain_plan(&self, query: &AnyQuery, verbose: bool) -> Result<String> {
-        if let AnyQuery::Query(request) = query
-            && let Some(offsets) = &request.take_offsets
-        {
-            return crate::query::explain_take_offsets_plan(self, request, offsets, verbose).await;
-        }
-
        let base_request = self
            .client
            .post(&format!("/v1/table/{}/explain_plan/", self.identifier));
@@ -2738,17 +2701,6 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
        query: &AnyQuery,
        options: QueryExecutionOptions,
    ) -> Result<String> {
-        let prepared_query = if let AnyQuery::Query(request) = query
-            && request.take_offsets.is_some()
-        {
-            Some(AnyQuery::Query(
-                crate::query::prepare_take_offsets_request(self, request).await?,
-            ))
-        } else {
-            None
-        };
-        let query = prepared_query.as_ref().unwrap_or(query);
-
        let mut request = self
            .client
            .post(&format!("/v1/table/{}/analyze_plan/", self.identifier));
@@ -3228,8 +3180,8 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
            self.schema().await?.as_ref(),
            "schema evolution",
        )?;
-        // The server plans the declaration against its table schema, including
-        // Blob v2 semantics inherited by a direct field projection.
+        // The server plans the declaration: expression validation, type
+        // inference and the persisted binding all happen there.
        let entries = columns
            .iter()
            .map(
@@ -3738,7 +3690,7 @@ mod tests {
    };
    use arrow_schema::{DataType, Field, Schema};
    use chrono::{DateTime, Utc};
-    use futures::{StreamExt, TryFutureExt, TryStreamExt, future::BoxFuture};
+    use futures::{StreamExt, TryFutureExt, future::BoxFuture};
    use lance_index::scalar::inverted::{DocumentGranularity, query::MatchQuery};
    use lance_index::scalar::{FullTextSearchQuery, InvertedIndexParams};
    use reqwest::Body;
@@ -5659,114 +5611,6 @@ mod tests {
        assert_eq!(result, "analyzed plan");
    }

-    #[tokio::test]
-    async fn test_take_offsets_explain_plan_does_not_execute_query() {
-        let table = Table::new_with_handler("my_table", |request| {
-            assert_eq!(request.method(), "POST");
-            assert_eq!(request.url().path(), "/v1/table/my_table/explain_plan/");
-
-            http::Response::builder()
-                .status(200)
-                .body(r#""RemoteLookupExec""#)
-                .unwrap()
-        });
-
-        let explained = table
-            .take_offsets(vec![0, 1, 0, 2])
-            .select(crate::query::Select::columns(&["id"]))
-            .limit(3)
-            .explain_plan(false)
-            .await
-            .unwrap();
-
-        assert!(explained.contains("GlobalLimitExec"));
-        assert!(explained.contains("TakeRestoreExec"));
-        assert!(!explained.contains("CoalescePartitionsExec"));
-        assert!(explained.contains("RemoteLookupExec"));
-    }
-
-    #[tokio::test]
-    async fn test_converted_take_request_restores_remote_occurrences() {
-        let table = Table::new_with_handler("my_table", |request| {
-            assert_eq!(request.method(), "POST");
-            assert_eq!(request.url().path(), "/v1/table/my_table/query/");
-
-            let body: serde_json::Value =
-                serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()).unwrap();
-            assert_eq!(body["columns"], json!(["id", "_rowoffset"]));
-
-            let data = RecordBatch::try_new(
-                Arc::new(Schema::new(vec![
-                    Field::new("id", DataType::Int32, false),
-                    Field::new("_rowoffset", DataType::UInt64, false),
-                ])),
-                vec![
-                    Arc::new(Int32Array::from(vec![5])),
-                    Arc::new(arrow_array::UInt64Array::from(vec![5])),
-                ],
-            )
-            .unwrap();
-            http::Response::builder()
-                .status(200)
-                .header(CONTENT_TYPE, ARROW_FILE_CONTENT_TYPE)
-                .body(write_ipc_file(&data))
-                .unwrap()
-        });
-
-        let request = table
-            .take_offsets(vec![5, 5])
-            .select(crate::query::Select::columns(&["id"]))
-            .into_request();
-        let batches = table
-            .base_table()
-            .query(&AnyQuery::Query(request), QueryExecutionOptions::default())
-            .await
-            .unwrap()
-            .try_collect::<Vec<_>>()
-            .await
-            .unwrap();
-
-        assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 2);
-        assert!(
-            batches
-                .iter()
-                .all(|batch| batch.schema().fields().len() == 1)
-        );
-    }
-
-    #[tokio::test]
-    async fn test_take_offsets_analyze_plan_delegates_to_remote() {
-        let table = Table::new_with_handler("my_table", |request| {
-            assert_eq!(request.method(), "POST");
-            assert_eq!(request.url().path(), "/v1/table/my_table/analyze_plan/");
-            assert_eq!(
-                request
-                    .url()
-                    .query_pairs()
-                    .find(|(key, _)| key == "distributed_metrics"),
-                Some(("distributed_metrics".into(), "per_worker".into()))
-            );
-
-            http::Response::builder()
-                .status(200)
-                .body(r#""Remote analyzed plan: worker metrics""#)
-                .unwrap()
-        });
-
-        let analyzed = table
-            .take_offsets(vec![0, 1, 0, 2])
-            .select(crate::query::Select::columns(&["id"]))
-            .limit(3)
-            .analyze_plan_with_options(QueryExecutionOptions {
-                analyze_plan_distributed_metrics: AnalyzePlanDistributedMetrics::PerWorker,
-                ..Default::default()
-            })
-            .await
-            .unwrap();
-
-        assert_eq!(analyzed, "Remote analyzed plan: worker metrics");
-    }
-
    #[tokio::test]
    async fn test_query_structured_fts() {
        let table =
@@ -7543,8 +7387,8 @@ mod tests {
        assert_eq!(result.version, if old_server { 0 } else { 43 });
    }

-    /// A declaration is sent as `{name, computed}` for the server to plan; the
-    /// client never types the expression itself.
+    /// A declaration is sent as `{name, computed}` entries for the server to
+    /// plan; the client never types the expression itself.
    #[tokio::test]
    async fn test_add_computed_columns_sends_the_expression() {
        let table = Table::new_with_handler("my_table", |request| match request.url().path() {
@@ -7620,93 +7464,6 @@ mod tests {
        assert_eq!(result.version, 8);
    }

-    #[tokio::test]
-    async fn test_add_function_column_allows_an_existing_binding() {
-        let binding = crate::function::FunctionBinding::from_json(include_str!(
-            "../../tests/fixtures/first_class_functions/v1/remote_function_binding.json"
-        ))
-        .unwrap();
-        let binding_metadata = crate::table::computed_columns::function_bindings_metadata(
-            std::slice::from_ref(&binding),
-        )
-        .unwrap();
-        let mut fields = vec![
-            Field::new("title", DataType::Utf8, true),
-            Field::new("body", DataType::Utf8, true),
-        ];
-        fields.extend(binding.outputs().iter().map(|output| {
-            let data_type = match output.arrow_type.as_str() {
-                "utf8" => DataType::Utf8,
-                "int64" => DataType::Int64,
-                other => panic!("unexpected fixture output type {other}"),
-            };
-            Field::new(&output.output_name, data_type, true).with_metadata(
-                crate::table::computed_columns::function_computed_column_metadata(
-                    binding.binding_id(),
-                    output.output_ordinal,
-                    &["title".into(), "body".into()],
-                ),
-            )
-        }));
-        let schema = Schema::new_with_metadata(
-            fields,
-            HashMap::from([(
-                crate::table::computed_columns::FUNCTION_BINDINGS_META_KEY.to_string(),
-                binding_metadata,
-            )]),
-        );
-        let table =
-            Table::new_with_handler("my_table", move |request| match request.url().path() {
-                "/v1/table/my_table/describe/" => http::Response::builder()
-                    .status(200)
-                    .body(describe_response(&schema))
-                    .unwrap(),
-                "/v1/table/my_table/add_columns/" => {
-                    let actual: serde_json::Value =
-                        serde_json::from_slice(request.body().unwrap().as_bytes().unwrap())
-                            .unwrap();
-                    assert_eq!(
-                        actual["new_columns"],
-                        serde_json::json!([
-                            {"name":"secondary_text","all_null":true},
-                            {"name":"secondary_token_count","all_null":true}
-                        ])
-                    );
-                    http::Response::builder()
-                        .status(200)
-                        .body(r#"{"version":10}"#.to_string())
-                        .unwrap()
-                }
-                path => panic!("Unexpected path: {path}"),
-            });
-        let application = crate::function::FunctionApplication::from_json(
-            r#"{
-                "function":{"name":"text_features","version":"fv_01K3TEXT"},
-                "inputs":[
-                    {"parameter":"title","kind":"column","value":{"path":"title"}},
-                    {"parameter":"body","kind":"column","value":{"path":"body"}}
-                ],
-                "output":{"kind":"named_struct","fields":[
-                    {"name":"normalized_text","arrow_type":"utf8","nullable":false},
-                    {"name":"token_count","arrow_type":"int64","nullable":false}
-                ]},
-                "columns":{
-                    "normalized_text":"secondary_text",
-                    "token_count":"secondary_token_count"
-                }
-            }"#,
-        )
-        .unwrap();
-
-        let result = table
-            .add_columns()
-            .function(application)
-            .execute()
-            .await
-            .unwrap();
-        assert_eq!(result.version, 10);
-    }
-
    #[tokio::test]
    async fn test_add_fixed_size_list_function_column_declares_the_vector_type() {
        let table = Table::new_with_handler("my_table", |request| {
@@ -595,14 +595,6 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
        query: &AnyQuery,
        options: QueryExecutionOptions,
    ) -> Result<String>;
-    /// Whether [`BaseTable::analyze_plan`] is provided by a remote service.
-    ///
-    /// Client-side query wrappers use this to preserve backend metrics and
-    /// distributed-analysis options instead of replacing them with a local plan.
-    #[doc(hidden)]
-    fn analyze_plan_is_remote(&self) -> bool {
-        false
-    }

    /// Add new records to the table.
    async fn add(&self, add: AddDataBuilder) -> Result<AddResult>;
@@ -758,8 +750,8 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
    /// Declare computed columns, each defined by a SQL expression.
    ///
    /// Where the declaration is planned depends on the backend: a local table
-    /// validates and types the expression itself, while a remote one sends the
-    /// expression for the server to plan.
+    /// validates and types the expression itself, a remote one sends the text
+    /// for the server to plan.
    async fn add_computed_columns(
        &self,
        _columns: &[(String, String)],
@@ -1660,9 +1652,9 @@ impl Table {
    /// Offsets are useful for sampling as the set of all valid offsets is easily
    /// known in advance to be [0, len(table)).
    ///
-    /// No guarantees are made regarding the order in which results are returned.
-    /// Repeated offsets produce repeated rows, which makes this method suitable for
-    /// sampling with replacement.
+    /// No guarantees are made regarding the order in which results are returned.  If you
+    /// desire an output order that matches the order of the given offsets, you will need
+    /// to add the row offset column to the output and align it yourself.
    ///
    /// Parameters
    /// ----------
@@ -5771,13 +5763,6 @@ mod tests {
        assert!(index_bytes > 0);
        assert_eq!(with_index, data_only + index_bytes);

-        // Release builds reject unstable overlay datasets unless explicitly opted in.
-        if !lance_table::feature_flags::can_read_dataset(
-            lance_table::feature_flags::FLAG_UNSTABLE_DATA_OVERLAY_FILES,
-        ) {
-            return;
-        }
-
        // Commit an overlay file supplying new `foo` values for the first three
        // rows of fragment 0. There is no high-level API that writes overlays
        // yet, so write the overlay's data file and commit the `DataOverlay`
@@ -36,14 +36,6 @@ pub(super) fn coerce_blob_expr(
    };

    let input_shape = match input_field.data_type() {
-        DataType::Null => {
-            let expr: Arc<dyn PhysicalExpr> = Arc::new(CastExpr::new(
-                input_expr,
-                table_field.data_type().clone(),
-                None,
-            ));
-            return Ok((expr, table_field.clone()));
-        }
        DataType::Binary | DataType::LargeBinary | DataType::BinaryView => BlobInputShape::Bytes,
        DataType::Utf8 | DataType::LargeUtf8 | DataType::Utf8View => BlobInputShape::String,
        DataType::Struct(children) => {
@@ -163,7 +155,7 @@ mod tests {
    use crate::blob::blob;
    use arrow_array::{
        Array, ArrayRef, BinaryArray, BinaryViewArray, Int32Array, Int64Array, LargeBinaryArray,
-        NullArray, RecordBatch, StringArray, StringViewArray, StructArray, UInt8Array, UInt64Array,
+        RecordBatch, StringArray, StringViewArray, StructArray, UInt8Array, UInt64Array,
    };
    use arrow_schema::Schema;
    use datafusion::prelude::SessionContext;
@@ -287,18 +279,6 @@ mod tests {
        assert_eq!(data.value(0), b"view");
    }

-    #[tokio::test]
-    async fn null_column_coerces_to_all_null_blob_struct() {
-        let batch = batch_with_image(
-            Field::new("image", DataType::Null, true),
-            Arc::new(NullArray::new(2)),
-        );
-        let coerced = coerce(batch, &blob_table_schema()).await;
-        let image = image_struct(&coerced);
-        assert!(image.is_null(0));
-        assert!(image.is_null(1));
-    }
-
    #[tokio::test]
    async fn binary_nulls_stay_null_after_coercion() {
        let batch = batch_with_image(
@@ -110,7 +110,7 @@ fn requires_local_namespace_execution(query: &AnyQuery) -> bool {
    // pushing these down would silently ignore the user's setting. For use_lsm that
    // is worse than a tuning miss: MemWAL read routing lives only in `create_plan`,
    // so a pushed-down query would return stale base-only data with no error.
-    if query.base().use_lsm.is_some() || query.base().take_offsets.is_some() {
+    if query.base().use_lsm.is_some() {
        return true;
    }
    matches!(
@@ -154,13 +154,6 @@ pub async fn create_plan(
    options: QueryExecutionOptions,
 ) -> Result<Arc<dyn ExecutionPlan>> {
    let query = query.canonicalized()?;
-    if let AnyQuery::Query(request) = &query
-        && let Some(offsets) = &request.take_offsets
-    {
-        return crate::query::create_take_offsets_plan(table, request, offsets, options, false)
-            .await;
-    }
-
    let query = match query {
        AnyQuery::VectorQuery(query) => query,
        AnyQuery::Query(query) => VectorQueryRequest::from_plain_query(query),
@@ -45,11 +45,7 @@ async fn local_function_catalog_operations_return_stable_not_supported() {
        .get_function("normalize_score", "fv_exact")
        .await
        .unwrap_err();
-    let drop_error = connection
-        .drop_function("normalize_score", "fv_exact")
-        .await
-        .unwrap_err();
-    for error in [create_error, lookup_error, drop_error] {
+    for error in [create_error, lookup_error] {
        assert!(matches!(
            error,
            Error::NotSupported { message }
@@ -78,12 +78,6 @@
        "type": "utf8"
      }
    },
-    {
-      "arrow_type": "large_utf8",
-      "json": {
-        "type": "large_utf8"
-      }
-    },
    {
      "arrow_type": "binary",
      "json": {
@@ -177,21 +171,6 @@
        ]
      }
    },
-    {
-      "arrow_type": "list<large_utf8>",
-      "json": {
-        "type": "list",
-        "fields": [
-          {
-            "name": "item",
-            "nullable": false,
-            "type": {
-              "type": "large_utf8"
-            }
-          }
-        ]
-      }
-    },
    {
      "arrow_type": "large_list<utf8>",
      "json": {
@@ -207,21 +186,6 @@
        ]
      }
    },
-    {
-      "arrow_type": "large_list<large_utf8>",
-      "json": {
-        "type": "large_list",
-        "fields": [
-          {
-            "name": "item",
-            "nullable": false,
-            "type": {
-              "type": "large_utf8"
-            }
-          }
-        ]
-      }
-    },
    {
      "arrow_type": "fixed_size_list<float32, 384>",
      "json": {
@@ -366,4 +330,4 @@
    "timestamp[us]",
    "struct<a: int32>"
  ]
-}
+}
Author	SHA1	Message	Date
Jack Ye	61e6614c09	chore: retain computed column support in beta.4 update	2026-08-27 17:24:32 -07:00
lancedb automation	d41ed98b52	chore: update lance dependency to v12.0.0-beta.4	2026-08-27 21:26:01 +00:00
Wyatt Alt	a10c2e39b9	feat: let a computed-column batch read its own earlier declarations `add_columns().computed()` accepted several columns in one call but bound each against the table's schema as it stood before the call, so `a` and `b = a + 1` had to be two commits. A server staging declarations behind other schema work has no atomic way to do that, and a caller reading the builder's plural signature reasonably expects the batch to be one. Each accepted column now joins the schema the next one resolves against, so the batch is planned and committed as one. Order is the dependency order; reading ahead is still an unknown column. `validate_declarations` exposes the schema-level checks -- the Function-binding guard and the planning -- without a commit, for callers that must reject before earlier work in the same request lands; LSM state is table state and stays a commit-time check. Dependent columns need refresh to be dependency-aware, or `b = coalesce(a, 0)` refreshed before `a` bakes zeros from `a`'s placeholder null and the fill-once contract keeps them. Refresh now walks the dependency graph once and fills each reachable column once in dependency order, each as its own commit, then the requested column. Every fill in the pass covers only the fragments of the snapshot the pass started from: a commit may rebase over a concurrent append, and the fragment that admits carries placeholder nulls no earlier fill covered, so it waits for a later refresh and is reported as `rows_remaining`. Two concurrent fills of one input collide on its field in lance's conflict check, so a dependent fill can only commit over inputs that were durable when it read them. `rows_filled` counts the requested column only; the async result's `published_version` is the last commit of the pass, inputs included.	2026-08-27 20:46:32 +00:00
lancedb automation	c05da95d4c	chore: update lance dependency to v12.0.0-beta.2	2026-08-26 01:31:44 +00:00