chore: update lance dependency to v12.0.0-beta.16

ci: add spell checking (#4148 )
Adds [typos](https://github.com/crate-ci/typos) as a CI check and pre-commit hook, the same way Lance does it, so misspellings like the ones fixed in #4146 get caught automatically going forward. This also fixes the misspellings `typos` found across the repo (Rust, Python, TypeScript source, comments, and generated docs), and adds a small `.typos.toml` with `extend-words` entries for terms that are correct but look like typos: `AKS` (Azure Kubernetes Service), `RabitQ` (a real quantization algorithm name), `mmaped` (the actual name of a `candle-core` API we call), and `Writeable` (from Python's `_typeshed.WriteableBuffer`). Third-party license files are excluded. Fixes #4147 Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-10 07:12:30 +00:00 · 2026-09-10 04:06:08 +00:00 · 2026-09-09 15:33:04 +08:00 · 2026-09-08 15:42:06 -07:00 · 2026-09-08 13:53:38 -07:00 · 2026-09-08 12:14:45 +00:00
146 changed files with 13859 additions and 13000 deletions
@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "0.38.0-beta.12"
+current_version = "0.39.0-beta.6"
 parse = """(?x)
    (?P<major>0|[1-9]\\d*)\\.
    (?P<minor>0|[1-9]\\d*)\\.
@@ -44,3 +44,27 @@ 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,12 +29,14 @@ jobs:
    steps:
      - uses: actions/setup-node@v6
        with:
-          node-version: "18"
+          node-version: "24"
+      - uses: pnpm/action-setup@v6
+        with:
+          version: 11.1.1
      # 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": {
@@ -43,7 +45,11 @@ jobs:
              "body-leading-blank": [0, "always"]
            }
          }' > .commitlintrc.js
-      - run: npx commitlint --extends @commitlint/config-conventional --verbose <<< $COMMIT_MSG
+      - run: >
+          pnpm dlx
+          --package @commitlint/cli@21.2.2
+          --package @commitlint/config-conventional@21.2.2
+          commitlint --extends @commitlint/config-conventional --verbose <<< $COMMIT_MSG
        env:
          COMMIT_MSG: >
            ${{ github.event.pull_request.title }}
@@ -54,7 +60,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 `npm run docs` in nodejs)
+          # API reference (the js/ tree comes from `pnpm 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,9 +55,7 @@ jobs:
      - name: Set up node
        uses: actions/setup-node@v6
        with:
-          node-version: 20
-          cache: 'npm'
-          cache-dependency-path: docs/package-lock.json
+          node-version: 24
      - name: Install node dependencies
        working-directory: nodejs
        run: |
@@ -47,9 +47,8 @@ jobs:
        version: 11.1.1
    - uses: actions/setup-node@v6
      with:
-        # 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).
+        # Build on a supported LTS; the matrix job below covers every
+        # Node version the library claims to support.
        node-version: 24
        cache: 'pnpm'
        cache-dependency-path: nodejs/pnpm-lock.yaml
@@ -84,7 +83,7 @@ jobs:
    timeout-minutes: 30
    strategy:
      matrix:
-        node-version: [ "18", "20" ]
+        node-version: [ "22", "24", "26" ]
    runs-on: "ubuntu-22.04"
    defaults:
      run:
@@ -101,9 +100,9 @@ jobs:
    - uses: actions/setup-node@v6
      name: Setup Node.js 24 for build
      with:
-        # 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.
+        # 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.
        node-version: 24
        cache: 'pnpm'
        cache-dependency-path: nodejs/pnpm-lock.yaml
@@ -152,9 +151,9 @@ jobs:
        S3_TEST: "1"
        # Newer @smithy/core uses dynamic ESM imports.
        NODE_OPTIONS: "--experimental-vm-modules"
-      # Invoke jest directly because pnpm 11 itself requires Node 22+
-      # while the matrix tests on older Node versions.
-      run: npx jest --verbose
+      # 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
    - name: Test examples
      working-directory: ./
      env:
@@ -164,7 +163,7 @@ jobs:
      run: |
        python ci/mock_openai.py &
        cd nodejs/examples
-        npx jest --testEnvironment jest-environment-node-single-context --verbose
+        node_modules/.bin/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
@@ -185,8 +184,7 @@ jobs:
        version: 11.1.1
    - uses: actions/setup-node@v6
      with:
-        # pnpm 11 requires Node >= 22.13; use 24 since 22 hits EOL
-        # in October.
+        # pnpm 11 requires Node >= 22.13.
        node-version: 24
        cache: 'pnpm'
        cache-dependency-path: nodejs/pnpm-lock.yaml
@@ -168,8 +168,7 @@ jobs:
      - name: Setup node
        uses: actions/setup-node@v6
        with:
-          # pnpm 11 requires Node >= 22.13; use 24 since 22 hits EOL
-          # in October.
+          # pnpm 11 requires Node >= 22.13.
          node-version: 24
          cache: pnpm
          cache-dependency-path: nodejs/pnpm-lock.yaml
@@ -251,7 +250,7 @@ jobs:
          run: |
            set -e
            ${{ matrix.settings.pre_build }}
-            npx napi build --platform --release \
+            node_modules/.bin/napi build --platform --release \
              --features ${{ matrix.settings.features }} \
              --target ${{ matrix.settings.target }} \
              --dts ../lancedb/native.d.ts \
@@ -271,7 +270,7 @@ jobs:
      - name: Build
        run: |
          ${{ matrix.settings.pre_build }}
-          npx napi build --platform --release \
+          node_modules/.bin/napi build --platform --release \
              --features ${{ matrix.settings.features }} \
              --target ${{ matrix.settings.target }} \
              --dts ../lancedb/native.d.ts \
@@ -339,7 +338,7 @@ jobs:
          - target: aarch64-unknown-linux-gnu
            host: ubuntu-2404-8x-arm64
        node:
-          - '20'
+          - '22'
    runs-on: ${{ matrix.settings.host }}
    defaults:
      run:
@@ -385,9 +384,9 @@ jobs:
      - name: Move built files
        run: cp dist/native.d.ts dist/native.js dist/*.node lancedb/
      - name: Test bindings
-        # Invoke jest directly because pnpm 11 itself requires Node 22+
-        # while the matrix tests on older Node versions.
-        run: npx jest --verbose
+        # 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
  publish:
    name: Publish
    runs-on: ubuntu-latest
@@ -232,7 +232,10 @@ jobs:
          ALL_FEATURES=`cargo metadata --format-version=1 --no-deps \
            | jq -r '.packages[] | .features | keys | .[]' \
            | grep -v s3-test | sort | uniq | paste -s -d "," -`
-          cargo test --profile ci --features $ALL_FEATURES --locked
+          # Run doctests before test binaries fill the runner disk. Examples are
+          # already built by the Linux job, so avoid retaining them here.
+          cargo test --profile ci --features $ALL_FEATURES --locked --doc
+          cargo test --profile ci --features $ALL_FEATURES --locked --lib --tests

  windows:
    strategy:
@@ -0,0 +1,20 @@
+name: Typo checker
+on:
+  push:
+    branches:
+      - main
+  pull_request:
+
+permissions:
+  contents: read
+
+jobs:
+  run:
+    name: Spell Check with Typos
+    runs-on: ubuntu-latest
+    steps:
+      - name: Check out code
+        uses: actions/checkout@v6
+
+      - name: Check spelling of the entire repository
+        uses: crate-ci/typos@6802cc60d4e7f78b9d5454f6cf3935c042d5e1e3 # v1.26.0
@@ -1,22 +0,0 @@
-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 }}
@@ -1,22 +0,0 @@
-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 }}
@@ -10,6 +10,10 @@ repos:
    rev: v0.9.9
    hooks:
      - id: ruff
+  - repo: https://github.com/crate-ci/typos
+    rev: v1.26.0
+    hooks:
+      - id: typos
  # - repo: https://github.com/RobertCraigie/pyright-python
  #   rev: v1.1.395
  #   hooks:
@@ -20,7 +24,10 @@ repos:
    hooks:
      - id: local-biome-check
        name: biome check
-        entry: npx @biomejs/biome@1.8.3 check --config-path nodejs/biome.json nodejs/
+        # 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/
        language: system
        types: [text]
        files: "nodejs/.*"
@@ -0,0 +1,19 @@
+[default]
+extend-ignore-re = ["(?Rm)^.*(#|//)\\s*spellchecker:disable-line$"]
+
+[default.extend-words]
+# Azure Kubernetes Service, mentioned in rust/lancedb/src/remote/oauth.rs.
+AKS = "AKS"
+# RabitQ is the name of a vector quantization algorithm, not a typo of "Rabbit".
+Rabit = "Rabit"
+# `VarBuilder::from_mmaped_safetensors` is the real (if oddly-spelled) name of
+# the candle-core API we call in rust/lancedb/src/embeddings/sentence_transformers.rs.
+mmaped = "mmaped"
+# `WriteableBuffer` is the real name of a type from Python's `_typeshed` stubs,
+# used in python/python/lancedb/_blob.py.
+Writeable = "Writeable"
+
+[files]
+extend-exclude = [
+    "*_THIRD_PARTY_LICENSES.*",
+]
@@ -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 `npm`/`pnpm` lint, format, build, and docs commands in `nodejs`.
+* TypeScript changes: run the relevant `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 `npm run build` to generate TypeScript definitions.
+2. Run `pnpm 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 `npm run docs` to generate TypeScript documentation.
+6. Run `pnpm run docs` to generate TypeScript documentation.

 ## Python API reference

@@ -332,6 +332,34 @@ dependencies = [
 "num-traits",
 ]

+[[package]]
+name = "arrow-flight"
+version = "58.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b2dbe34824c639e43136af8f106992792ab456540d54b880bc320a3192502d2e"
+dependencies = [
+ "arrow-arith",
+ "arrow-array",
+ "arrow-buffer",
+ "arrow-cast",
+ "arrow-data",
+ "arrow-ipc",
+ "arrow-ord",
+ "arrow-row",
+ "arrow-schema",
+ "arrow-select",
+ "arrow-string",
+ "base64 0.22.1",
+ "bytes",
+ "futures",
+ "once_cell",
+ "paste",
+ "prost",
+ "prost-types",
+ "tonic",
+ "tonic-prost",
+]
+
 [[package]]
 name = "arrow-ipc"
 version = "58.4.0"
@@ -535,9 +563,9 @@ dependencies = [

 [[package]]
 name = "async-trait"
-version = "0.1.91"
+version = "0.1.92"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ae36dc4177970ef04fde5178d3e2429882def40e57a451f919c098f72baa6cec"
+checksum = "82f6aeea286b8eb4dd3431a1be1b59d290ace00f5bfd8e2a159bc2a05e2c1667"
 dependencies = [
 "proc-macro2",
 "quote",
@@ -1129,7 +1157,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "edca88bc138befd0323b20752846e6587272d3b03b0343c8ea28a6f819e6e71f"
 dependencies = [
 "async-trait",
- "axum-core",
+ "axum-core 0.4.5",
 "bytes",
 "futures-util",
 "http 1.5.0",
@@ -1138,7 +1166,7 @@ dependencies = [
 "hyper 1.9.0",
 "hyper-util",
 "itoa",
- "matchit",
+ "matchit 0.7.3",
 "memchr",
 "mime",
 "percent-encoding",
@@ -1156,6 +1184,31 @@ dependencies = [
 "tracing",
 ]

+[[package]]
+name = "axum"
+version = "0.8.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90"
+dependencies = [
+ "axum-core 0.5.6",
+ "bytes",
+ "futures-util",
+ "http 1.5.0",
+ "http-body 1.1.0",
+ "http-body-util",
+ "itoa",
+ "matchit 0.8.4",
+ "memchr",
+ "mime",
+ "percent-encoding",
+ "pin-project-lite",
+ "serde_core",
+ "sync_wrapper",
+ "tower",
+ "tower-layer",
+ "tower-service",
+]
+
 [[package]]
 name = "axum-core"
 version = "0.4.5"
@@ -1177,6 +1230,24 @@ dependencies = [
 "tracing",
 ]

+[[package]]
+name = "axum-core"
+version = "0.5.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1"
+dependencies = [
+ "bytes",
+ "futures-core",
+ "http 1.5.0",
+ "http-body 1.1.0",
+ "http-body-util",
+ "mime",
+ "pin-project-lite",
+ "sync_wrapper",
+ "tower-layer",
+ "tower-service",
+]
+
 [[package]]
 name = "backoff"
 version = "0.4.0"
@@ -1443,9 +1514,9 @@ checksum = "175812e0be2bccb6abe50bb8d566126198344f707e304f45c648fd8f2cc0365e"

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

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

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

 [[package]]
 name = "lance"
-version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+version = "12.0.0-beta.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
 dependencies = [
 "arc-swap",
 "arrow",
@@ -4888,8 +4959,8 @@ dependencies = [

 [[package]]
 name = "lance-arrow"
-version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+version = "12.0.0-beta.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4911,7 +4982,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.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4925,7 +4996,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.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -4934,8 +5005,8 @@ dependencies = [

 [[package]]
 name = "lance-bitpacking"
-version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+version = "12.0.0-beta.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
 dependencies = [
 "arrayref",
 "crunchy",
@@ -4945,8 +5016,8 @@ dependencies = [

 [[package]]
 name = "lance-core"
-version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+version = "12.0.0-beta.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4983,8 +5054,8 @@ dependencies = [

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

 [[package]]
 name = "lance-datagen"
-version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+version = "12.0.0-beta.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5031,8 +5103,8 @@ dependencies = [

 [[package]]
 name = "lance-derive"
-version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+version = "12.0.0-beta.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
 dependencies = [
 "proc-macro2",
 "quote",
@@ -5041,8 +5113,8 @@ dependencies = [

 [[package]]
 name = "lance-encoding"
-version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+version = "12.0.0-beta.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
 dependencies = [
 "arrow-arith",
 "arrow-array",
@@ -5075,8 +5147,8 @@ dependencies = [

 [[package]]
 name = "lance-file"
-version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+version = "12.0.0-beta.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
 dependencies = [
 "arrow-arith",
 "arrow-array",
@@ -5107,8 +5179,8 @@ dependencies = [

 [[package]]
 name = "lance-index"
-version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+version = "12.0.0-beta.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
 dependencies = [
 "arc-swap",
 "arrow",
@@ -5172,8 +5244,8 @@ dependencies = [

 [[package]]
 name = "lance-index-core"
-version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+version = "12.0.0-beta.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -5195,8 +5267,8 @@ dependencies = [

 [[package]]
 name = "lance-io"
-version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+version = "12.0.0-beta.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5236,8 +5308,8 @@ dependencies = [

 [[package]]
 name = "lance-linalg"
-version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+version = "12.0.0-beta.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -5251,27 +5323,29 @@ dependencies = [

 [[package]]
 name = "lance-namespace"
-version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+version = "12.0.0-beta.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
 dependencies = [
 "arrow",
 "async-trait",
 "bytes",
 "lance-core",
 "lance-namespace-reqwest-client",
+ "serde",
+ "serde_json",
 "snafu 0.9.0",
 ]

 [[package]]
 name = "lance-namespace-impls"
-version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+version = "12.0.0-beta.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
 dependencies = [
 "arrow",
 "arrow-ipc",
 "arrow-schema",
 "async-trait",
- "axum",
+ "axum 0.7.9",
 "base64 0.22.1",
 "bytes",
 "chrono",
@@ -5304,9 +5378,9 @@ dependencies = [

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

 [[package]]
 name = "lance-select"
-version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+version = "12.0.0-beta.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -5333,8 +5407,8 @@ dependencies = [

 [[package]]
 name = "lance-table"
-version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+version = "12.0.0-beta.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5374,8 +5448,8 @@ dependencies = [

 [[package]]
 name = "lance-testing"
-version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+version = "12.0.0-beta.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -5388,8 +5462,8 @@ dependencies = [

 [[package]]
 name = "lance-tokenizer"
-version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+version = "12.0.0-beta.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
 dependencies = [
 "frostem",
 "icu_segmenter",
@@ -5402,7 +5476,7 @@ dependencies = [

 [[package]]
 name = "lancedb"
-version = "0.38.0-beta.12"
+version = "0.39.0-beta.6"
 dependencies = [
 "ahash",
 "anyhow",
@@ -5411,6 +5485,7 @@ dependencies = [
 "arrow-buffer",
 "arrow-cast",
 "arrow-data",
+ "arrow-flight",
 "arrow-ipc",
 "arrow-ord",
 "arrow-schema",
@@ -5466,6 +5541,7 @@ dependencies = [
 "polars",
 "polars-arrow",
 "pprof 0.14.1",
+ "prost",
 "rand 0.9.5",
 "random_word",
 "regex",
@@ -5482,6 +5558,7 @@ dependencies = [
 "test-log",
 "tokenizers",
 "tokio",
+ "tonic",
 "url",
 "urlencoding",
 "uuid",
@@ -5490,7 +5567,7 @@ dependencies = [

 [[package]]
 name = "lancedb-nodejs"
-version = "0.38.0-beta.12"
+version = "0.39.0-beta.6"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -5515,7 +5592,7 @@ dependencies = [

 [[package]]
 name = "lancedb-python"
-version = "0.38.0-beta.12"
+version = "0.39.0-beta.6"
 dependencies = [
 "arrow",
 "async-trait",
@@ -5539,6 +5616,7 @@ dependencies = [
 "serde_json",
 "snafu 0.8.9",
 "tokio",
+ "uuid",
 ]

 [[package]]
@@ -5748,9 +5826,9 @@ dependencies = [

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

 [[package]]
 name = "loom"
@@ -5859,6 +5937,12 @@ version = "0.7.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0e7465ac9959cc2b1404e8e2367b43684a6d13790fe23056cc8c6c5a6b7bcb94"

+[[package]]
+name = "matchit"
+version = "0.8.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3"
+
 [[package]]
 name = "matrixmultiply"
 version = "0.3.10"
@@ -6001,9 +6085,9 @@ dependencies = [

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

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

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

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

 [[package]]
 name = "napi-derive-backend"
-version = "6.1.1"
+version = "6.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d60b5d773ad46c698c8cc2cd9fde0b283d39cbb7f71c04bee633c7bdba4423bd"
+checksum = "df4056ac7c18e4438ccf0edaed4340ca0d269278c8ec19284f7b23cb039fd0ae"
 dependencies = [
 "convert_case",
 "proc-macro2",
@@ -7733,6 +7818,7 @@ dependencies = [
 "pyo3-build-config",
 "pyo3-ffi",
 "pyo3-macros",
+ "uuid",
 ]

 [[package]]
@@ -8601,9 +8687,9 @@ dependencies = [

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

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

 [[package]]
 name = "serde_with_macros"
-version = "3.21.0"
+version = "3.22.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "84d57bc0c8b9a17920c178daa6bb924850d54a9c97ab45194bb8c17ad66bb660"
+checksum = "8705578779c2b6bd90d84d66eb2e206b708b1a4d7b9f17641b293545bf1c7e46"
 dependencies = [
 "darling 0.23.0",
 "proc-macro2",
@@ -10084,6 +10171,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ac2a5518c70fa84342385732db33fb3f44bc4cc748936eb5833d2df34d6445ef"
 dependencies = [
 "async-trait",
+ "axum 0.8.9",
 "base64 0.22.1",
 "bytes",
 "h2 0.4.16",
@@ -10095,9 +10183,11 @@ dependencies = [
 "hyper-util",
 "percent-encoding",
 "pin-project",
+ "rustls-native-certs",
 "socket2 0.6.3",
 "sync_wrapper",
 "tokio",
+ "tokio-rustls 0.26.4",
 "tokio-stream",
 "tower",
 "tower-layer",
@@ -10452,9 +10542,9 @@ checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821"

 [[package]]
 name = "uuid"
-version = "1.24.0"
+version = "1.26.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bf3923a6f5c4c6382e0b653c4117f48d631ea17f38ed86e2a828e6f7412f5239"
+checksum = "b5772d71c9be8a8a6ac2117d949c5b224c1b72241bb611d9a3012edcf8af7812"
 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.2", default-features = false, "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-core = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-datagen = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-file = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-io = { "version" = "=12.0.0-beta.2", default-features = false, "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-index = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-linalg = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace-impls = { "version" = "=12.0.0-beta.2", default-features = false, "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-table = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-testing = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-datafusion = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-encoding = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-arrow = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
+lance = { "version" = "=12.0.0-beta.16", default-features = false, "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
+lance-core = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
+lance-datagen = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
+lance-file = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
+lance-io = { "version" = "=12.0.0-beta.16", default-features = false, "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
+lance-index = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
+lance-linalg = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace-impls = { "version" = "=12.0.0-beta.16", default-features = false, "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
+lance-table = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
+lance-testing = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
+lance-datafusion = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
+lance-encoding = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
+lance-arrow = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "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
@@ -39,6 +39,7 @@ arrow-ord = "58.0.0"
 arrow-schema = "58.0.0"
 arrow-select = "58.0.0"
 arrow-cast = "58.0.0"
+arrow-flight = { version = "58.0.0", features = ["flight-sql-experimental"] }
 async-trait = "0"
 bytes = "1"
 datafusion = { version = "54.0.0", default-features = false }
@@ -71,7 +72,8 @@ serde = "1"
 serde_json = "1"
 tempfile = "3.5.0"
 tokio = { version = "1.23", features = ["rt-multi-thread", "sync"] }
-uuid = { version = "1.7.0", features = ["v4"] }
+tonic = { version = "0.14", features = ["tls-native-roots", "tls-ring"] }
+uuid = { version = "1.7.0", features = ["v4", "v7"] }
 chrono = { version = "0.4", default-features = false, features = ["clock"] }

 [profile.ci]
@@ -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 && npx license-checker --markdown --out NODEJS_THIRD_PARTY_LICENSES.md
+	cd nodejs && pnpm dlx license-checker@25 --markdown --out NODEJS_THIRD_PARTY_LICENSES.md
 	cd java && ./mvnw license:aggregate-add-third-party -q
@@ -12,16 +12,12 @@ 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 nodejs/package-lock.json
+  git add Cargo.lock
  git commit --amend --no-edit
 else
-  git add Cargo.lock nodejs/package-lock.json
+  git add Cargo.lock
  git commit -m "Update lockfiles"
 fi
@@ -131,18 +131,13 @@ 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",
@@ -150,12 +145,7 @@ 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 = [
-    # 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" },
-]
+exceptions = []
 # 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,22 +47,24 @@ pytest -vv python/tests/docs

 ### Checking typescript examples

-The `@lancedb/lancedb` package must be built before running the tests:
+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.

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

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

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

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

 ```shell
 pushd nodejs
-npm run docs
+# `pnpm docs` would invoke pnpm's built-in `docs` command, not the script.
+pnpm run docs
 popd
 ```
@@ -155,7 +155,7 @@ paths:
                vector:
                  type: FixedSizeList
                  description: |
-                    The targetted vector to search for. Required.
+                    The targeted vector to search for. Required.
                vector_column:
                  type: string
                  description: |
@@ -446,6 +446,15 @@ paths:
              properties:
                column:
                  type: string
+                name:
+                  type: string
+                  description: Optional name for the created index.
+                replace:
+                  type: boolean
+                  default: true
+                  description: |
+                    Whether to replace an existing index with the same resolved
+                    name. Defaults to true.
                metric_type:
                  type: string
                  nullable: false
@@ -1,135 +0,0 @@
-{
-  "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
-    }
-  }
-}
@@ -1,20 +0,0 @@
-{
-  "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.38.0-beta.12</version>
+    <version>0.39.0-beta.6</version>
 </dependency>
 ```

@@ -448,26 +448,6 @@ on the returned job to know when cleanup has finished.

 ***

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

 ```ts
@@ -482,48 +462,6 @@ Return true if the connection has not been closed

 ***

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

 ```ts
@@ -648,6 +586,30 @@ A page of table names and an

 ***

+### openJob()
+
+```ts
+abstract openJob(jobId): Promise<Job>
+```
+
+Open a server-side job by id, returning a handle with its record already
+populated. Rejects when the server has no such job, the way
+[Connection.openTable](Connection.md#opentable) does for a missing table.
+
+The returned [Job](Job.md) answers for its own state, specification,
+result, failure and event history, so there is no separate
+connection-level call for any of them.
+
+#### Parameters
+
+* **jobId**: `string`
+
+#### Returns
+
+`Promise`&lt;[`Job`](Job.md)&gt;
+
+***
+
 ### openMaterializedView()

 ```ts
@@ -8,19 +8,46 @@

 A handle to an operation that may still be running.

-## Constructors
+The operation may already be complete when the handle is created.

-### new Job()
+The detail getters read what the handle last observed. Submitting an
+operation returns only a job id, so populating them eagerly would cost an
+extra round trip on every call:
+
+- [Job.refresh](Job.md#refresh) and [Job.status](Job.md#status) fetch the whole record.
+- [Job.wait](Job.md#wait) records the terminal state it establishes, but not the
+  rest of the record.
+- Everything is null until one of those runs.
+
+## Accessors
+
+### creationMs

 ```ts
-new Job(): Job
+get creationMs(): null | number
 ```

+When the job was created, in milliseconds since the epoch.
+
 #### Returns

-[`Job`](Job.md)
+`null` \| `number`

-## Accessors
+***
+
+### failure
+
+```ts
+get failure(): null | JobFailureInfo
+```
+
+Why the job failed, when it failed and the server reports a reason.
+
+#### Returns
+
+`null` \| [`JobFailureInfo`](../interfaces/JobFailureInfo.md)
+
+***

 ### id

@@ -28,8 +55,69 @@ new Job(): Job
 get id(): null | string
 ```

-Identifies the operation on the server that is running it. Operations
-that run in this process have no server id. The value is opaque.
+Identifies the operation on the server that is running it.
+
+Operations that run in this process have no server id. The value is
+opaque: parsing it or storing it to resume the job later is not supported.
+
+#### Returns
+
+`null` \| `string`
+
+***
+
+### jobType
+
+```ts
+get jobType(): null | string
+```
+
+The job's type, as the server names it. Null for an in-process job, which
+has no server-side record.
+
+#### Returns
+
+`null` \| `string`
+
+***
+
+### result
+
+```ts
+get result(): any
+```
+
+The job-type-specific terminal result. Null until the job succeeds, so a
+job that never terminates reports its progress through [Job.events](Job.md#events)
+instead.
+
+#### Returns
+
+`any`
+
+***
+
+### spec
+
+```ts
+get spec(): any
+```
+
+The job-type-specific specification it was submitted with.
+
+#### Returns
+
+`any`
+
+***
+
+### state
+
+```ts
+get state(): null | string
+```
+
+The last observed lifecycle state, without contacting the backend.

 #### Returns

@@ -51,18 +139,61 @@ Request cancellation. Cancelling a finished operation is a no-op.

 ***

+### events()
+
+```ts
+events(options?): Promise<Table<any>>
+```
+
+This job's recorded lifecycle events.
+
+Where the getters above report a terminal result only once the job reaches
+one, events are written as the job runs and outlive the workers that
+produced them. A distributed job records a `claim`/`claim_complete` pair
+per unit of work, each carrying `rows_processed`, so a job that never
+finishes still accounts for what it did.
+
+The server caps results at 1000 rows by default and 10,000 at most, and
+truncates without saying so, so pass `limit` for a job that emits an event
+per fragment. `filter` is a SQL-like expression over the `state`,
+`updated_by`, `emitted_from`, `emitted_by`, and `claim_entity` columns.
+
+#### Parameters
+
+* **options?**: [`JobEventsOptions`](../interfaces/JobEventsOptions.md)
+
+#### Returns
+
+`Promise`&lt;`Table`&lt;`any`&gt;&gt;
+
+***
+
+### refresh()
+
+```ts
+refresh(): Promise<void>
+```
+
+Ask the backend for this job's current state, and for a server-side job
+its full record, then cache it for the getters above.
+
+#### Returns
+
+`Promise`&lt;`void`&gt;
+
+***
+
 ### status()

 ```ts
 status(): Promise<string>
 ```

-The operation's current lifecycle state: "running", "finished",
-"failed", or "cancelled".
+The operation's current lifecycle state: "running", "finished", "failed",
+or "cancelled".

-A point snapshot; unlike [Job.wait](Job.md#wait) it does not block or reject
-on a terminal failure state. States a newer server reports that this
-client version does not know pass through as-is.
+A point snapshot; unlike [Job.wait](Job.md#wait) it does not block or reject on a
+terminal failure state. Also refreshes the getters above.

 #### Returns

@@ -70,6 +201,22 @@ client version does not know pass through as-is.

 ***

+### toString()
+
+```ts
+toString(): string
+```
+
+Every field the handle currently knows, one per line, with the JSON
+payloads indented -- a refresh job's spec and result are the point of
+printing it.
+
+#### Returns
+
+`string`
+
+***
+
 ### wait()

 ```ts
@@ -141,7 +141,7 @@ Currently this causes multiple copies of the row to be created
 but that behavior is subject to change.

 An optional condition may be specified.  If it is, then only
-matched rows that satisfy the condtion will be updated.  Any
+matched rows that satisfy the condition will be updated.  Any
 rows that do not satisfy the condition will be left as they
 are.  Failing to satisfy the condition does not cause a
 "matched row" to become a "not matched" row.
@@ -676,9 +676,17 @@ List all the versions of the table
 abstract mergeInsert(on): MergeInsertBuilder
 ```

+Create a [MergeInsertBuilder](MergeInsertBuilder.md), which combines new data with the
+existing table in a single transaction — inserting, updating and deleting
+rows depending on how they match.
+
 #### Parameters

 * **on**: `string` \| `string`[]
+    The column, or columns, to match source rows against target
+    rows on. Typically a key or id column. Several columns match on the
+    composite key: a source row updates a target row only when it agrees on
+    every one of them.

 #### Returns

@@ -1258,7 +1266,7 @@ value is 0")
 Note: if your condition is something like "some_id_column == 7" and
 you are updating many rows (with different ids) then you will get
 better performance with a single [`merge_insert`] call instead of
-repeatedly calilng this method.
+repeatedly calling this method.

 ##### Parameters

@@ -96,7 +96,7 @@
 - [IvfFlatOptions](interfaces/IvfFlatOptions.md)
 - [IvfPqOptions](interfaces/IvfPqOptions.md)
 - [IvfRqOptions](interfaces/IvfRqOptions.md)
- [JobDescription](interfaces/JobDescription.md)
+- [JobEventsOptions](interfaces/JobEventsOptions.md)
 - [JobFailureInfo](interfaces/JobFailureInfo.md)
 - [JobInfo](interfaces/JobInfo.md)
 - [ListNamespacesOptions](interfaces/ListNamespacesOptions.md)
@@ -118,7 +118,7 @@ Number of sub-vectors of PQ.
 This value controls how much the vector is compressed during the quantization step.
 The more sub vectors there are the less the vector is compressed.  The default is
 the dimension of the vector divided by 16.  If the dimension is not evenly divisible
-by 16 we use the dimension divded by 8.
+by 16 we use the dimension divided by 8.

 The above two cases are highly preferred.  Having 8 or 16 values per subvector allows
 us to use efficient SIMD instructions.
@@ -16,7 +16,7 @@ optional config: Index;

 Advanced index configuration

-This option allows you to specify a specfic index to create and also
+This option allows you to specify a specific index to create and also
 allows you to pass in configuration for training the index.

 See the static methods on Index for details on the various index types.
@@ -112,7 +112,7 @@ Number of sub-vectors of PQ.
 This value controls how much the vector is compressed during the quantization step.
 The more sub vectors there are the less the vector is compressed.  The default is
 the dimension of the vector divided by 16.  If the dimension is not evenly divisible
-by 16 we use the dimension divded by 8.
+by 16 we use the dimension divided by 8.

 The above two cases are highly preferred.  Having 8 or 16 values per subvector allows
 us to use efficient SIMD instructions.
@@ -1,66 +0,0 @@
-[**@lancedb/lancedb**](../README.md) • **Docs**
-
-***
-
-[@lancedb/lancedb](../globals.md) / JobDescription
-
-# Interface: JobDescription
-
-A described job from `Connection.getJob`.
-
-## Properties
-
-### creationMs
-
-```ts
-creationMs: number;
-```
-
-When the job was created, in milliseconds since the epoch.
-
-***
-
-### failure?
-
-```ts
-optional failure: JobFailureInfo;
-```
-
-Why the job failed, when the job is failed and the server reports a
-reason.
-
-***
-
-### jobId
-
-```ts
-jobId: string;
-```
-
-***
-
-### jobType
-
-```ts
-jobType: string;
-```
-
-***
-
-### specJson?
-
-```ts
-optional specJson: string;
-```
-
-The job-type-specific specification as a JSON string, when present.
-
-***
-
-### state
-
-```ts
-state: string;
-```
-
-Lifecycle state: "running", "finished", "failed", or "cancelled".
@@ -0,0 +1,29 @@
+[**@lancedb/lancedb**](../README.md) • **Docs**
+
+***
+
+[@lancedb/lancedb](../globals.md) / JobEventsOptions
+
+# Interface: JobEventsOptions
+
+Which of a job's events [Job.events](../classes/Job.md#events) returns.
+
+## Properties
+
+### filter?
+
+```ts
+optional filter: string;
+```
+
+SQL-like filter over the event columns.
+
+***
+
+### limit?
+
+```ts
+optional limit: number;
+```
+
+Maximum event rows to return, up to the server maximum of 10,000.
@@ -26,7 +26,7 @@ When the job was created, in milliseconds since the epoch.
 jobId: string;
 ```

-The job id -- what `Connection.getJob` and `Connection.cancelJob`
+The job id -- what `Connection.openJob` and `Connection.cancelJob`
 accept.

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

 ***

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

 ```ts
@@ -28,6 +28,59 @@ is also an [asynchronous API client](#connections-asynchronous).

 ::: lancedb.Session

+## Remote SQL
+
+Submit SQL against a remote LanceDB database through the connection.
+The connected database and `default_namespace_path=["public"]` are used for
+unqualified tables. Fully qualified references can still query other databases
+and namespaces available to the same deployment. `execute_query` returns a
+reader as soon as its initial result stream is available. `execute_query_async`
+returns a query handle immediately; use it to inspect progress, open a reader,
+or cancel the query. The SQL client is initialized by the first query and
+retained for the lifetime of the remote connection. Query ids are random,
+connection-scoped references rather than encoded SQL or durable resume tokens:
+
+```python
+import lancedb
+
+db = lancedb.connect(
+    "db://analytics",
+    api_key="ldb_...",
+    host_override="https://api.example.com",
+    sql_host_override="grpc+tls://sql.example.com:10026",
+)
+reader = db.execute_query(
+    """
+    SELECT events.id, accounts.name
+    FROM analytics.public.events AS events
+    JOIN users.public.accounts AS accounts ON events.user_id = accounts.id
+    """,
+    default_namespace_path=["public"],
+)
+for batch in reader:
+    print(batch.num_rows)
+
+query = db.execute_query_async("SELECT * FROM events")
+print(query.id)
+print(query.describe().status)
+for batch in query.reader():
+    print(batch.num_rows)
+
+# The async connection exposes the same lifecycle without blocking:
+# async_db = await lancedb.connect_async(
+#     "db://analytics",
+#     api_key="ldb_...",
+#     host_override="https://api.example.com",
+#     sql_host_override="grpc+tls://sql.example.com:10026",
+# )
+# reader = await async_db.execute_query("SELECT * FROM events")
+# query = await async_db.execute_query_async("SELECT * FROM events")
+# description = await async_db.describe_query(query.id)
+# async for batch in await query.reader():
+#     print(batch.num_rows)
+# await query.cancel()
+```
+
 ## Namespaces (Synchronous)

 A namespace-backed connection resolves tables through a
@@ -94,6 +147,8 @@ listing a storage directory.

 ::: lancedb.functions.OutputMapping

+::: lancedb.functions.AssignmentMapping
+
 ::: lancedb.functions.FunctionBinding

 ::: lancedb.functions.RefreshColumnResult
@@ -102,6 +157,18 @@ listing a storage directory.

 ::: lancedb.job.AsyncJob

+::: lancedb.job.JobInfo
+
+::: lancedb.job.JobDescription
+
+::: lancedb.job.JobFailureInfo
+
+::: lancedb.sql.Query
+
+::: lancedb.sql.AsyncQuery
+
+::: lancedb.sql.QueryDescription
+
 ## Materialized Views (Synchronous)

 ::: lancedb.materialized_view.MaterializedView
@@ -249,6 +316,12 @@ still work. Queries return descriptors. Call

 ::: lancedb.exceptions.MissingColumnError

+::: lancedb.exceptions.JobNotFoundError
+
+::: lancedb.exceptions.JobFailedError
+
+::: lancedb.exceptions.JobCancelledError
+
 ## Integrations

 ## Pydantic
@@ -1,17 +0,0 @@
-{
-  "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.38.0-beta.12</version>
+      <version>0.39.0-beta.6</version>
      <relativePath>../pom.xml</relativePath>
    </parent>

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

    <groupId>com.lancedb</groupId>
    <artifactId>lancedb-parent</artifactId>
-    <version>0.38.0-beta.12</version>
+    <version>0.39.0-beta.6</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.2</lance-core.version>
+        <lance-core.version>12.0.0-beta.16</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.38.0-beta.12"
+version = "0.39.0-beta.6"
 publish = false
 license.workspace = true
 description.workspace = true
@@ -48,6 +48,28 @@ 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(">= 18");
-    expect(packageJson.peerDependencies["@types/node"]).toBe(">=18");
+    expect(packageJson.engines.node).toBe(">= 22");
+    expect(packageJson.peerDependencies["@types/node"]).toBe(">=22");
    expect(packageJson.peerDependenciesMeta["@types/node"]).toEqual({
      optional: true,
    });
@@ -3,6 +3,7 @@

 import * as http from "http";
 import { RequestListener } from "http";
+import packageJson = require("../package.json");
 import {
  ClientConfig,
  Connection,
@@ -70,7 +71,13 @@ async function withMockDatabase(
  try {
    await callback(db);
  } finally {
-    server.close();
+    // `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());
+    });
  }
 }

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

        const body = JSON.stringify({ tables: [] });
@@ -932,6 +939,7 @@ describe("remote connection jobs surface", () => {
    const { tableFromArrays, tableToIPC } = await import("apache-arrow");
    const eventsTable = tableFromArrays({ state: ["created", "succeeded"] });
    const eventsBody = Buffer.from(tableToIPC(eventsTable, "stream"));
+    const queryEventsPayloads: Record<string, unknown>[] = [];

    await withMockDatabase(
      (req, res) => {
@@ -960,6 +968,16 @@ describe("remote connection jobs surface", () => {
                );
            }
          } else if (req.url === "/v1/jobs/describe") {
+            if (payload["job_id"] === "job-2") {
+              res
+                .writeHead(200, { "Content-Type": "application/json" })
+                .end(
+                  '{"job_id": "job-2", "job_type": "refresh_column", ' +
+                    '"job_state": "DONE", "creation_ms": 2000, ' +
+                    '"result": {"rows_assigned": 1000000}}',
+                );
+              return;
+            }
            if (payload["job_id"] !== "job-1") {
              res.writeHead(404).end("no such job");
              return;
@@ -981,6 +999,7 @@ describe("remote connection jobs surface", () => {
              .writeHead(200, { "Content-Type": "application/json" })
              .end('{"job_id": "job-1"}');
          } else if (req.url === "/v1/jobs/query_events") {
+            queryEventsPayloads.push(payload);
            res
              .writeHead(200, {
                "Content-Type": "application/vnd.apache.arrow.stream",
@@ -997,22 +1016,65 @@ describe("remote connection jobs surface", () => {
        expect(jobs[0].state).toEqual("running");
        expect(jobs[1].state).toEqual("finished");

-        const description = await db.getJob("job-1");
-        expect(description?.state).toEqual("failed");
-        expect(JSON.parse(description?.specJson ?? "")).toEqual({
-          column: "vec",
-        });
-        expect(description?.failure?.message).toEqual("worker died");
-        expect(await db.getJob("missing")).toBeNull();
-
        expect(await db.cancelJob("job-1")).toBe(true);
        expect(await db.cancelJob("missing")).toBe(false);

-        const history = await db.jobHistory("job-1");
-        expect(history.numRows).toEqual(2);
+        // Opening a job hands back a populated handle; a missing one rejects.
+        await expect(db.openJob("missing")).rejects.toThrow("not found");
+        const finished = await db.openJob("job-2");
+        expect(finished.state).toEqual("finished");
+        expect(finished.result).toEqual({
+          // biome-ignore lint/style/useNamingConvention: snake_case mandated by the server wire format
+          rows_assigned: 1000000,
+        });

-        const job = db.job("job-1");
+        const job = await db.openJob("job-1");
        expect(job.id).toEqual("job-1");
+
+        // openJob already populated the handle; refresh() re-reads it.
+        expect(job.state).toEqual("failed");
+        await job.refresh();
+        expect(job.state).toEqual("failed");
+        expect(job.jobType).toEqual("create_index");
+        expect(job.creationMs).toEqual(1000);
+        expect(job.spec).toEqual({ column: "vec" });
+        expect(job.result).toBeNull();
+        expect(job.failure?.message).toEqual("worker died");
+
+        // The handle reaches its own events, supplying its job id.
+        const jobEvents = await job.events({
+          limit: 500,
+          filter: "state = 'claim_complete'",
+        });
+        expect(jobEvents.numRows).toEqual(2);
+        expect(queryEventsPayloads.pop()).toEqual({
+          // biome-ignore lint/style/useNamingConvention: snake_case mandated by the server wire format
+          job_id: "job-1",
+          limit: 500,
+          filter: "state = 'claim_complete'",
+        });
+
+        // Printing lays every known field out on its own line, with the JSON
+        // payloads indented rather than crammed onto one line.
+        expect(`${job}`).toEqual(
+          [
+            "Job(",
+            '    id="job-1",',
+            '    state="failed",',
+            '    jobType="create_index",',
+            "    creationMs=1000,",
+            "    spec={",
+            '        "column": "vec"',
+            "    },",
+            "    failure={",
+            '        "phase": "execute",',
+            '        "message": "worker died",',
+            '        "retryable": true',
+            "    },",
+            ")",
+          ].join("\n"),
+        );
+
        expect(await job.status()).toEqual("failed");
        await expect(job.wait()).rejects.toThrow("worker died");
      },
@@ -281,7 +281,7 @@ describe.each([arrow15, arrow16, arrow17, arrow18])(
        numIndices: 0,
        numRows: 3,
        // Full on-disk size of the two data files, footers and metadata included.
-        totalBytes: 684,
+        totalBytes: 550,
      });

      // Index files count toward totalBytes too (only deletion files and
@@ -289,7 +289,7 @@ describe.each([arrow15, arrow16, arrow17, arrow18])(
      await table.createIndex("id", { config: Index.btree() });
      const statsWithIndex = await table.stats();
      expect(statsWithIndex.numIndices).toBe(1);
-      expect(statsWithIndex.totalBytes).toBeGreaterThan(684);
+      expect(statsWithIndex.totalBytes).toBeGreaterThan(550);
    });

    it("should overwrite data if asked", async () => {
@@ -737,11 +737,12 @@ it("should query documents with LangChain PDF metadata", async () => {

 describe("merge insert", () => {
  let tmpDir: tmp.DirResult;
+  let conn: Connection;
  let table: Table;

  beforeEach(async () => {
    tmpDir = tmp.dirSync({ unsafeCleanup: true });
-    const conn = await connect(tmpDir.name);
+    conn = await connect(tmpDir.name);

    table = await conn.createTable("some_table", [
      { a: 1, b: "a" },
@@ -779,6 +780,38 @@ describe("merge insert", () => {

    expect(result.map((row) => ({ ...row }))).toEqual(expected);
  });
+  test("upsert on a composite key", async () => {
+    const composite = await conn.createTable("composite", [
+      { shard: "a", id: 1, val: "x" },
+      { shard: "a", id: 2, val: "y" },
+      { shard: "b", id: 1, val: "z" },
+    ]);
+
+    // ("a", 1) matches an existing row and updates it. ("b", 2) agrees with an
+    // existing row on each key column separately but on neither pair, so it is
+    // an insert.
+    const mergeInsertRes = await composite
+      .mergeInsert(["shard", "id"])
+      .whenMatchedUpdateAll()
+      .whenNotMatchedInsertAll()
+      .execute([
+        { shard: "a", id: 1, val: "X" },
+        { shard: "b", id: 2, val: "W" },
+      ]);
+    expect(mergeInsertRes.numUpdatedRows).toBe(1);
+    expect(mergeInsertRes.numInsertedRows).toBe(1);
+
+    const result = (await composite.toArrow())
+      .toArray()
+      .sort((a, b) => a.shard.localeCompare(b.shard) || a.id - b.id);
+
+    expect(result.map((row) => ({ ...row }))).toEqual([
+      { shard: "a", id: 1, val: "X" },
+      { shard: "a", id: 2, val: "y" },
+      { shard: "b", id: 1, val: "z" },
+      { shard: "b", id: 2, val: "W" },
+    ]);
+  });
  test("conditional update", async () => {
    const newData = [
      { a: 2, b: "x" },
@@ -3219,7 +3252,7 @@ describe.each([arrow15, arrow16, arrow17, arrow18])(
      const db = await connect(tmpDir.name);
      const data = [
        { text: "fa", vector: [0.1, 0.2, 0.3] },
-        { text: "fo", vector: [0.4, 0.5, 0.6] },
+        { text: "fo", vector: [0.4, 0.5, 0.6] }, // spellchecker:disable-line
        { text: "fob", vector: [0.4, 0.5, 0.6] },
        { text: "focus", vector: [0.4, 0.5, 0.6] },
        { text: "foo", vector: [0.4, 0.5, 0.6] },
@@ -3244,7 +3277,7 @@ describe.each([arrow15, arrow16, arrow17, arrow18])(
      const resultSet = new Set(fuzzyResults.map((r) => r.text));
      expect(resultSet.has("foo")).toBe(true);
      expect(resultSet.has("fob")).toBe(true);
-      expect(resultSet.has("fo")).toBe(true);
+      expect(resultSet.has("fo")).toBe(true); // spellchecker:disable-line
      expect(resultSet.has("food")).toBe(true);

      const prefixResults = await table
@@ -8,7 +8,8 @@
    "//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",
-    "test": "node --experimental-vm-modules node_modules/.bin/jest --testEnvironment jest-environment-node-single-context --verbose",
+    "//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",
    "lint": "biome check *.ts && biome format *.ts",
    "lint-ci": "biome ci .",
    "lint-fix": "biome check --write *.ts && pnpm format",
@@ -600,7 +600,7 @@ function makeVector(
  }
  if (values.length === 0) {
    throw Error(
-      "makeVector requires at least one value or the type must be specfied",
+      "makeVector requires at least one value or the type must be specified",
    );
  }
  const sampleValue = values.find((val) => val !== null && val !== undefined);
@@ -858,7 +858,7 @@ async function applyEmbeddings<T>(
 * customized by the `embeddingDataType` property of the embedding function.
 *
 * If a schema is provided in `makeTableOptions` then it should include the
- * embedding columns.  If no schema is provded then embedding columns will
+ * embedding columns.  If no schema is provided then embedding columns will
 * be placed at the end of the table, after all of the input columns.
 */
 export async function convertToTable(
@@ -1,7 +1,6 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright The LanceDB Authors

-import { tableFromIPC } from "apache-arrow";
 import {
  Data,
  SchemaLike,
@@ -16,6 +15,7 @@ import {
  makeEmptyTable,
 } from "./arrow";
 import { EmbeddingFunctionConfig, getRegistry } from "./embedding/registry";
+import { Job } from "./job";
 import {
  MaterializedView,
  MaterializedViewSelect,
@@ -27,8 +27,6 @@ import type {
  CreateNamespaceResponse,
  DescribeNamespaceResponse,
  DropNamespaceResponse,
-  Job,
-  JobDescription,
  JobInfo,
  ListNamespacesResponse,
  ListTablesResponse,
@@ -557,24 +555,19 @@ export abstract class Connection {
  ): Promise<void>;

  /**
-   * A {@link Job} handle for a server-side job by id.
+   * Open a server-side job by id, returning a handle with its record already
+   * populated. Rejects when the server has no such job, the way
+   * {@link Connection.openTable} does for a missing table.
   *
-   * The handle is constructed without a server round trip; an unknown id
-   * surfaces when the handle is used. Dropping the handle has no effect on
-   * the job itself.
+   * The returned {@link Job} answers for its own state, specification,
+   * result, failure and event history, so there is no separate
+   * connection-level call for any of them.
   */
-  abstract job(jobId: string): Job;
+  abstract openJob(jobId: string): Promise<Job>;

  /** List server-side jobs across the database's tables. */
  abstract listJobs(): Promise<JobInfo[]>;

-  /**
-   * Describe a single server-side job by id.
-   *
-   * Resolves to `null` when the server has no such job.
-   */
-  abstract getJob(jobId: string): Promise<JobDescription | null>;
-
  /**
   * Request cancellation of a server-side job by id.
   *
@@ -582,13 +575,6 @@ export abstract class Connection {
   * such job exists. Cancelling an already-terminal job is a no-op success.
   */
  abstract cancelJob(jobId: string): Promise<boolean>;
-
-  /**
-   * The lifecycle event history of a server-side job, as an Arrow table.
-   *
-   * Lists history across all jobs when `jobId` is omitted.
-   */
-  abstract jobHistory(jobId?: string): Promise<ArrowTable>;
 }

 /** @hideconstructor */
@@ -869,7 +855,7 @@ export class LocalConnection extends Connection {
  }

  async dropTableAsync(name: string, namespacePath?: string[]): Promise<Job> {
-    return this.inner.dropTableAsync(name, namespacePath ?? []);
+    return new Job(await this.inner.dropTableAsync(name, namespacePath ?? []));
  }

  async dropAllTables(namespacePath?: string[]): Promise<void> {
@@ -928,29 +914,17 @@ export class LocalConnection extends Connection {
    );
  }

-  job(jobId: string): Job {
-    return this.inner.job(jobId);
+  async openJob(jobId: string): Promise<Job> {
+    return new Job(await this.inner.openJob(jobId));
  }

  async listJobs(): Promise<JobInfo[]> {
    return this.inner.listJobs();
  }

-  async getJob(jobId: string): Promise<JobDescription | null> {
-    return this.inner.getJob(jobId);
-  }
-
  async cancelJob(jobId: string): Promise<boolean> {
    return this.inner.cancelJob(jobId);
  }
-
-  async jobHistory(jobId?: string): Promise<ArrowTable> {
-    const buf = await this.inner.jobHistory(jobId);
-    if (buf.length === 0) {
-      return new ArrowTable();
-    }
-    return tableFromIPC(buf);
-  }
 }

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

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

 export {
  AutoQuery,
@@ -26,7 +26,7 @@ export interface IvfPqOptions {
   * This value controls how much the vector is compressed during the quantization step.
   * The more sub vectors there are the less the vector is compressed.  The default is
   * the dimension of the vector divided by 16.  If the dimension is not evenly divisible
-   * by 16 we use the dimension divded by 8.
+   * by 16 we use the dimension divided by 8.
   *
   * The above two cases are highly preferred.  Having 8 or 16 values per subvector allows
   * us to use efficient SIMD instructions.
@@ -228,7 +228,7 @@ export interface HnswPqOptions {
   * This value controls how much the vector is compressed during the quantization step.
   * The more sub vectors there are the less the vector is compressed.  The default is
   * the dimension of the vector divided by 16.  If the dimension is not evenly divisible
-   * by 16 we use the dimension divded by 8.
+   * by 16 we use the dimension divided by 8.
   *
   * The above two cases are highly preferred.  Having 8 or 16 values per subvector allows
   * us to use efficient SIMD instructions.
@@ -825,7 +825,7 @@ export interface IndexOptions {
  /**
   * Advanced index configuration
   *
-   * This option allows you to specify a specfic index to create and also
+   * This option allows you to specify a specific index to create and also
   * allows you to pass in configuration for training the index.
   *
   * See the static methods on Index for details on the various index types.
@@ -0,0 +1,188 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+import { Table as ArrowTable, tableFromIPC } from "apache-arrow";
+import { JobFailureInfo, Job as NativeJob } from "./native";
+
+/** Which of a job's events {@link Job.events} returns. */
+export interface JobEventsOptions {
+  /** Maximum event rows to return, up to the server maximum of 10,000. */
+  limit?: number;
+  /** SQL-like filter over the event columns. */
+  filter?: string;
+}
+
+/**
+ * A handle to an operation that may still be running.
+ *
+ * The operation may already be complete when the handle is created.
+ *
+ * The detail getters read what the handle last observed. Submitting an
+ * operation returns only a job id, so populating them eagerly would cost an
+ * extra round trip on every call:
+ *
+ * - {@link Job.refresh} and {@link Job.status} fetch the whole record.
+ * - {@link Job.wait} records the terminal state it establishes, but not the
+ *   rest of the record.
+ * - Everything is null until one of those runs.
+ *
+ * @hideconstructor
+ */
+export class Job {
+  private readonly inner: NativeJob;
+
+  constructor(inner: NativeJob) {
+    this.inner = inner;
+  }
+
+  /**
+   * Identifies the operation on the server that is running it.
+   *
+   * Operations that run in this process have no server id. The value is
+   * opaque: parsing it or storing it to resume the job later is not supported.
+   */
+  get id(): string | null {
+    return this.inner.id ?? null;
+  }
+
+  /** The last observed lifecycle state, without contacting the backend. */
+  get state(): string | null {
+    return this.inner.state ?? null;
+  }
+
+  /**
+   * The job's type, as the server names it. Null for an in-process job, which
+   * has no server-side record.
+   */
+  get jobType(): string | null {
+    return this.inner.jobType ?? null;
+  }
+
+  /** When the job was created, in milliseconds since the epoch. */
+  get creationMs(): number | null {
+    return this.inner.creationMs ?? null;
+  }
+
+  /** The job-type-specific specification it was submitted with. */
+  // biome-ignore lint/suspicious/noExplicitAny: shape varies by job type
+  get spec(): any | null {
+    return parseJson(this.inner.specJson);
+  }
+
+  /**
+   * The job-type-specific terminal result. Null until the job succeeds, so a
+   * job that never terminates reports its progress through {@link Job.events}
+   * instead.
+   */
+  // biome-ignore lint/suspicious/noExplicitAny: shape varies by job type
+  get result(): any | null {
+    return parseJson(this.inner.resultJson);
+  }
+
+  /** Why the job failed, when it failed and the server reports a reason. */
+  get failure(): JobFailureInfo | null {
+    return this.inner.failure ?? null;
+  }
+
+  /**
+   * The operation's current lifecycle state: "running", "finished", "failed",
+   * or "cancelled".
+   *
+   * A point snapshot; unlike {@link Job.wait} it does not block or reject on a
+   * terminal failure state. Also refreshes the getters above.
+   */
+  async status(): Promise<string> {
+    return this.inner.status();
+  }
+
+  /** Wait until the operation reaches a terminal state. */
+  async wait(): Promise<void> {
+    return this.inner.wait();
+  }
+
+  /** Request cancellation. Cancelling a finished operation is a no-op. */
+  async cancel(): Promise<void> {
+    return this.inner.cancel();
+  }
+
+  /**
+   * Ask the backend for this job's current state, and for a server-side job
+   * its full record, then cache it for the getters above.
+   */
+  async refresh(): Promise<void> {
+    return this.inner.refresh();
+  }
+
+  /**
+   * This job's recorded lifecycle events.
+   *
+   * Where the getters above report a terminal result only once the job reaches
+   * one, events are written as the job runs and outlive the workers that
+   * produced them. A distributed job records a `claim`/`claim_complete` pair
+   * per unit of work, each carrying `rows_processed`, so a job that never
+   * finishes still accounts for what it did.
+   *
+   * The server caps results at 1000 rows by default and 10,000 at most, and
+   * truncates without saying so, so pass `limit` for a job that emits an event
+   * per fragment. `filter` is a SQL-like expression over the `state`,
+   * `updated_by`, `emitted_from`, `emitted_by`, and `claim_entity` columns.
+   */
+  async events(options?: JobEventsOptions): Promise<ArrowTable> {
+    const buf = await this.inner.events(options?.limit, options?.filter);
+    if (buf.length === 0) {
+      return new ArrowTable();
+    }
+    return tableFromIPC(buf);
+  }
+
+  /**
+   * Every field the handle currently knows, one per line, with the JSON
+   * payloads indented -- a refresh job's spec and result are the point of
+   * printing it.
+   */
+  toString(): string {
+    if (this.state === null) {
+      const known = this.id === null ? "" : `id=${JSON.stringify(this.id)}, `;
+      return `Job(${known}not refreshed)`;
+    }
+    const fields: string[] = [];
+    if (this.id !== null) {
+      fields.push(`id=${JSON.stringify(this.id)}`);
+    }
+    fields.push(`state=${JSON.stringify(this.state)}`);
+    if (this.jobType !== null) {
+      fields.push(`jobType=${JSON.stringify(this.jobType)}`);
+    }
+    if (this.creationMs !== null) {
+      fields.push(`creationMs=${this.creationMs}`);
+    }
+    for (const [name, value] of [
+      ["spec", this.spec],
+      ["result", this.result],
+    ] as const) {
+      if (value !== null) {
+        fields.push(`${name}=${indentJson(value)}`);
+      }
+    }
+    if (this.failure !== null) {
+      fields.push(`failure=${indentJson(this.failure)}`);
+    }
+    return `Job(${fields.map((field) => `\n${REPR_INDENT}${field},`).join("")}\n)`;
+  }
+
+  [Symbol.for("nodejs.util.inspect.custom")](): string {
+    return this.toString();
+  }
+}
+
+const REPR_INDENT = "    ";
+
+// biome-ignore lint/suspicious/noExplicitAny: shape varies by job type
+function indentJson(value: any): string {
+  return JSON.stringify(value, null, 4).replace(/\n/g, `\n${REPR_INDENT}`);
+}
+
+// biome-ignore lint/suspicious/noExplicitAny: shape varies by job type
+function parseJson(raw: string | null | undefined): any | null {
+  return raw === null || raw === undefined ? null : JSON.parse(raw);
+}
@@ -19,6 +19,8 @@ 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[];
 }

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

@@ -27,7 +27,7 @@ export class MergeInsertBuilder {
   * but that behavior is subject to change.
   *
   * An optional condition may be specified.  If it is, then only
-   * matched rows that satisfy the condtion will be updated.  Any
+   * matched rows that satisfy the condition will be updated.  Any
   * rows that do not satisfy the condition will be left as they
   * are.  Failing to satisfy the condition does not cause a
   * "matched row" to become a "not matched" row.
@@ -3,7 +3,7 @@

 // The utilities in this file help sanitize data from the user's arrow
 // library into the types expected by vectordb's arrow library.  Node
-// generally allows for mulitple versions of the same library (and sometimes
+// generally allows for multiple versions of the same library (and sometimes
 // even multiple copies of the same version) to be installed at the same
 // time.  However, arrow-js uses instanceof which expected that the input
 // comes from the exact same library instance.  This is not always the case
@@ -19,6 +19,7 @@ import {

 import { EmbeddingFunctionConfig, getRegistry } from "./embedding/registry";
 import { IndexOptions } from "./indices";
+import { Job } from "./job";
 import { MergeInsertBuilder } from "./merge";
 import {
  AddColumnsResult,
@@ -30,7 +31,6 @@ import {
  DropColumnsResult,
  IndexConfig,
  IndexStatistics,
-  Job,
  LsmStats,
  Branches as NativeBranches,
  OptimizeStats,
@@ -313,7 +313,7 @@ export abstract class Table {
   * Note: if your condition is something like "some_id_column == 7" and
   * you are updating many rows (with different ids) then you will get
   * better performance with a single [`merge_insert`] call instead of
-   * repeatedly calilng this method.
+   * repeatedly calling this method.
   * @param {Map<string, string> | Record<string, string>} updates - the
   * columns to update
   * @returns {Promise<UpdateResult>} A promise that resolves to an object
@@ -919,6 +919,16 @@ export abstract class Table {
  /** Return the table as an arrow table */
  abstract toArrow(): Promise<ArrowTable>;

+  /**
+   * Create a {@link MergeInsertBuilder}, which combines new data with the
+   * existing table in a single transaction — inserting, updating and deleting
+   * rows depending on how they match.
+   *
+   * @param on - The column, or columns, to match source rows against target
+   * rows on. Typically a key or id column. Several columns match on the
+   * composite key: a source row updates a target row only when it agrees on
+   * every one of them.
+   */
  abstract mergeInsert(on: string | string[]): MergeInsertBuilder;

  /** List all the stats of a specified index
@@ -1114,13 +1124,15 @@ export class LocalTable extends Table {
  ): Promise<Job> {
    // biome-ignore lint/suspicious/noExplicitAny: skip
    const nativeIndex = (options?.config as any)?.inner;
-    return await this.inner.createIndexAsync(
-      nativeIndex,
-      column,
-      options?.replace,
-      options?.waitTimeoutSeconds,
-      options?.name,
-      options?.train,
+    return new Job(
+      await this.inner.createIndexAsync(
+        nativeIndex,
+        column,
+        options?.replace,
+        options?.waitTimeoutSeconds,
+        options?.name,
+        options?.train,
+      ),
    );
  }

@@ -1303,7 +1315,7 @@ export class LocalTable extends Table {
  }

  async refreshColumnAsync(column: string): Promise<Job> {
-    return await this.inner.refreshColumnAsync(column);
+    return new Job(await this.inner.refreshColumnAsync(column));
  }

  async refreshMaterializedView(
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-darwin-arm64",
-	"version": "0.38.0-beta.12",
+	"version": "0.39.0-beta.6",
 	"os": ["darwin"],
 	"cpu": ["arm64"],
 	"main": "lancedb.darwin-arm64.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-gnu",
-	"version": "0.38.0-beta.12",
+	"version": "0.39.0-beta.6",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-gnu.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-musl",
-	"version": "0.38.0-beta.12",
+	"version": "0.39.0-beta.6",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-musl.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-gnu",
-	"version": "0.38.0-beta.12",
+	"version": "0.39.0-beta.6",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-gnu.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-musl",
-	"version": "0.38.0-beta.12",
+	"version": "0.39.0-beta.6",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-musl.node",
@@ -1,6 +1,6 @@
 {
  "name": "@lancedb/lancedb-win32-arm64-msvc",
-  "version": "0.38.0-beta.12",
+  "version": "0.39.0-beta.6",
  "os": [
    "win32"
  ],
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-win32-x64-msvc",
-	"version": "0.38.0-beta.12",
+	"version": "0.39.0-beta.6",
 	"os": ["win32"],
 	"cpu": ["x64"],
 	"main": "lancedb.win32-x64-msvc.node",
@@ -11,7 +11,7 @@
    "ann"
  ],
  "private": false,
-  "version": "0.38.0-beta.12",
+  "version": "0.39.0-beta.6",
  "main": "dist/index.js",
  "exports": {
    ".": "./dist/index.js",
@@ -44,7 +44,7 @@
    "@biomejs/biome": "^1.7.3",
    "@jest/globals": "^29.7.0",
    "@napi-rs/cli": "3.7.0",
-    "@opentelemetry/sdk-metrics": "^1.30.0",
+    "@opentelemetry/sdk-metrics": "^2.10.0",
    "@types/axios": "^0.14.0",
    "@types/jest": "^29.1.2",
    "@types/node": "22.7.4",
@@ -56,7 +56,7 @@
    "eslint": "^8.57.0",
    "jest": "^29.7.0",
    "shx": "^0.3.4",
-    "tmp": "^0.2.3",
+    "tmp": "^0.2.7",
    "ts-jest": "^29.1.2",
    "typedoc": "0.26.4",
    "typedoc-plugin-markdown": "4.2.1",
@@ -67,7 +67,7 @@
    "timeout": "3m"
  },
  "engines": {
-    "node": ">= 18"
+    "node": ">= 22"
  },
  "packageManager": "pnpm@11.1.1",
  "cpu": ["x64", "arm64"],
@@ -101,7 +101,7 @@
    "openai": "4.29.2"
  },
  "peerDependencies": {
-    "@types/node": ">=18",
+    "@types/node": ">=22",
    "apache-arrow": ">=15.0.0 <=18.1.0"
  },
  "peerDependenciesMeta": {
@@ -16,3 +16,41 @@ allowBuilds:
  onnxruntime-node: true
  protobufjs: true
  sharp: true
+
+minimumReleaseAgeExclude:
+  - protobufjs@7.5.8
+  - tmp@0.2.6
+  - form-data@4.0.6
+  - tar@7.5.16
+  - markdown-it@14.1.2
+  - linkify-it@5.0.1
+  - js-yaml@3.15.0
+  - js-yaml@4.1.2
+  - protobufjs@7.6.1
+  - protobufjs@7.6.3
+  - '@babel/core@7.29.1'
+  - axios@1.18.0
+  - brace-expansion@2.1.2
+  - brace-expansion@1.1.16
+  - js-yaml@4.3.0
+  - tar@7.5.18
+  - tar@7.5.19
+  - tar@7.5.17
+  - protobufjs@7.6.5
+  - linkify-it@5.0.2
+  - sharp@0.35.0
+  - brace-expansion@1.1.17
+  - brace-expansion@2.1.3
+  - brace-expansion@2.1.4
+  - brace-expansion@1.1.18
+  - js-yaml@3.15.1
+  - js-yaml@4.3.1
+  - tar@7.5.21
+  - '@opentelemetry/core@2.8.0'
+
+# @huggingface/transformers pins sharp ^0.33.5 and no released version has moved
+# past ^0.34.5, all of which inherit the libvips CVEs in GHSA-f88m-g3jw-g9cj.
+# Force the patched line. sharp is only reached by transformers' image pipeline,
+# which LanceDB's text embedding function never uses.
+overrides:
+  sharp: ^0.35.4
@@ -442,13 +442,15 @@ impl Connection {
        self.get_inner()?.drop_all_tables(&ns).await.default_error()
    }

-    /// A `Job` handle for a server-side job by id.
+    /// Open a server-side job by id, returning a handle with its record
+    /// already populated. Rejects when the server has no such job.
    ///
-    /// The handle is constructed without a server round trip; an unknown id
-    /// surfaces when the handle is used.
-    #[napi]
-    pub fn job(&self, job_id: String) -> napi::Result<crate::job::Job> {
-        let job = self.get_inner()?.job(job_id).default_error()?;
+    /// The returned handle answers for its own state, specification, result,
+    /// failure and event history, so there is no separate connection-level
+    /// call for any of them.
+    #[napi(catch_unwind)]
+    pub async fn open_job(&self, job_id: String) -> napi::Result<crate::job::Job> {
+        let job = self.get_inner()?.open_job(&job_id).await.default_error()?;
        Ok(crate::job::Job::new(job))
    }

@@ -459,17 +461,6 @@ impl Connection {
        Ok(jobs.into_iter().map(Into::into).collect())
    }

-    /// Describe a single server-side job by id. `null` when the server has
-    /// no such job.
-    #[napi(catch_unwind)]
-    pub async fn get_job(
-        &self,
-        job_id: String,
-    ) -> napi::Result<Option<crate::job::JobDescription>> {
-        let description = self.get_inner()?.get_job(&job_id).await.default_error()?;
-        Ok(description.map(Into::into))
-    }
-
    /// Request cancellation of a server-side job by id. Returns true if the
    /// server accepted the cancellation, false if no such job exists.
    #[napi(catch_unwind)]
@@ -477,34 +468,6 @@ impl Connection {
        self.get_inner()?.cancel_job(&job_id).await.default_error()
    }

-    /// The lifecycle event history of a server-side job (all jobs when
-    /// `job_id` is null), as an Arrow IPC stream buffer. Empty when there is
-    /// no history.
-    #[napi(catch_unwind)]
-    pub async fn job_history(&self, job_id: Option<String>) -> napi::Result<Buffer> {
-        let batches = self
-            .get_inner()?
-            .job_history(job_id.as_deref())
-            .await
-            .default_error()?;
-        let Some(first) = batches.first() else {
-            return Ok(Buffer::from(Vec::<u8>::new()));
-        };
-        let mut out = Vec::new();
-        let mut writer = arrow_ipc::writer::StreamWriter::try_new(&mut out, &first.schema())
-            .map_err(|e| napi::Error::from_reason(e.to_string()))?;
-        for batch in &batches {
-            writer
-                .write(batch)
-                .map_err(|e| napi::Error::from_reason(e.to_string()))?;
-        }
-        writer
-            .finish()
-            .map_err(|e| napi::Error::from_reason(e.to_string()))?;
-        drop(writer);
-        Ok(Buffer::from(out))
-    }
-
    #[napi(catch_unwind)]
    /// Describe a namespace and return its properties.
    pub async fn describe_namespace(
@@ -3,6 +3,9 @@

 use std::sync::Arc;

+use arrow_array::RecordBatch;
+use lancedb::job::JobEventsRequest;
+use napi::bindgen_prelude::Buffer;
 use napi_derive::napi;

 use crate::error::NapiErrorExt;
@@ -55,12 +58,98 @@ impl Job {
    pub async fn cancel(&self) -> napi::Result<()> {
        self.inner.cancel().await.default_error()
    }
+
+    /// Ask the backend for this job's current state, and for a server-side job
+    /// its full record, then cache it for the getters below.
+    ///
+    /// They are all null until this runs, because submitting an operation
+    /// returns only a job id. {@link Job.status} fetches the whole record too;
+    /// {@link Job.wait} records only the terminal state it establishes.
+    #[napi(catch_unwind)]
+    pub async fn refresh(&self) -> napi::Result<()> {
+        self.inner.refresh().await.default_error()
+    }
+
+    /// The last observed lifecycle state, without contacting the backend.
+    #[napi(getter)]
+    pub fn state(&self) -> Option<String> {
+        self.inner.state()
+    }
+
+    /// The job's type, as the server names it. Null for an in-process job,
+    /// which has no server-side record.
+    #[napi(getter)]
+    pub fn job_type(&self) -> Option<String> {
+        self.inner.job_type()
+    }
+
+    /// When the job was created, in milliseconds since the epoch.
+    #[napi(getter)]
+    pub fn creation_ms(&self) -> Option<i64> {
+        self.inner.creation_ms()
+    }
+
+    /// The job-type-specific specification as a JSON string, when present.
+    #[napi(getter)]
+    pub fn spec_json(&self) -> Option<String> {
+        self.inner.spec().map(|spec| spec.to_string())
+    }
+
+    /// The job-type-specific terminal result as a JSON string. Null until the
+    /// job succeeds, so a job that never terminates reports its progress
+    /// through {@link Job.events} instead.
+    #[napi(getter)]
+    pub fn result_json(&self) -> Option<String> {
+        self.inner.result().map(|result| result.to_string())
+    }
+
+    /// Why the job failed, when it failed and the server reports a reason.
+    #[napi(getter)]
+    pub fn failure(&self) -> Option<JobFailureInfo> {
+        self.inner.failure().map(|failure| JobFailureInfo {
+            phase: failure.phase,
+            message: failure.message,
+            retryable: failure.retryable,
+        })
+    }
+
+    /// This job's recorded lifecycle events, as an Arrow IPC stream buffer.
+    /// The TypeScript wrapper turns it into an Arrow table.
+    #[napi(catch_unwind)]
+    pub async fn events(&self, limit: Option<u32>, filter: Option<String>) -> napi::Result<Buffer> {
+        let batches = self
+            .inner
+            .events(JobEventsRequest { limit, filter })
+            .await
+            .default_error()?;
+        batches_to_ipc_buffer(&batches)
+    }
+}
+
+/// Serialise Arrow batches as a single IPC stream for the TypeScript layer.
+fn batches_to_ipc_buffer(batches: &[RecordBatch]) -> napi::Result<Buffer> {
+    let Some(first) = batches.first() else {
+        return Ok(Buffer::from(Vec::<u8>::new()));
+    };
+    let mut out = Vec::new();
+    let mut writer = arrow_ipc::writer::StreamWriter::try_new(&mut out, &first.schema())
+        .map_err(|e| napi::Error::from_reason(e.to_string()))?;
+    for batch in batches {
+        writer
+            .write(batch)
+            .map_err(|e| napi::Error::from_reason(e.to_string()))?;
+    }
+    writer
+        .finish()
+        .map_err(|e| napi::Error::from_reason(e.to_string()))?;
+    drop(writer);
+    Ok(Buffer::from(out))
 }

 /// A row from `Connection.listJobs`: one server-side job.
 #[napi(object)]
 pub struct JobInfo {
-    /// The job id -- what `Connection.getJob` and `Connection.cancelJob`
+    /// The job id -- what `Connection.openJob` and `Connection.cancelJob`
    /// accept.
    pub job_id: String,
    /// The table the job runs against, without URI or namespace.
@@ -91,36 +180,3 @@ pub struct JobFailureInfo {
    pub message: Option<String>,
    pub retryable: Option<bool>,
 }
-
-/// A described job from `Connection.getJob`.
-#[napi(object)]
-pub struct JobDescription {
-    pub job_id: String,
-    pub job_type: String,
-    /// Lifecycle state: "running", "finished", "failed", or "cancelled".
-    pub state: String,
-    /// When the job was created, in milliseconds since the epoch.
-    pub creation_ms: i64,
-    /// The job-type-specific specification as a JSON string, when present.
-    pub spec_json: Option<String>,
-    /// Why the job failed, when the job is failed and the server reports a
-    /// reason.
-    pub failure: Option<JobFailureInfo>,
-}
-
-impl From<lancedb::database::JobDescription> for JobDescription {
-    fn from(description: lancedb::database::JobDescription) -> Self {
-        Self {
-            job_id: description.job_id,
-            job_type: description.job_type,
-            state: description.state,
-            creation_ms: description.creation_ms,
-            spec_json: (!description.spec.is_null()).then(|| description.spec.to_string()),
-            failure: description.failure.map(|failure| JobFailureInfo {
-                phase: failure.phase,
-                message: failure.message,
-                retryable: failure.retryable,
-            }),
-        }
-    }
-}
@@ -664,7 +664,11 @@ impl JsFullTextQuery {
 }

 fn parse_fts_query(query: Object) -> napi::Result<FullTextSearchQuery> {
-    if let Ok(Some(query)) = query.get::<&JsFullTextQuery>("query") {
+    // `&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") {
        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.38.0-beta.12"
+version = "0.39.0-beta.6"
 publish = false
 edition.workspace = true
 description = "Python bindings for LanceDB"
@@ -28,7 +28,7 @@ env_logger.workspace = true
 log.workspace = true
 # Maturin enables extension-module mode for Python builds. Keeping it out of
 # Cargo features lets Rust unit tests link against libpython.
-pyo3 = { version = "0.28", features = ["abi3-py310", "chrono"] }
+pyo3 = { version = "0.28", features = ["abi3-py310", "chrono", "uuid"] }
 chrono.workspace = true
 pyo3-async-runtimes = { version = "0.28", features = [
    "attributes",
@@ -40,6 +40,7 @@ serde.workspace = true
 serde_json.workspace = true
 snafu.workspace = true
 tokio.workspace = true
+uuid.workspace = true
 libc = "0.2"

 [build-dependencies]
@@ -139,6 +139,7 @@ include = [
    "python/lancedb/exceptions.py",
    "python/lancedb/background_loop.py",
    "python/lancedb/schema.py",
+    "python/lancedb/sql.py",
    "python/lancedb/remote/__init__.py",
    "python/lancedb/remote/errors.py",
    "python/lancedb/embeddings/__init__.py",
@@ -22,7 +22,11 @@ from .remote.db import RemoteDBConnection
 from .expr import Expr, col, lit, func
 from .schema import blob, vector
 from .job import AsyncJob, Job
+from .sql import AsyncQuery as AsyncSqlQuery
+from .sql import Query as SqlQuery
+from .sql import QueryDescription
 from .functions import (
+    AssignmentMapping as AssignmentMapping,
    FunctionArtifactRequest as FunctionArtifactRequest,
    FunctionApplication as FunctionApplication,
    FunctionBinding as FunctionBinding,
@@ -101,6 +105,7 @@ def connect(
    api_key: Optional[str] = None,
    region: str = "us-east-1",
    host_override: Optional[str] = None,
+    sql_host_override: Optional[str] = None,
    read_consistency_interval: Optional[timedelta] = None,
    request_thread_pool: Optional[Union[int, ThreadPoolExecutor]] = None,
    client_config: Union[ClientConfig, Dict[str, Any], None] = None,
@@ -129,6 +134,9 @@ def connect(
        The region to use for LanceDB Cloud.
    host_override: str, optional
        The override url for LanceDB Cloud.
+    sql_host_override: str, optional
+        The remote SQL service endpoint override. The client connects lazily when SQL
+        is first executed and retains that connection.
    read_consistency_interval: timedelta, default None
        The interval at which to check for updates to the table from other
        processes. If None, then consistency is not checked. For performance
@@ -270,6 +278,7 @@ def connect(
            api_key,
            region,
            host_override,
+            sql_host_override=sql_host_override,
            # TODO: remove this (deprecation warning downstream)
            request_thread_pool=request_thread_pool,
            client_config=client_config,
@@ -412,6 +421,7 @@ def deserialize_conn(
            parsed["api_key"],
            parsed.get("region", "us-east-1"),
            host_override=parsed.get("host_override"),
+            sql_host_override=parsed.get("sql_host_override"),
            client_config=parsed.get("client_config"),
            storage_options=storage_options,
        )
@@ -425,6 +435,7 @@ async def connect_async(
    api_key: Optional[str] = None,
    region: str = "us-east-1",
    host_override: Optional[str] = None,
+    sql_host_override: Optional[str] = None,
    read_consistency_interval: Optional[timedelta] = None,
    client_config: Optional[Union[ClientConfig, Dict[str, Any]]] = None,
    storage_options: Optional[Dict[str, str]] = None,
@@ -447,6 +458,9 @@ async def connect_async(
        The region to use for LanceDB Cloud.
    host_override: str, optional
        The override url for LanceDB Cloud.
+    sql_host_override: str, optional
+        The remote SQL service endpoint override. The client connects lazily when SQL
+        is first executed and retains that connection.
    read_consistency_interval: timedelta, default None
        The interval at which to check for updates to the table from other
        processes. If None, then consistency is not checked. For performance
@@ -534,6 +548,7 @@ async def connect_async(
            api_key,
            region,
            host_override,
+            sql_host_override,
            read_consistency_interval_secs,
            client_config,
            storage_options,
@@ -556,6 +571,7 @@ __all__ = [
    "connect_namespace_async",
    "AsyncConnection",
    "AsyncJob",
+    "AsyncSqlQuery",
    "AsyncLanceNamespaceDBConnection",
    "AsyncTable",
    "FtsToken",
@@ -570,6 +586,8 @@ __all__ = [
    "vector",
    "DBConnection",
    "Job",
+    "QueryDescription",
+    "SqlQuery",
    "LanceDBConnection",
    "LanceNamespaceDBConnection",
    "LsmWriteSpec",
@@ -1,6 +1,7 @@
 from datetime import date, datetime, timedelta
 from decimal import Decimal
 from typing import Dict, List, Optional, Tuple, Any, TypedDict, Union, Literal
+from uuid import UUID

 import pyarrow as pa

@@ -147,15 +148,20 @@ class Connection(object):
        start_after: Optional[str],
        limit: Optional[int],
    ) -> list[str]: ...  # Deprecated: Use list_tables instead
-    def job(self, job_id: str) -> Job: ...
+    async def open_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 list_functions(self) -> List[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: ...
-    async def job_history(
-        self, job_id: Optional[str] = None
-    ) -> List[pa.RecordBatch]: ...
+    async def execute_query_async(
+        self,
+        query: str,
+        *,
+        default_namespace_path: Optional[List[str]] = None,
+    ) -> SqlQuery: ...
+    async def describe_query(self, query_id: UUID) -> QueryDescription: ...
    async def create_table(
        self,
        name: str,
@@ -234,9 +240,20 @@ class BlobFile:
 class Job:
    @property
    def id(self) -> Optional[str]: ...
+    @property
+    def _state(self) -> Optional[str]: ...
+    @property
+    def _description(self) -> Optional[JobDescription]: ...
    async def status(self) -> str: ...
    async def wait(self) -> Optional[str]: ...
    async def cancel(self) -> None: ...
+    async def refresh(self) -> None: ...
+    async def events(
+        self,
+        *,
+        limit: Optional[int] = None,
+        filter: Optional[str] = None,
+    ) -> pa.Table: ...

 class JobInfo:
    @property
@@ -268,10 +285,33 @@ class JobDescription:
    @property
    def creation_ms(self) -> int: ...
    @property
-    def spec_json(self) -> Optional[str]: ...
+    def _spec_json(self) -> Optional[str]: ...
+    @property
+    def _result_json(self) -> Optional[str]: ...
+    @property
+    def spec(self) -> Optional[Any]: ...
+    @property
+    def result(self) -> Optional[Any]: ...
    @property
    def failure(self) -> Optional[JobFailureInfo]: ...

+class SqlQuery:
+    @property
+    def id(self) -> UUID: ...
+    async def describe(self) -> QueryDescription: ...
+    async def reader(self) -> RecordBatchStream: ...
+    async def cancel(self) -> None: ...
+
+class QueryDescription:
+    @property
+    def id(self) -> UUID: ...
+    @property
+    def status(self) -> str: ...
+    @property
+    def progress(self) -> Optional[float]: ...
+    @property
+    def expires_at(self) -> Optional[datetime]: ...
+
 class Table:
    def name(self) -> str: ...
    def __repr__(self) -> str: ...
@@ -450,6 +490,7 @@ async def connect(
    api_key: Optional[str],
    region: Optional[str],
    host_override: Optional[str],
+    sql_host_override: Optional[str],
    read_consistency_interval: Optional[float],
    client_config: Optional[Union[ClientConfig, Dict[str, Any]]],
    storage_options: Optional[Dict[str, str]],
@@ -606,6 +647,7 @@ 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]]]
@@ -19,6 +19,7 @@ from typing import (
    Optional,
    Union,
 )
+from uuid import UUID

 if sys.version_info >= (3, 12):
    from typing import override
@@ -47,6 +48,9 @@ from . import __version__
 from ._lancedb import connect as lancedb_connect  # type: ignore
 from .functions import FunctionVersion, UdfDefinition
 from .job import AsyncJob, Job, _typed_job
+from .sql import AsyncQuery as AsyncSqlQuery
+from .sql import Query as SqlQuery
+from .sql import QueryDescription
 from .materialized_view import (
    AsyncMaterializedView,
    MaterializedView,
@@ -68,10 +72,11 @@ import deprecation

 if TYPE_CHECKING:
    import pyarrow as pa
+    from .arrow import AsyncRecordBatchReader
    from .pydantic import LanceModel

    from ._lancedb import Connection as LanceDbConnection
-    from ._lancedb import JobDescription, JobInfo
+    from ._lancedb import JobInfo
    from .common import DATA, URI
    from .embeddings import EmbeddingFunctionConfig
    from ._lancedb import Session
@@ -712,26 +717,51 @@ class DBConnection(EnforceOverrides):
            "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.
+    def list_functions(self) -> List[FunctionVersion]:
+        """List every published immutable Function version.

-        The handle is constructed without a server round trip; an unknown id
-        surfaces when the handle is used. Dropping the handle has no effect
-        on the job itself.
+        Results are ordered by Function name then version. Local connections
+        raise ``NotImplementedError``.
+
+        Examples
+        --------
+        List the identities available to use in Function-backed columns:
+
+        ```python
+        [(function.name, function.version) for function in db.list_functions()]
+        ```
        """
-        raise NotImplementedError("job is not supported for this connection type")
+        raise NotImplementedError(
+            "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 open_job(self, job_id: str) -> Job:
+        """Open a server-side job by id, returning a handle with its record
+        already populated.
+
+        The returned [Job][lancedb.job.Job] answers for its own state,
+        specification, result, failure and event history, so there is no
+        separate connection-level call for any of them.
+
+        Raises `JobNotFoundError` when the server has no such job, the way
+        `open_table` does for a missing table.
+        """
+        raise NotImplementedError("open_job is not supported for this connection type")

    def list_jobs(self) -> List[JobInfo]:
        """List server-side jobs across the database's tables."""
        raise NotImplementedError("list_jobs is not supported for this connection type")

-    def get_job(self, job_id: str) -> Optional[JobDescription]:
-        """Describe a single server-side job by id.
-
-        Returns None when the server has no such job.
-        """
-        raise NotImplementedError("get_job is not supported for this connection type")
-
    def cancel_job(self, job_id: str) -> bool:
        """Request cancellation of a server-side job by id.

@@ -743,14 +773,38 @@ class DBConnection(EnforceOverrides):
            "cancel_job is not supported for this connection type"
        )

-    def job_history(self, job_id: Optional[str] = None) -> List[pa.RecordBatch]:
-        """The lifecycle event history of a server-side job, as Arrow batches.
+    def execute_query(
+        self,
+        query: str,
+        *,
+        default_namespace_path: Optional[List[str]] = None,
+    ) -> pa.RecordBatchReader:
+        """Execute SQL and return a blocking Arrow reader.

-        Lists history across all jobs when `job_id` is None.
+        This submits through :meth:`execute_query_async` and waits until the
+        initial result stream is readable. It does not wait for the full query
+        to finish.
        """
-        raise NotImplementedError(
-            "job_history is not supported for this connection type"
-        )
+        return self.execute_query_async(
+            query,
+            default_namespace_path=default_namespace_path,
+        ).reader()
+
+    def execute_query_async(
+        self,
+        query: str,
+        *,
+        default_namespace_path: Optional[List[str]] = None,
+    ) -> SqlQuery:
+        """Start executing SQL and return its query handle.
+
+        Local connections do not support SQL.
+        """
+        raise NotImplementedError("SQL is not supported for this connection type")
+
+    def describe_query(self, query_id: UUID) -> QueryDescription:
+        """Describe a submitted SQL query by its connection-scoped id."""
+        raise NotImplementedError("SQL is not supported for this connection type")


 class LanceDBConnection(DBConnection):
@@ -847,6 +901,7 @@ class LanceDBConnection(DBConnection):
                None,
                None,
                None,
+                None,
                read_consistency_interval_secs,
                None,
                storage_options,
@@ -1395,14 +1450,11 @@ class LanceDBConnection(DBConnection):
        )

    @override
-    def job(self, job_id: str) -> Job:
-        """A [Job][lancedb.job.Job] handle for a server-side job by id.
-
-        The handle is constructed without a server round trip; an unknown id
-        surfaces when the handle is used. Dropping the handle has no effect
-        on the job itself.
+    def open_job(self, job_id: str) -> Job:
+        """Open a server-side job by id. See
+        [DBConnection.open_job][lancedb.db.DBConnection.open_job].
        """
-        return Job(self._conn.job(job_id))
+        return Job(LOOP.run(self._conn.open_job(job_id)))

    @override
    def create_function_async(self, definition: UdfDefinition) -> Job[FunctionVersion]:
@@ -1413,19 +1465,19 @@ class LanceDBConnection(DBConnection):
    def get_function(self, name: str, *, version: str) -> FunctionVersion:
        return LOOP.run(self._conn.get_function(name, version=version))

+    @override
+    def list_functions(self) -> List[FunctionVersion]:
+        return LOOP.run(self._conn.list_functions())
+
+    @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."""
        return LOOP.run(self._conn.list_jobs())

-    @override
-    def get_job(self, job_id: str) -> Optional[JobDescription]:
-        """Describe a single server-side job by id.
-
-        Returns None when the server has no such job.
-        """
-        return LOOP.run(self._conn.get_job(job_id))
-
    @override
    def cancel_job(self, job_id: str) -> bool:
        """Request cancellation of a server-side job by id.
@@ -1436,14 +1488,6 @@ class LanceDBConnection(DBConnection):
        """
        return LOOP.run(self._conn.cancel_job(job_id))

-    @override
-    def job_history(self, job_id: Optional[str] = None) -> List[pa.RecordBatch]:
-        """The lifecycle event history of a server-side job, as Arrow batches.
-
-        Lists history across all jobs when `job_id` is None.
-        """
-        return LOOP.run(self._conn.job_history(job_id))
-
    @override
    def namespace_client(self) -> LanceNamespace:
        """Get the equivalent namespace client for this connection.
@@ -2214,15 +2258,11 @@ class AsyncConnection(object):
            namespace_path = []
        await self._inner.drop_all_tables(namespace_path=namespace_path)

-    def job(self, job_id: str) -> AsyncJob:
-        """An [AsyncJob][lancedb.job.AsyncJob] handle for a server-side job
-        by id.
-
-        The handle is constructed without a server round trip; an unknown id
-        surfaces when the handle is used. Dropping the handle has no effect
-        on the job itself.
+    async def open_job(self, job_id: str) -> AsyncJob:
+        """Open a server-side job by id. See
+        [DBConnection.open_job][lancedb.db.DBConnection.open_job].
        """
-        return AsyncJob(self._inner.job(job_id))
+        return AsyncJob(await self._inner.open_job(job_id))

    async def create_function_async(
        self, definition: UdfDefinition
@@ -2243,17 +2283,25 @@ 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 list_functions(self) -> List[FunctionVersion]:
+        """List every published immutable Function version.
+
+        Results are ordered by Function name then version. Local connections
+        raise ``NotImplementedError``.
+        """
+        return [
+            FunctionVersion.from_json(value)
+            for value in await self._inner.list_functions()
+        ]
+
+    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()

-    async def get_job(self, job_id: str) -> Optional[JobDescription]:
-        """Describe a single server-side job by id.
-
-        Returns None when the server has no such job.
-        """
-        return await self._inner.get_job(job_id)
-
    async def cancel_job(self, job_id: str) -> bool:
        """Request cancellation of a server-side job by id.

@@ -2263,12 +2311,46 @@ class AsyncConnection(object):
        """
        return await self._inner.cancel_job(job_id)

-    async def job_history(self, job_id: Optional[str] = None) -> List[pa.RecordBatch]:
-        """The lifecycle event history of a server-side job, as Arrow batches.
+    async def execute_query(
+        self,
+        query: str,
+        *,
+        default_namespace_path: Optional[List[str]] = None,
+    ) -> AsyncRecordBatchReader:
+        """Execute SQL and return an asynchronous Arrow reader.

-        Lists history across all jobs when `job_id` is None.
+        This submits through :meth:`execute_query_async` and waits until the
+        initial result stream is readable. It does not wait for the full query
+        to finish.
        """
-        return await self._inner.job_history(job_id)
+        submitted = await self.execute_query_async(
+            query,
+            default_namespace_path=default_namespace_path,
+        )
+        return await submitted.reader()
+
+    async def execute_query_async(
+        self,
+        query: str,
+        *,
+        default_namespace_path: Optional[List[str]] = None,
+    ) -> AsyncSqlQuery:
+        """Start executing SQL and return its query handle.
+
+        The database from ``connect_async`` is used for unqualified database
+        references. The namespace defaults to ``["public"]``. Local
+        connections raise ``NotImplementedError``.
+        """
+        return AsyncSqlQuery(
+            await self._inner.execute_query_async(
+                query,
+                default_namespace_path=default_namespace_path,
+            )
+        )
+
+    async def describe_query(self, query_id: UUID) -> QueryDescription:
+        """Describe a submitted SQL query by its connection-scoped id."""
+        return await self._inner.describe_query(query_id)

    async def namespace_client(self) -> LanceNamespace:
        """Get the equivalent namespace client for this connection.
@@ -21,7 +21,7 @@ class GteEmbeddings(TextEmbeddingFunction):
    An embedding function that uses GTE-LARGE MLX format(for Apple silicon devices only)
    as well as the standard cpu/gpu version from: https://huggingface.co/thenlper/gte-large.

-    For Apple users, you will need the mlx package insalled, which can be done with:
+    For Apple users, you will need the mlx package installed, which can be done with:
        pip install mlx

    Parameters
@@ -60,7 +60,7 @@ class InstructorEmbeddingFunction(TextEmbeddingFunction):

    import lancedb
    from lancedb.pydantic import LanceModel, Vector
-    from lancedb.embeddings import get_registry, InstuctorEmbeddingFunction
+    from lancedb.embeddings import get_registry, InstructorEmbeddingFunction

    instructor = get_registry().get("instructor").create(
        source_instruction="represent the document for retrieval",
@@ -35,3 +35,9 @@ class JobCancelledError(RuntimeError):
    """Exception raised when an asynchronous job was cancelled."""

    pass
+
+
+class JobNotFoundError(ValueError):
+    """Exception raised when opening a job the server does not have."""
+
+    pass
@@ -49,11 +49,25 @@ 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")
@@ -239,6 +253,23 @@ 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):
@@ -247,18 +278,28 @@ 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.

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

    name: str
@@ -429,11 +470,7 @@ class InputBinding(_RemoteValue):


 class OutputMapping(_RemoteValue):
-    """One stable result-field mapping.
-
-    Assignment state is outside the Slice 1 client contract. During the NULL
-    transition Lance exposes no public cell-flag identifier to persist here.
-    """
+    """One stable result-field mapping."""

    result_field: str
    output_name: str
@@ -443,6 +480,13 @@ class OutputMapping(_RemoteValue):
    nullable: bool


+class AssignmentMapping(_RemoteValue):
+    """Internal physical column preserving flattened struct validity."""
+
+    output_name: str
+    output_field_id: _Int32
+
+
 class FunctionBinding(_RemoteValue):
    """Immutable Function binding persisted by the Enterprise table service."""

@@ -450,6 +494,7 @@ class FunctionBinding(_RemoteValue):
    function: FunctionVersionRef
    inputs: tuple[InputBinding, ...]
    outputs: tuple[OutputMapping, ...]
+    assignment: Optional[AssignmentMapping] = None
    input_schema: Optional[Mapping[str, Any]] = None
    output_schema: Optional[Mapping[str, Any]] = None

@@ -479,6 +524,13 @@ class RefreshColumnResult(_RemoteValue):


 _FUNCTION_NAME = re.compile(r"^[A-Za-z_][A-Za-z0-9_.-]*$")
+_FUNCTION_BLOB_V2_TYPE = "blob_v2"
+_ARROW_EXTENSION_NAME_KEY = "ARROW:extension:name"
+_BLOB_V2_EXTENSION_NAME = "lance.blob.v2"
+_NESTED_BLOB_COLLECTION_ERROR = (
+    "unsupported Arrow type for Function signature: Blob v2 fields nested under "
+    "collection types are not supported"
+)


 _GRAMMAR_PRIMITIVES = (
@@ -495,6 +547,7 @@ _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"),
@@ -502,31 +555,258 @@ _GRAMMAR_PRIMITIVES = (


 def _canonical_arrow_type(data_type: pa.DataType) -> str:
-    """The server's V1 Function type grammar. Anything outside it is rejected
-    here rather than at registration."""
+    """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]:
    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}<{_canonical_list_item(data_type)}>"
+        return f"{prefix}<{item}>"
    if pa.types.is_fixed_size_list(data_type) and data_type.list_size > 0:
-        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}")
+        item = _grammar_list_item(data_type)
+        if item is not None:
+            return f"fixed_size_list<{item}, {data_type.list_size}>"
+    return None


-def _canonical_list_item(data_type: pa.DataType) -> str:
+def _grammar_list_item(data_type: pa.DataType) -> Optional[str]:
    """The grammar names only the item type; it always means a non-nullable
-    child called `item`, so any other child metadata cannot be represented."""
+    child called `item`, so other child properties require exact JSON."""
    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: list items must be a "
-            f"non-nullable field named 'item', got {child}"
+            "unsupported Arrow type for Function signature: field names "
+            "must not be empty"
        )
-    return _canonical_arrow_type(child.type)
+    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}"
+            )
+        metadata = {
+            (key.decode() if isinstance(key, bytes) else key): (
+                value.decode() if isinstance(value, bytes) else value
+            )
+            for key, value in (field.metadata or {}).items()
+        }
+        if metadata and metadata != {
+            _ARROW_EXTENSION_NAME_KEY: _BLOB_V2_EXTENSION_NAME
+        }:
+            raise TypeError(
+                "unsupported Arrow type for Function signature: lance.blob.v2 "
+                "field metadata must contain only its canonical extension marker"
+            )
+    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 _blob_storage_type(field: pa.Field) -> pa.DataType:
+    data_type = field.type
+    if isinstance(data_type, pa.ExtensionType):
+        return data_type.storage_type
+    return data_type
+
+
+def _exact_blob_storage_type(field: pa.Field) -> dict[str, Any]:
+    storage = _blob_storage_type(field)
+    if not pa.types.is_struct(storage):
+        raise TypeError(
+            "unsupported Arrow type for Function signature: lance.blob.v2 "
+            "requires struct storage"
+        )
+    return {
+        "type": "struct",
+        "fields": [
+            {
+                "name": child.name,
+                "nullable": child.nullable,
+                "type": (
+                    {"type": "large_binary"}
+                    if pa.types.is_large_binary(child.type)
+                    else _exact_arrow_type(child.type)
+                ),
+            }
+            for child in storage
+        ],
+    }
+
+
+def _data_type_has_blob_v2(data_type: pa.DataType) -> bool:
+    if pa.types.is_struct(data_type):
+        return any(
+            _is_blob_v2_field(field) or _data_type_has_blob_v2(field.type)
+            for field in data_type
+        )
+    if (
+        pa.types.is_list(data_type)
+        or pa.types.is_large_list(data_type)
+        or pa.types.is_fixed_size_list(data_type)
+    ):
+        field = data_type.value_field
+        return _is_blob_v2_field(field) or _data_type_has_blob_v2(field.type)
+    if pa.types.is_map(data_type):
+        return any(
+            _is_blob_v2_field(field) or _data_type_has_blob_v2(field.type)
+            for field in (data_type.key_field, data_type.item_field)
+        )
+    return False
+
+
+def _exact_arrow_field(
+    field: pa.Field, *, inside_collection: bool = False
+) -> dict[str, Any]:
+    _validate_exact_arrow_field(field)
+    if _is_blob_v2_field(field):
+        if inside_collection:
+            raise TypeError(_NESTED_BLOB_COLLECTION_ERROR)
+        return {
+            "name": field.name,
+            "nullable": field.nullable,
+            "type": _exact_blob_storage_type(field),
+            "metadata": {
+                _ARROW_EXTENSION_NAME_KEY: _BLOB_V2_EXTENSION_NAME,
+            },
+        }
+    value = {
+        "name": field.name,
+        "nullable": field.nullable,
+        "type": _exact_arrow_type(field.type, inside_collection=inside_collection),
+    }
+    return value
+
+
+def _exact_arrow_type(
+    data_type: pa.DataType, *, inside_collection: bool = False
+) -> 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, inside_collection=inside_collection)
+                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, inside_collection=True)
+            ],
+        }
+        if pa.types.is_fixed_size_list(data_type):
+            value["length"] = data_type.list_size
+        return value
+    if pa.types.is_map(data_type) and _data_type_has_blob_v2(data_type):
+        raise TypeError(_NESTED_BLOB_COLLECTION_ERROR)
+    raise TypeError(f"unsupported Arrow type for Function signature: {data_type}")


 def _list_of(item: pa.DataType) -> pa.DataType:
@@ -600,8 +880,15 @@ 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 pa.types.is_struct(output.type):
+    elif (
+        isinstance(output, pa.Field)
+        and not _is_blob_v2_field(output)
+        and pa.types.is_struct(output.type)
+    ):
+        _validate_exact_arrow_field(output)
        if output.nullable:
            raise ValueError("Function output must be non-nullable")
        fields = tuple(output.type)
@@ -617,18 +904,19 @@ 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_type(field.type),
+            arrow_type=_canonical_arrow_field(field),
            nullable=False,
        )

    if not fields:
        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")
@@ -637,8 +925,8 @@ def _function_output(output: pa.DataType | pa.Field | pa.Schema) -> FunctionOutp
        fields=tuple(
            FunctionResultField(
                name=field.name,
-                arrow_type=_canonical_arrow_type(field.type),
-                nullable=False,
+                arrow_type=_canonical_arrow_field(field),
+                nullable=field.nullable,
            )
            for field in fields
        ),
@@ -657,6 +945,10 @@ 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:
@@ -667,7 +959,7 @@ def _infer_signature(
        inputs = tuple(
            FunctionParameter(
                name=field.name,
-                arrow_type=_canonical_arrow_type(field.type),
+                arrow_type=_canonical_arrow_field(field),
                nullable=field.nullable,
            )
            for field in input_schema
@@ -690,7 +982,9 @@ def _infer_signature(
        inputs.append(
            FunctionParameter(
                name=parameter.name,
-                arrow_type=_canonical_arrow_type(data_type),
+                arrow_type=_canonical_arrow_field(
+                    pa.field(parameter.name, data_type, nullable=nullable)
+                ),
                nullable=nullable,
            )
        )
@@ -910,6 +1204,7 @@ class UdfDefinition:
        pip: tuple[str, ...],
        env: Mapping[str, str],
        python_version: Optional[str],
+        gpu: bool = False,
        conda: tuple[str, ...] = (),
        conda_channels: tuple[str, ...] = (),
    ):
@@ -938,12 +1233,14 @@ 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",
+            kind="python_v2" if gpu_marker is not None else "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(
@@ -989,6 +1286,7 @@ 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]: ...
@@ -1003,6 +1301,7 @@ 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] = (),
 ):
@@ -1010,8 +1309,9 @@ def udf(

    Input and output signatures are inferred from supported annotations. For
    Arrow types annotations cannot express precisely, pass ``input_schema``
-    and ``output_schema`` together. Nullable outputs are rejected because V1
-    uses physical NULL to represent unassigned computed-column rows.
+    and ``output_schema`` together. Scalar outputs must be non-nullable. Every
+    named-struct field may be nullable; Enterprise preserves the struct's
+    validity when the result is expanded into sibling columns.

    Parameters
    ----------
@@ -1023,8 +1323,8 @@ def udf(
        Explicit input fields in the exact order of the callable parameters.
        Must be provided together with ``output_schema``.
    output_schema : pyarrow.DataType, pyarrow.Field, or pyarrow.Schema, optional
-        Explicit scalar or named-struct output. Must be non-nullable and be
-        provided together with ``input_schema``.
+        Explicit scalar or named-struct output. Scalar outputs must be
+        non-nullable. Must be provided together with ``input_schema``.
    pip : sequence of str, optional
        Pip requirements for the remote environment.
    conda : sequence of str, optional
@@ -1035,6 +1335,10 @@ 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
@@ -1059,6 +1363,11 @@ 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:
@@ -1070,6 +1379,7 @@ 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),
        )
@@ -1080,6 +1390,7 @@ def udf(


 __all__ = [
+    "AssignmentMapping",
    "ApplicationInput",
    "FunctionApplication",
    "FunctionArtifact",
@@ -751,7 +751,7 @@ class IvfPq:
        This value controls how much the vector is compressed during the
        quantization step.  The more sub vectors there are the less the vector is
        compressed.  The default is the dimension of the vector divided by 16.  If
-        the dimension is not evenly divisible by 16 we use the dimension divded by
+        the dimension is not evenly divisible by 16 we use the dimension divided by
        8.

        The above two cases are highly preferred.  Having 8 or 16 values per
@@ -4,15 +4,27 @@
 """Handles to operations a server may run asynchronously."""

 import asyncio
+import json
 from datetime import timedelta
 from typing import Any, Callable, Generic, Optional, TypeVar, cast

+import pyarrow as pa
+
 from lancedb.background_loop import LOOP

 from . import _lancedb
+from ._lancedb import JobDescription, JobFailureInfo, JobInfo

 T = TypeVar("T")

+__all__ = [
+    "AsyncJob",
+    "Job",
+    "JobDescription",
+    "JobFailureInfo",
+    "JobInfo",
+]
+

 class AsyncJob(Generic[T]):
    """A handle to an operation that may still be running.
@@ -78,6 +90,149 @@ class AsyncJob(Generic[T]):
            return
        await self._inner.cancel()

+    async def refresh(self) -> None:
+        """Ask the backend for this job's current state, and for a server-side
+        job its full record, then cache it for the properties below.
+
+        The properties are all `None` until this runs, because submitting an
+        operation returns only a job id. `status` fetches the whole record too;
+        `wait` records only the terminal state it establishes.
+        """
+        if self._inner is None:
+            return
+        await self._inner.refresh()
+
+    @property
+    def state(self) -> Optional[str]:
+        """The last observed lifecycle state, without contacting the backend.
+
+        `None` until the handle has talked to it. See :meth:`AsyncJob.refresh`.
+        """
+        if self._inner is None:
+            return "finished"
+        return self._inner._state
+
+    @property
+    def job_type(self) -> Optional[str]:
+        """The job's type, as the server names it.
+
+        `None` for an in-process job, which has no server-side record.
+        """
+        return self._field("job_type")
+
+    @property
+    def creation_ms(self) -> Optional[int]:
+        """When the job was created, in milliseconds since the epoch."""
+        return self._field("creation_ms")
+
+    @property
+    def spec(self) -> Optional[Any]:
+        """The job-type-specific specification it was submitted with."""
+        return self._field("spec")
+
+    @property
+    def result(self) -> Optional[Any]:
+        """The job-type-specific terminal result, as reported data rather than
+        the typed model :meth:`AsyncJob.wait` returns.
+
+        `None` until the job succeeds, so a job that never terminates reports
+        its progress through :meth:`AsyncJob.events` instead.
+        """
+        return self._field("result")
+
+    @property
+    def failure(self) -> Optional[JobFailureInfo]:
+        """Why the job failed, when it failed and the server reports a reason."""
+        return self._field("failure")
+
+    @property
+    def _spec_json(self) -> Optional[str]:
+        return self._field("_spec_json")
+
+    @property
+    def _result_json(self) -> Optional[str]:
+        return self._field("_result_json")
+
+    def _field(self, name: str) -> Optional[Any]:
+        description = self._inner._description if self._inner is not None else None
+        return getattr(description, name) if description is not None else None
+
+    async def events(
+        self,
+        *,
+        limit: Optional[int] = None,
+        filter: Optional[str] = None,
+    ) -> "pa.Table":
+        """This job's recorded lifecycle events.
+
+        Where the properties above report a terminal result only once the job
+        reaches one, events are written as the job runs and outlive the workers
+        that produced them. A distributed job records a `claim`/`claim_complete`
+        pair per unit of work, each carrying `rows_processed`, so a job that
+        never finishes still accounts for what it did.
+
+        Parameters
+        ----------
+        limit: int, optional
+            Maximum event rows to return. The server caps results at 1000 by
+            default and 10,000 at most, and truncates without saying so, so
+            pass this for a job that emits an event per fragment.
+        filter: str, optional
+            SQL-like expression over the `state`, `updated_by`, `emitted_from`,
+            `emitted_by`, and `claim_entity` columns, such as
+            ``state = 'claim_complete'``.
+        """
+        if self._inner is None:
+            raise NotImplementedError(
+                "job event history is only available for server-side jobs"
+            )
+        return await self._inner.events(limit=limit, filter=filter)
+
+    def __repr__(self) -> str:
+        return _job_repr("AsyncJob", self)
+
+
+_REPR_INDENT = " " * 4
+
+
+def _repr_payload(value: Any) -> str:
+    """Render a job payload as indented JSON, aligned under its field."""
+    try:
+        rendered = json.dumps(value, indent=4)
+    except TypeError:
+        return repr(value)
+    return rendered.replace("\n", "\n" + _REPR_INDENT)
+
+
+def _job_repr(kind: str, job: Any) -> str:
+    """Render every field the handle currently knows, omitting the rest.
+
+    One field per line, with the JSON payloads indented, because a refresh
+    job's spec and result are the point of printing it.
+    """
+    state = job.state
+    if state is None:
+        # Nothing has been fetched yet, so there is nothing to lay out.
+        known = f"id={job.id!r}, " if job.id is not None else ""
+        return f"{kind}({known}not refreshed)"
+
+    fields = []
+    if job.id is not None:
+        fields.append(f"id={job.id!r}")
+    fields.append(f"state={state!r}")
+    for name in ("job_type", "creation_ms"):
+        value = getattr(job, name)
+        if value is not None:
+            fields.append(f"{name}={value!r}")
+    for name in ("spec", "result"):
+        value = getattr(job, name)
+        if value is not None:
+            fields.append(f"{name}={_repr_payload(value)}")
+    if job.failure is not None:
+        fields.append(f"failure={job.failure!r}")
+    body = "".join(f"\n{_REPR_INDENT}{field}," for field in fields)
+    return f"{kind}({body}\n)"
+

 class Job(Generic[T]):
    """Synchronous counterpart of `AsyncJob` with the same result type."""
@@ -122,6 +277,75 @@ class Job(Generic[T]):
            return
        LOOP.run(self._inner.cancel())

+    def refresh(self) -> None:
+        """Ask the backend for this job's current state and record.
+
+        See :meth:`AsyncJob.refresh`.
+        """
+        if self._inner is None:
+            return
+        LOOP.run(self._inner.refresh())
+
+    @property
+    def state(self) -> Optional[str]:
+        """The last observed lifecycle state. See :attr:`AsyncJob.state`."""
+        return self._inner.state if self._inner is not None else "finished"
+
+    @property
+    def job_type(self) -> Optional[str]:
+        """The job's type. See :attr:`AsyncJob.job_type`."""
+        return self._field("job_type")
+
+    @property
+    def creation_ms(self) -> Optional[int]:
+        """When the job was created. See :attr:`AsyncJob.creation_ms`."""
+        return self._field("creation_ms")
+
+    @property
+    def spec(self) -> Optional[Any]:
+        """The job's specification. See :attr:`AsyncJob.spec`."""
+        return self._field("spec")
+
+    @property
+    def result(self) -> Optional[Any]:
+        """The job's terminal result. See :attr:`AsyncJob.result`."""
+        return self._field("result")
+
+    @property
+    def failure(self) -> Optional[JobFailureInfo]:
+        """Why the job failed. See :attr:`AsyncJob.failure`."""
+        return self._field("failure")
+
+    @property
+    def _spec_json(self) -> Optional[str]:
+        return self._field("_spec_json")
+
+    @property
+    def _result_json(self) -> Optional[str]:
+        return self._field("_result_json")
+
+    def _field(self, name: str) -> Optional[Any]:
+        return getattr(self._inner, name) if self._inner is not None else None
+
+    def events(
+        self,
+        *,
+        limit: Optional[int] = None,
+        filter: Optional[str] = None,
+    ) -> "pa.Table":
+        """This job's recorded lifecycle events.
+
+        See :meth:`AsyncJob.events`.
+        """
+        if self._inner is None:
+            raise NotImplementedError(
+                "job event history is only available for server-side jobs"
+            )
+        return LOOP.run(self._inner.events(limit=limit, filter=filter))
+
+    def __repr__(self) -> str:
+        return _job_repr("Job", self)
+

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


@@ -12,6 +12,7 @@ from __future__ import annotations

 import sys
 from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Optional, Union
+from uuid import UUID

 if sys.version_info >= (3, 12):
    from typing import override
@@ -48,8 +49,11 @@ from lancedb._lancedb import (
    connect_namespace_client as _connect_namespace_client,
 )
 from lancedb.background_loop import LOOP
+from lancedb.arrow import AsyncRecordBatchReader
 from lancedb.db import AsyncConnection, DBConnection
 from lancedb.job import AsyncJob, Job
+from lancedb.sql import AsyncQuery as AsyncSqlQuery
+from lancedb.sql import QueryDescription
 from lance_namespace import (
    LanceNamespace,
    connect as namespace_connect,
@@ -1447,6 +1451,37 @@ class AsyncLanceNamespaceDBConnection:
            namespace_path=namespace_path, page_token=page_token, limit=limit
        )

+    async def execute_query(
+        self,
+        query: str,
+        *,
+        default_namespace_path: Optional[List[str]] = None,
+    ) -> AsyncRecordBatchReader:
+        """Execute SQL when supported by the underlying connection."""
+        return await self._inner.execute_query(
+            query,
+            default_namespace_path=default_namespace_path,
+        )
+
+    async def execute_query_async(
+        self,
+        query: str,
+        *,
+        default_namespace_path: Optional[List[str]] = None,
+    ) -> AsyncSqlQuery:
+        """Start executing SQL when supported by the underlying connection.
+
+        Namespace-backed local connections do not support SQL.
+        """
+        return await self._inner.execute_query_async(
+            query,
+            default_namespace_path=default_namespace_path,
+        )
+
+    async def describe_query(self, query_id: UUID) -> QueryDescription:
+        """Describe a submitted SQL query when supported."""
+        return await self._inner.describe_query(query_id)
+
    async def namespace_client(self) -> LanceNamespace:
        """Get the namespace client for this connection.

@@ -78,6 +78,10 @@ if TYPE_CHECKING:
 T = TypeVar("T", bound="LanceModel")
 AnalyzePlanDistributedMetrics = Literal["aggregate", "per_worker", "full"]

+# Number of rows a hybrid query returns when no limit was set on it. This
+# mirrors the default the Rust query builder applies to its sub-queries.
+DEFAULT_HYBRID_LIMIT = 10
+

@runtime_checkable
 class _LanceScanner(Protocol):
@@ -109,6 +113,7 @@ 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
    )
@@ -804,6 +809,10 @@ 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

@@ -825,6 +834,7 @@ 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
@@ -853,7 +863,7 @@ class Query(pydantic.BaseModel):
        return query

    # This tells pydantic to allow custom types (needed for the `vector` query since
-    # pa.Array wouln't be allowed otherwise)
+    # pa.Array wouldn't be allowed otherwise)
    model_config = pydantic.ConfigDict(arbitrary_types_allowed=True)


@@ -3887,14 +3897,54 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):

        return self

+    def _create_child_queries(
+        self,
+    ) -> Tuple["AsyncFTSQuery", "AsyncVectorQuery", int, int]:
+        """Build the sub-queries that make up this hybrid query.
+
+        Execution, `explain_plan` and `analyze_plan` all go through here so that
+        the plans that are reported are the plans that actually run.
+
+        Returns the two sub-queries along with the effective limit and offset of
+        the hybrid query itself.
+        """
+        fts_query = AsyncFTSQuery(self._inner.to_fts_query(), self._table)
+        vec_query = AsyncVectorQuery(self._inner.to_vector_query(), self._table)
+
+        fts_req = fts_query._inner.to_query_request()
+        vec_req = vec_query._inner.to_query_request()
+
+        # Only one of the two sub-queries carries the limit when it was never
+        # set explicitly: nearest_to()/nearest_to_text() build the sibling query
+        # from scratch, and that is where the default gets filled in. Which one
+        # that is depends on the order the hybrid query was built in, so look at
+        # both rather than at a single side.
+        limit = fts_req.limit if fts_req.limit is not None else vec_req.limit
+        if limit is None:
+            limit = DEFAULT_HYBRID_LIMIT
+        offset = fts_req.offset or vec_req.offset or 0
+
+        fts_query.with_row_id()
+        vec_query.with_row_id()
+
+        # offset() pushes the offset down into both sub-queries, which would make
+        # each of them skip its own first `offset` rows. The window has to be
+        # taken out of the combined, reranked results instead, so fetch the
+        # skipped prefix here too and slice it off afterwards.
+        fts_query.limit(limit + offset)
+        vec_query.limit(limit + offset)
+        fts_query.offset(0)
+        vec_query.offset(0)
+
+        return fts_query, vec_query, limit, offset
+
    async def to_batches(
        self,
        *,
        max_batch_length: Optional[int] = None,
        timeout: Optional[timedelta] = None,
    ) -> AsyncRecordBatchReader:
-        fts_query = AsyncFTSQuery(self._inner.to_fts_query(), self._table)
-        vec_query = AsyncVectorQuery(self._inner.to_vector_query(), self._table)
+        fts_query, vec_query, limit, offset = self._create_child_queries()

        req = fts_query._inner.to_query_request()
        blob_auto_row_id = False
@@ -3914,9 +3964,6 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
        self._blob_auto_row_id = blob_auto_row_id
        self._blob_paths = blob_paths

-        fts_query.with_row_id()
-        vec_query.with_row_id()
-
        fts_results, vector_results = await asyncio.gather(
            fts_query.to_arrow(timeout=timeout),
            vec_query.to_arrow(timeout=timeout),
@@ -3928,8 +3975,9 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
            norm=self._norm,
            fts_query=fts_query.get_query(),
            reranker=self._reranker,
-            limit=self._inner.get_limit(),
+            limit=limit,
            with_row_ids=True,
+            offset=offset,
        )
        if (
            not self._user_requested_row_id()
@@ -3958,14 +4006,14 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
        ...     print(plan)
        >>> asyncio.run(doctest_example()) # doctest: +ELLIPSIS, +NORMALIZE_WHITESPACE
        RRFReranker(K=60)
-            ProjectionExec: expr=[vector@0 as vector, text@3 as text, _distance@2 as _distance]
+            ProjectionExec: expr=[vector@0 as vector, text@3 as text, _distance@2 as _distance, _rowid@1 as _rowid]
              LanceRead: uri=..., projection=[text], source=stream(_rowid)
                GlobalLimitExec: skip=0, fetch=10
                  FilterExec: _distance@2 IS NOT NULL
                    SortExec: TopK(fetch=10), expr=[_distance@2 ASC NULLS LAST, _rowid@1 ASC NULLS LAST], preserve_partitioning=[false]
                      KNNVectorDistance: metric=l2
                        LanceRead: uri=..., projection=[vector], ...
-            ProjectionExec: expr=[vector@2 as vector, text@3 as text, _score@1 as _score]
+            ProjectionExec: expr=[vector@2 as vector, text@3 as text, _score@1 as _score, _rowid@0 as _rowid]
              LanceRead: uri=..., projection=[vector, text], source=stream(_rowid)
                GlobalLimitExec: skip=0, fetch=10
                  MatchQuery: column=text, query=[hello]
@@ -3980,8 +4028,9 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
        plan : str
        """  # noqa: E501

-        vector_plan = await self._inner.to_vector_query().explain_plan(verbose)
-        fts_plan = await self._inner.to_fts_query().explain_plan(verbose)
+        fts_query, vec_query, _, _ = self._create_child_queries()
+        vector_plan = await vec_query.explain_plan(verbose)
+        fts_plan = await fts_query.explain_plan(verbose)
        # Indent sub-plans under the reranker
        indented_vector = "\n".join("  " + line for line in vector_plan.splitlines())
        indented_fts = "\n".join("  " + line for line in fts_plan.splitlines())
@@ -4008,14 +4057,12 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
        -------
        plan : str
        """
+        fts_query, vec_query, _, _ = self._create_child_queries()
+
        results = ["Vector Search Query:"]
-        results.append(
-            await self._inner.to_vector_query().analyze_plan(distributed_metrics)
-        )
+        results.append(await vec_query.analyze_plan(distributed_metrics))
        results.append("FTS Search Query:")
-        results.append(
-            await self._inner.to_fts_query().analyze_plan(distributed_metrics)
-        )
+        results.append(await fts_query.analyze_plan(distributed_metrics))

        return "\n".join(results)

@@ -9,6 +9,7 @@ from concurrent.futures import ThreadPoolExecutor
 import sys
 from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Optional, Union
 from urllib.parse import urlparse
+from uuid import UUID
 import warnings

 if sys.version_info >= (3, 12):
@@ -25,10 +26,12 @@ from ..common import DATA
 from ..db import DBConnection, LOOP
 from ..functions import FunctionVersion, UdfDefinition
 from ..job import AsyncJob, Job
+from ..sql import Query as SqlQuery
+from ..sql import QueryDescription
 from ..materialized_view import MaterializedView, SelectArg

 if TYPE_CHECKING:
-    from .._lancedb import JobDescription, JobInfo
+    from .._lancedb import JobInfo
 from ..embeddings import EmbeddingFunctionConfig
 from lance_namespace import (
    LanceNamespace,
@@ -116,6 +119,7 @@ class RemoteDBConnection(DBConnection):
        read_timeout: Optional[float] = None,
        storage_options: Optional[Dict[str, str]] = None,
        read_consistency_interval: Optional[timedelta] = None,
+        sql_host_override: Optional[str] = None,
    ):
        """Connect to a remote LanceDB database."""
        if isinstance(client_config, dict):
@@ -161,6 +165,7 @@ class RemoteDBConnection(DBConnection):
        self.api_key = api_key
        self.region = region
        self.host_override = host_override
+        self.sql_host_override = sql_host_override
        self.storage_options = storage_options
        self.db_name = parsed.netloc

@@ -175,6 +180,7 @@ class RemoteDBConnection(DBConnection):
                api_key=api_key,
                region=region,
                host_override=host_override,
+                sql_host_override=sql_host_override,
                client_config=client_config,
                storage_options=storage_options,
                read_consistency_interval=read_consistency_interval,
@@ -193,6 +199,7 @@ class RemoteDBConnection(DBConnection):
                "api_key": self.api_key,
                "region": self.region,
                "host_override": self.host_override,
+                "sql_host_override": self.sql_host_override,
                "client_config": _client_config_to_dict(self.client_config),
                "storage_options": self.storage_options,
            }
@@ -732,14 +739,11 @@ class RemoteDBConnection(DBConnection):
        )

    @override
-    def job(self, job_id: str) -> Job:
-        """A [Job][lancedb.job.Job] handle for a server-side job by id.
-
-        The handle is constructed without a server round trip; an unknown id
-        surfaces when the handle is used. Dropping the handle has no effect
-        on the job itself.
+    def open_job(self, job_id: str) -> Job:
+        """Open a server-side job by id. See
+        [DBConnection.open_job][lancedb.db.DBConnection.open_job].
        """
-        return Job(self._conn.job(job_id))
+        return Job(LOOP.run(self._conn.open_job(job_id)))

    @override
    def create_function_async(self, definition: UdfDefinition) -> Job[FunctionVersion]:
@@ -749,19 +753,19 @@ class RemoteDBConnection(DBConnection):
    def get_function(self, name: str, *, version: str) -> FunctionVersion:
        return LOOP.run(self._conn.get_function(name, version=version))

+    @override
+    def list_functions(self) -> List[FunctionVersion]:
+        return LOOP.run(self._conn.list_functions())
+
+    @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."""
        return LOOP.run(self._conn.list_jobs())

-    @override
-    def get_job(self, job_id: str) -> Optional["JobDescription"]:
-        """Describe a single server-side job by id.
-
-        Returns None when the server has no such job.
-        """
-        return LOOP.run(self._conn.get_job(job_id))
-
    @override
    def cancel_job(self, job_id: str) -> bool:
        """Request cancellation of a server-side job by id.
@@ -773,12 +777,35 @@ class RemoteDBConnection(DBConnection):
        return LOOP.run(self._conn.cancel_job(job_id))

    @override
-    def job_history(self, job_id: Optional[str] = None) -> List[pa.RecordBatch]:
-        """The lifecycle event history of a server-side job, as Arrow batches.
+    def execute_query_async(
+        self,
+        query: str,
+        *,
+        default_namespace_path: Optional[List[str]] = None,
+    ) -> SqlQuery:
+        """Start executing SQL through this remote connection.

-        Lists history across all jobs when `job_id` is None.
+        Unqualified tables use this connection's database and the
+        ``["public"]`` namespace by default. Fully qualified table names may
+        reference other databases available to the same deployment.
        """
-        return LOOP.run(self._conn.job_history(job_id))
+        return SqlQuery(
+            LOOP.run(
+                self._conn.execute_query_async(
+                    query,
+                    default_namespace_path=default_namespace_path,
+                )
+            )
+        )
+
+    @override
+    def describe_query(self, query_id: UUID) -> QueryDescription:
+        """Describe a submitted SQL query by its connection-scoped id."""
+        return LOOP.run(
+            self._conn.describe_query(
+                query_id,
+            )
+        )

    @override
    def namespace_client(self) -> LanceNamespace:
@@ -177,4 +177,7 @@ class OAuthProvider(HeaderProvider):
        if not self._current_token:
            raise RuntimeError("Failed to obtain OAuth token")

-        return {"Authorization": f"Bearer {self._current_token}"}
+        return {
+            "Authorization": f"Bearer {self._current_token}",
+            "x-lancedb-credential-type": "oidc",
+        }
@@ -67,7 +67,15 @@ 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


@@ -540,6 +548,7 @@ class RemoteTable(Table):
        LOOP.run(
            self._table.create_index(
                column,
+                replace=replace,
                config=config,
                wait_timeout=wait_timeout,
                name=name,
@@ -711,7 +720,7 @@ class RemoteTable(Table):
        Parameters
        ----------
        query: list/np.ndarray/str/PIL.Image.Image, default None
-            The targetted vector to search for.
+            The targeted vector to search for.

            - *default None*.
            Acceptable types are: list, np.ndarray, PIL.Image.Image
@@ -175,7 +175,7 @@ class Reranker(ABC):
            if the results haven't been executed yet or the results in arrow format.
        query : str or None,
            The input query. Some rerankers might not need the query to rerank.
-            In that case, it can be set to None explicitly. This is inteded to
+            In that case, it can be set to None explicitly. This is intended to
            be handled by the reranker implementations.
        deduplicate : bool, optional
            Whether to deduplicate the results based on the `_rowid` column,
@@ -0,0 +1,88 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+"""Handles to SQL queries running on a remote database."""
+
+from uuid import UUID
+
+import pyarrow as pa
+
+from lancedb.background_loop import LOOP
+
+from . import _lancedb
+from .arrow import AsyncRecordBatchReader
+
+QueryDescription = _lancedb.QueryDescription
+
+
+class AsyncQuery:
+    """A handle to a submitted SQL query on an asynchronous connection."""
+
+    def __init__(self, inner: "_lancedb.SqlQuery"):
+        self._inner = inner
+
+    @property
+    def id(self) -> UUID:
+        """The stable identifier scoped to the connection that submitted it."""
+        return self._inner.id
+
+    async def describe(self) -> QueryDescription:
+        """Get a point-in-time description of the query."""
+        return await self._inner.describe()
+
+    async def reader(self) -> AsyncRecordBatchReader:
+        """Wait for the initial result stream and return its Arrow reader.
+
+        Results are single-consumer. Calling this method more than once on the
+        same query raises an error. Later batches are streamed as they become
+        available without waiting for the full query to finish.
+        """
+        return AsyncRecordBatchReader(await self._inner.reader())
+
+    async def cancel(self) -> None:
+        """Request cancellation of the query."""
+        await self._inner.cancel()
+
+
+class Query:
+    """Synchronous counterpart of :class:`AsyncQuery`."""
+
+    def __init__(self, inner: AsyncQuery):
+        self._inner = inner
+
+    @property
+    def id(self) -> UUID:
+        """The stable identifier scoped to the connection that submitted it."""
+        return self._inner.id
+
+    def describe(self) -> QueryDescription:
+        """Get a point-in-time description of the query."""
+        return LOOP.run(self._inner.describe())
+
+    def reader(self) -> pa.RecordBatchReader:
+        """Wait for the initial result stream and return a blocking reader.
+
+        Results are single-consumer. Calling this method more than once on the
+        same query raises an error. Later batches block only until they become
+        available, without waiting for the full query to finish.
+        """
+        reader = LOOP.run(self._inner.reader())
+
+        def next_batch():
+            try:
+                return LOOP.run(reader.__anext__())
+            except StopAsyncIteration:
+                return None
+
+        def batches():
+            while (batch := next_batch()) is not None:
+                yield batch
+
+        return pa.RecordBatchReader.from_batches(reader.schema, batches())
+
+    def cancel(self) -> None:
+        """Request cancellation of the query."""
+        LOOP.run(self._inner.cancel())
+
+
+__all__ = ["AsyncQuery", "Query", "QueryDescription"]
@@ -1547,7 +1547,9 @@ class Table(ABC):
        on: Union[str, Iterable[str]]
            A column (or columns) to join on.  This is how records from the
            source table and target table are matched.  Typically this is some
-            kind of key or id column.
+            kind of key or id column.  Passing several columns matches on the
+            composite key: a source row updates a target row only when it
+            agrees on every one of them.

        Examples
        --------
@@ -1617,7 +1619,7 @@ class Table(ABC):
        Parameters
        ----------
        query: list/np.ndarray/str/PIL.Image.Image, default None
-            The targetted vector to search for.
+            The targeted vector to search for.

            - *default None*.
            Acceptable types are: list, np.ndarray, PIL.Image.Image
@@ -1678,9 +1680,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.  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.
+        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.

        Parameters
        ----------
@@ -1791,6 +1793,9 @@ class Table(ABC):
        The result has the same length and order as ``row_ids``. Null blobs
        produce null slots; valid empty blobs produce ``b""``.

+        ``_rowid`` values stay valid after compaction when the table has stable
+        row ids.
+
        Convenience for small payloads. For large values use
        :meth:`fetch_blob_files`.
        """
@@ -1808,6 +1813,9 @@ class Table(ABC):
        The result has the same length and order as ``requests``; null blobs
        produce null slots and empty ranges on non-null blobs produce ``b""``.

+        ``_rowid`` values stay valid after compaction when the table has stable
+        row ids.
+
        Row IDs can be obtained from a query with ``with_row_id(True)``. This
        API is currently supported only by local tables.
        """
@@ -1823,6 +1831,9 @@ class Table(ABC):
        ``_rowid`` or a ``_lance_row_id`` field on the blob descriptor. Null
        rows are ``None``. Remote tables require LanceDB Cloud server 0.5.0 or
        newer.
+
+        ``_rowid`` values stay valid after compaction when the table has stable
+        row ids.
        """

    @abstractmethod
@@ -2165,9 +2176,11 @@ class Table(ABC):
            Function columns are supported only on LanceDB Cloud and
            Enterprise.
        computed: Dict[str, str], optional
-            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.
+            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.

            Unlike ``transforms``, the expression is stored rather than
            evaluated now: the column is committed with no values, and rows get
@@ -3828,7 +3841,7 @@ class LanceTable(Table):
        Parameters
        ----------
        query: list/np.ndarray/str/PIL.Image.Image, default None
-            The targetted vector to search for.
+            The targeted vector to search for.

            - *default None*.
            Acceptable types are: list, np.ndarray, PIL.Image.Image
@@ -4088,6 +4101,7 @@ 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

@@ -5624,7 +5638,7 @@ class AsyncTable:
        if fill_value is None:
            fill_value = 0.0

-        # _santitize_data is an old code path, but we will use it until the
+        # _sanitize_data is an old code path, but we will use it until the
        # new code path is ready.
        if mode == "overwrite":
            # For overwrite, apply the same preprocessing as create_table
@@ -5698,7 +5712,9 @@ class AsyncTable:
        on: Union[str, Iterable[str]]
            A column (or columns) to join on.  This is how records from the
            source table and target table are matched.  Typically this is some
-            kind of key or id column.
+            kind of key or id column.  Passing several columns matches on the
+            composite key: a source row updates a target row only when it
+            agrees on every one of them.

        Examples
        --------
@@ -5798,7 +5814,7 @@ class AsyncTable:
        Parameters
        ----------
        query: list/np.ndarray/str/PIL.Image.Image, default None
-            The targetted vector to search for.
+            The targeted vector to search for.

            - *default None*.
            Acceptable types are: list, np.ndarray, PIL.Image.Image
@@ -5981,7 +5997,23 @@ class AsyncTable:

    def _sync_query_to_async(
        self, query: Query
-    ) -> AsyncHybridQuery | AsyncFTSQuery | AsyncVectorQuery | AsyncQuery:
+    ) -> (
+        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
+
        async_query = self.query()
        if query.limit is not None:
            async_query = async_query.limit(query.limit)
@@ -6046,6 +6078,7 @@ 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

@@ -6268,8 +6301,11 @@ class AsyncTable:
            Function columns are supported only on LanceDB Cloud and
            Enterprise.
        computed: Dict[str, str], optional
-            A map of column name to a SQL expression defining the column. The
-            column's type and inputs are derived from the expression.
+            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.

            Unlike ``transforms``, the expression is stored rather than
            evaluated now: the column is committed with no values, and rows get
@@ -6540,6 +6576,9 @@ 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]
@@ -66,6 +66,25 @@ def _row_ids_by_id(table):
    return dict(zip(hits["id"].to_pylist(), hits["_rowid"].to_pylist()))


+def _assert_missing_blob_row_ids(exc_info):
+    message = str(exc_info.value)
+    assert "row ids" in message
+    assert "rowaddr" not in message
+    assert "fragment" not in message
+
+
+def _assert_fetch_apis_reject_missing_row_ids(table, row_ids):
+    with pytest.raises(ValueError) as exc_info:
+        table.fetch_blobs("image", row_ids)
+    _assert_missing_blob_row_ids(exc_info)
+    with pytest.raises(ValueError) as exc_info:
+        table.fetch_blob_files("image", row_ids)
+    _assert_missing_blob_row_ids(exc_info)
+    with pytest.raises(ValueError) as exc_info:
+        table.fetch_blob_ranges("image", [(row_id, 0, 1) for row_id in row_ids])
+    _assert_missing_blob_row_ids(exc_info)
+
+
 def test_blob_factory_declares_v2_field():
    field = lancedb.blob("image")
    assert isinstance(field.type, pa.ExtensionType)
@@ -278,7 +297,10 @@ def test_blob_v2_projection_sources_use_typed_column_name():


 def _legacy_v1_table(name):
-    db = lancedb.connect("memory:///")
+    # Legacy v1 blob columns are only writable at file version <= 2.1.
+    db = lancedb.connect(
+        "memory:///", storage_options={"new_table_data_storage_version": "2.1"}
+    )
    schema = pa.schema(
        [
            pa.field("id", pa.int64()),
@@ -691,6 +713,25 @@ def test_fetch_blobs_accepts_query_result():
    assert {blobs[i].as_py() for i in range(len(blobs))} == {b"gamma"}


+def test_fetch_blobs_after_compact_with_stable_row_ids(tmp_path):
+    db = lancedb.connect(
+        tmp_path, storage_options={"new_table_enable_stable_row_ids": "true"}
+    )
+    schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("image")])
+    table = db.create_table("t", schema=schema)
+    table.add([{"id": 1, "image": b"frag-one"}])
+    table.add([{"id": 2, "image": b"frag-two"}])
+    by_id = _row_ids_by_id(table)
+    ids = [by_id[1], by_id[2]]
+
+    table.optimize()
+
+    blobs = table.fetch_blobs("image", ids)
+    assert blobs.to_pylist() == [b"frag-one", b"frag-two"]
+    ranges = table.fetch_blob_ranges("image", [(ids[0], 5, 3), (ids[1], 5, 3)])
+    assert ranges.to_pylist() == [b"one", b"two"]
+
+
 def test_fetch_blobs_preserves_null_and_empty_values():
    table = _blob_table(
        "nulls",
@@ -739,8 +780,25 @@ def test_fetch_blob_ranges_validates_requests():
    with pytest.raises(ValueError, match="offset \\+ length overflowed"):
        table.fetch_blob_ranges("image", [(row_id, 2**64 - 1, 1)])

-    with pytest.raises(ValueError, match="row IDs"):
+    with pytest.raises(ValueError) as exc_info:
        table.fetch_blob_ranges("image", [(2**64 - 1, 0, 1)])
+    _assert_missing_blob_row_ids(exc_info)
+
+
+def test_fetch_blob_apis_reject_missing_fragment_row_addr():
+    table = _blob_table("missing_frag", [{"id": 1, "image": b"x"}])
+    live = _row_ids_by_id(table)[1]
+    _assert_fetch_apis_reject_missing_row_ids(table, [1 << 32, live])
+
+
+def test_fetch_blob_apis_reject_deleted_row_ids():
+    table = _blob_table(
+        "deleted_rows",
+        [{"id": 1, "image": b"one"}, {"id": 2, "image": b"two"}],
+    )
+    by_id = _row_ids_by_id(table)
+    table.delete("id = 2")
+    _assert_fetch_apis_reject_missing_row_ids(table, [by_id[2], by_id[1]])


 def test_fetch_blob_ranges_empty_requests_returns_empty_array():
@@ -327,8 +327,8 @@ def test_embedding_function_with_pandas(tmp_path):
        ) -> List[np.array]:
            return [np.random.randn(self.ndims()).tolist() for _ in range(len(texts))]

-    registery = get_registry()
-    func = registery.get("mock-embedding").create()
+    registry = get_registry()
+    func = registry.get("mock-embedding").create()

    class TestSchema(LanceModel):
        text: str = func.SourceField()
@@ -394,9 +394,9 @@ def test_multiple_embeddings_for_pandas(tmp_path):
        ) -> List[np.array]:
            return [np.random.randn(self.ndims()).tolist() for _ in range(len(texts))]

-    registery = get_registry()
-    func1 = registery.get("mock-embedding").create()
-    func2 = registery.get("mock-embedding2").create()
+    registry = get_registry()
+    func1 = registry.get("mock-embedding").create()
+    func2 = registry.get("mock-embedding2").create()

    class TestSchema(LanceModel):
        text: str = func1.SourceField()
@@ -12,6 +12,8 @@ from datetime import date
 import http.server
 import json
 from pathlib import Path
+import subprocess
+import sys
 import threading
 from typing import Optional

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

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

 THRESHOLD = 20
 _CACHE = None
@@ -61,6 +69,80 @@ 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)
@@ -89,6 +171,58 @@ 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:
@@ -168,9 +302,7 @@ def test_udf_resolves_module_globals_before_builtins(tmp_path):
        udf(module.uses_callable_shadow)


-def test_canonical_arrow_type_is_exactly_the_grammar():
-    from lancedb.functions import _GRAMMAR_PRIMITIVES, _canonical_arrow_type
-
+def test_canonical_arrow_type_prefers_the_compact_grammar():
    golden = json.loads(
        (
            Path(__file__).parents[3]
@@ -181,14 +313,19 @@ def test_canonical_arrow_type_is_exactly_the_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")),
    ]:
@@ -378,14 +515,27 @@ def test_udf_recursion_versus_a_rebound_module_name(tmp_path):
        udf(raw_fact)


-def test_canonical_arrow_type_rejects_unrepresentable_list_children():
-    from lancedb.functions import _canonical_arrow_type
-
+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"
    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)
@@ -395,6 +545,29 @@ def test_canonical_arrow_type_rejects_unrepresentable_list_children():
        )
        == "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:
@@ -432,6 +605,7 @@ 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(),
@@ -448,8 +622,6 @@ 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"]
@@ -482,6 +654,432 @@ 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_whole_named_struct_function_can_include_a_blob_result_field():
+    @udf(
+        input_schema=pa.schema([lancedb.blob("image", nullable=False)]),
+        output_schema=pa.field(
+            "payload",
+            pa.struct(
+                [
+                    pa.field("mime_type", pa.string(), nullable=False),
+                    lancedb.blob("image", nullable=False),
+                ]
+            ),
+            nullable=False,
+        ),
+    )
+    def inspect_blob(image):
+        return {"mime_type": "image/png", "image": image}
+
+    output = inspect_blob.registration_request.signature.output
+    assert output.kind == "named_struct"
+    assert [(field.name, field.arrow_type) for field in output.fields] == [
+        ("mime_type", "utf8"),
+        ("image", "blob_v2"),
+    ]
+
+
+def test_struct_blob_signature_fields_preserve_exact_metadata_and_nullability():
+    nested_input = pa.field(
+        "payload",
+        pa.struct(
+            [
+                pa.field("mime_type", pa.string(), nullable=False),
+                pa.field(
+                    "nested",
+                    pa.struct([lancedb.blob("image", nullable=True)]),
+                    nullable=True,
+                ),
+            ]
+        ),
+        nullable=True,
+    )
+    nested_output = pa.field(
+        "result",
+        pa.struct(
+            [
+                pa.field("mime_type", pa.string(), nullable=False),
+                pa.field(
+                    "nested",
+                    pa.struct([lancedb.blob("image", nullable=True)]),
+                    nullable=False,
+                ),
+            ]
+        ),
+        nullable=False,
+    )
+
+    @udf(input_schema=pa.schema([nested_input]), output_schema=nested_output)
+    def copy_payload(payload):
+        return payload
+
+    signature = copy_payload.registration_request.signature
+    input_type = json.loads(signature.inputs[0].arrow_type)
+    assert input_type["fields"][1]["nullable"] is True
+    input_blob = input_type["fields"][1]["type"]["fields"][0]
+    assert input_blob["nullable"] is True
+    assert input_blob["metadata"] == {"ARROW:extension:name": "lance.blob.v2"}
+
+    assert signature.output.kind == "named_struct"
+    nested_result = next(
+        field for field in signature.output.fields if field.name == "nested"
+    )
+    output_type = json.loads(nested_result.arrow_type)
+    output_blob = output_type["fields"][0]
+    assert output_blob["nullable"] is True
+    assert output_blob["metadata"] == {"ARROW:extension:name": "lance.blob.v2"}
+
+
+def test_struct_blob_signature_supports_multiple_struct_levels():
+    recursive = pa.field(
+        "value",
+        pa.struct(
+            [
+                pa.field(
+                    "level_1",
+                    pa.struct(
+                        [
+                            pa.field(
+                                "level_2",
+                                pa.struct([lancedb.blob("image", nullable=False)]),
+                                nullable=False,
+                            )
+                        ]
+                    ),
+                    nullable=False,
+                )
+            ]
+        ),
+        nullable=False,
+    )
+
+    @udf(
+        input_schema=pa.schema([recursive]),
+        output_schema=pa.field("size", pa.int64(), nullable=False),
+    )
+    def blob_size(value):
+        return len(value["level_1"]["level_2"]["image"])
+
+    encoded = json.loads(blob_size.registration_request.signature.inputs[0].arrow_type)
+    blob = encoded["fields"][0]["type"]["fields"][0]["type"]["fields"][0]
+    assert blob["metadata"]["ARROW:extension:name"] == "lance.blob.v2"
+
+
+@pytest.mark.parametrize(
+    "data_type",
+    [
+        pa.list_(lancedb.blob("item", nullable=False)),
+        pa.large_list(lancedb.blob("item", nullable=False)),
+        pa.list_(lancedb.blob("item", nullable=False), 2),
+        pa.map_(pa.string(), lancedb.blob("value", nullable=False).type),
+    ],
+)
+def test_blob_signature_rejects_collection_ancestors(data_type):
+    with pytest.raises(
+        TypeError,
+        match="Blob v2 fields nested under collection types are not supported",
+    ):
+
+        @udf(
+            input_schema=pa.schema([pa.field("value", data_type, nullable=False)]),
+            output_schema=pa.field("size", pa.int64(), nullable=False),
+        )
+        def blob_size(value):
+            return len(value)
+
+
+def test_blob_signature_rejects_collection_below_a_struct():
+    nested = pa.field(
+        "value",
+        pa.struct(
+            [
+                pa.field(
+                    "images",
+                    pa.list_(lancedb.blob("item", nullable=False)),
+                    nullable=False,
+                )
+            ]
+        ),
+        nullable=False,
+    )
+    with pytest.raises(
+        TypeError,
+        match="Blob v2 fields nested under collection types 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["images"])
+
+
+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_named_struct_function_preserves_nullable_result_fields():
+    @udf(
+        input_schema=pa.schema([pa.field("value", pa.int64(), nullable=False)]),
+        output_schema=pa.schema(
+            [
+                pa.field("result", pa.int64(), nullable=True),
+                pa.field("failure_code", pa.int32(), nullable=False),
+            ]
+        ),
+    )
+    def nullable_result(value):
+        return {"result": value, "failure_code": 0}
+
+    output = nullable_result.registration_request.signature.output
+    assert [(field.name, field.nullable) for field in output.fields] == [
+        ("result", True),
+        ("failure_code", False),
+    ]
+
+    @udf(
+        input_schema=pa.schema([pa.field("value", pa.int64(), nullable=False)]),
+        output_schema=pa.schema(
+            [
+                pa.field("result", pa.int64(), nullable=True),
+                pa.field("failure_code", pa.int32(), nullable=True),
+            ]
+        ),
+    )
+    def all_nullable(value):
+        return {"result": value, "failure_code": None}
+
+    assert all(
+        field.nullable
+        for field in all_nullable.registration_request.signature.output.fields
+    )
+
+
+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_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"):

@@ -525,6 +1123,72 @@ 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)
@@ -535,6 +1199,10 @@ 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.list_functions()
+    with pytest.raises(NotImplementedError, match=message):
+        db.drop_function("normalize_score", version="fv_exact")


@contextlib.contextmanager
@@ -580,6 +1248,28 @@ def _mock_remote_function_catalog():
                    "version": "fv_exact",
                }
                response = state["version"]
+            elif self.path == "/v1/functions/list":
+                assert body["include_definition"] is True
+                if "page_token" not in body:
+                    response = {
+                        "functions": [
+                            {
+                                "name": "normalize_score",
+                                "version": "fv_exact",
+                                "definition": state["version"],
+                            }
+                        ],
+                        "page_token": "next",
+                    }
+                else:
+                    assert body["page_token"] == "next"
+                    response = {"functions": []}
+            elif self.path == "/v1/functions/drop":
+                assert body == {
+                    "name": "normalize_score",
+                    "version": "fv_exact",
+                }
+                response = {"dropped": True}
            else:
                status = 404
                response = {"error": "not found"}
@@ -638,3 +1328,83 @@ def test_blocking_remote_registration_returns_function_version():
        "/v1/functions/create",
        "/v1/jobs/describe",
    ]
+
+
+def test_remote_list_functions_paginates_and_returns_typed_versions():
+    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}},
+        )
+        created = db.create_function(normalize_score)
+        state["requests"].clear()
+        functions = db.list_functions()
+
+    assert functions == [created]
+    assert state["requests"] == [
+        ("/v1/functions/list", {"include_definition": True}),
+        (
+            "/v1/functions/list",
+            {"include_definition": True, "page_token": "next"},
+        ),
+    ]
+
+
+@pytest.mark.asyncio
+async def test_async_remote_list_functions_returns_typed_versions():
+    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}},
+        )
+        registration = await db.create_function_async(normalize_score)
+        created = await registration.wait()
+        state["requests"].clear()
+        functions = await db.list_functions()
+
+    assert functions == [created]
+    assert [path for path, _ in state["requests"]] == [
+        "/v1/functions/list",
+        "/v1/functions/list",
+    ]
+
+
+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"},
+        )
+    ]
@@ -1011,8 +1011,13 @@ def test_fts_ngram(mem_db: DBConnection):
    assert set(r["text"] for r in results) == {"lance database", "lance is cool"}

    results = (
-        table.search("nce", query_type="fts").limit(10).to_list()
-    )  # spellchecker:disable-line
+        table.search(
+            "nce",  # spellchecker:disable-line
+            query_type="fts",
+        )
+        .limit(10)
+        .to_list()
+    )
    assert len(results) == 2
    assert set(r["text"] for r in results) == {"lance database", "lance is cool"}

@@ -1034,8 +1039,13 @@ def test_fts_ngram(mem_db: DBConnection):
    assert set(r["text"] for r in results) == {"lance database", "lance is cool"}

    results = (
-        table.search("nce", query_type="fts").limit(10).to_list()
-    )  # spellchecker:disable-line
+        table.search(
+            "nce",  # spellchecker:disable-line
+            query_type="fts",
+        )
+        .limit(10)
+        .to_list()
+    )
    assert len(results) == 0

    results = table.search("la", query_type="fts").limit(10).to_list()
@@ -54,7 +54,10 @@ class TestOAuthProvider:
        provider = OAuthProvider(fetcher)
        headers = provider.get_headers()

-        assert headers == {"Authorization": "Bearer token123"}
+        assert headers == {
+            "Authorization": "Bearer token123",
+            "x-lancedb-credential-type": "oidc",
+        }
        assert provider._current_token == "token123"
        assert provider._token_expires_at is not None

@@ -73,14 +76,20 @@ class TestOAuthProvider:

        # First call
        headers1 = provider.get_headers()
-        assert headers1 == {"Authorization": "Bearer token1"}
+        assert headers1 == {
+            "Authorization": "Bearer token1",
+            "x-lancedb-credential-type": "oidc",
+        }

        # Wait for token to expire
        time.sleep(1.1)

        # Second call should refresh
        headers2 = provider.get_headers()
-        assert headers2 == {"Authorization": "Bearer token2"}
+        assert headers2 == {
+            "Authorization": "Bearer token2",
+            "x-lancedb-credential-type": "oidc",
+        }
        assert call_count == 2

    def test_no_expiry_info(self):
@@ -92,12 +101,18 @@ class TestOAuthProvider:
        provider = OAuthProvider(fetcher)
        headers = provider.get_headers()

-        assert headers == {"Authorization": "Bearer permanent_token"}
+        assert headers == {
+            "Authorization": "Bearer permanent_token",
+            "x-lancedb-credential-type": "oidc",
+        }
        assert provider._token_expires_at is None

        # Should not refresh on second call
        headers2 = provider.get_headers()
-        assert headers2 == {"Authorization": "Bearer permanent_token"}
+        assert headers2 == {
+            "Authorization": "Bearer permanent_token",
+            "x-lancedb-credential-type": "oidc",
+        }

    def test_missing_access_token(self):
        """Test error handling when access_token is missing."""
@@ -121,7 +136,10 @@ class TestOAuthProvider:
        provider = OAuthProvider(fetcher)
        headers = provider.get_headers()

-        assert headers == {"Authorization": "Bearer sync_token"}
+        assert headers == {
+            "Authorization": "Bearer sync_token",
+            "x-lancedb-credential-type": "oidc",
+        }


 class TestClientConfigIntegration:
@@ -203,6 +203,93 @@ async def test_async_hybrid_query_default_limit(table: AsyncTable):
    assert texts.count("a") == 1


+@pytest.mark.asyncio
+async def test_async_hybrid_query_offset(table: AsyncTable):
+    # The offset window of a hybrid query must be a suffix of the same query
+    # run without an offset. Skipping the first rows of each sub-query instead
+    # of the first rows of the fused result silently changes which rows land in
+    # the window.
+    full = await (
+        table.query()
+        .nearest_to([0.0, 0.4])
+        .nearest_to_text("dog")
+        .limit(4)
+        .with_row_id()
+        .to_arrow()
+    )
+    assert len(full) == 4
+
+    second_page = await (
+        table.query()
+        .nearest_to([0.0, 0.4])
+        .nearest_to_text("dog")
+        .offset(2)
+        .limit(2)
+        .with_row_id()
+        .to_arrow()
+    )
+    assert second_page["_rowid"].to_pylist() == full["_rowid"].to_pylist()[2:]
+
+    first_page = await (
+        table.query()
+        .nearest_to([0.0, 0.4])
+        .nearest_to_text("dog")
+        .limit(2)
+        .with_row_id()
+        .to_arrow()
+    )
+    # Paging through the result must visit every row exactly once: no row
+    # repeated from the previous page and none dropped between the two.
+    paged = first_page["_rowid"].to_pylist() + second_page["_rowid"].to_pylist()
+    assert sorted(paged) == sorted(full["_rowid"].to_pylist())
+
+
+@pytest.mark.asyncio
+async def test_async_hybrid_query_fts_first_default_limit(table: AsyncTable):
+    # nearest_to() and nearest_to_text() build their new sibling sub-query from
+    # scratch, and that is the sub-query the default limit ends up on. So the
+    # side that carries the limit depends on the order the hybrid query was
+    # built in, and looking at only one side loses the limit for half the ways
+    # a hybrid query can be written. Without a limit the combined results are
+    # not truncated at all and the whole union of both candidate lists is
+    # returned.
+    await table.add([{"text": "dog", "vector": [50.0 + i, 50.0]} for i in range(10)])
+
+    result = await (
+        table.query().nearest_to_text("dog").nearest_to([0.1, 0.1]).to_arrow()
+    )
+    assert len(result) == 10
+
+    offset_result = await (
+        table.query().nearest_to_text("dog").nearest_to([0.1, 0.1]).offset(2).to_arrow()
+    )
+    assert len(offset_result) == 10
+
+
+@pytest.mark.asyncio
+async def test_async_hybrid_query_explain_plan_matches_execution(table: AsyncTable):
+    # Paging rewrites the sub-queries: each one fetches limit + offset rows with
+    # no offset of its own, and the window is sliced out after fusion. The plans
+    # have to be built from those rewritten sub-queries, otherwise explain_plan
+    # and analyze_plan describe a query that is never run.
+    query = (
+        table.query().nearest_to([0.0, 0.4]).nearest_to_text("dog").offset(2).limit(2)
+    )
+    await query.to_arrow()
+
+    plan = await query.explain_plan()
+    assert [
+        line.strip() for line in plan.splitlines() if "GlobalLimitExec" in line
+    ] == [
+        "GlobalLimitExec: skip=0, fetch=4",
+        "GlobalLimitExec: skip=0, fetch=4",
+    ]
+
+    analyzed = await query.analyze_plan()
+    assert analyzed.count("skip=0, fetch=4") == 2
+    assert "skip=2" not in analyzed
+
+
 def test_hybrid_query_offset(sync_table: Table):
    # The offset window of a hybrid query must be a suffix of the same query
    # run without an offset -- it must not be silently ignored.
@@ -266,3 +266,38 @@ 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"
+        )
@@ -193,7 +193,13 @@ class TestNamespaceConnection:
            ),
        )

-        table = db.create_table("blob_table", data, namespace_path=["test_ns"])
+        # Legacy v1 blob columns are only writable at file version <= 2.1.
+        table = db.create_table(
+            "blob_table",
+            data,
+            namespace_path=["test_ns"],
+            storage_options={"new_table_data_storage_version": "2.1"},
+        )
        df = table.to_pandas(blob_mode="lazy").sort_values("id")

        blob = df["blob"].iloc[0]
@@ -40,6 +40,10 @@ from utils import exception_output
 from importlib.util import find_spec


+# Legacy v1 blob columns are only writable at file version <= 2.1.
+LEGACY_BLOB_STORAGE_OPTIONS = {"new_table_data_storage_version": "2.1"}
+
+
 def _blob_query_data():
    return pa.table(
        {
@@ -119,13 +123,17 @@ def _assert_blob_bytes_projection(df):

 def _blob_query_table(db, name, blob_schema):
    if blob_schema == "v1":
-        return db.create_table(name, _blob_query_data())
+        return db.create_table(
+            name, _blob_query_data(), storage_options=LEGACY_BLOB_STORAGE_OPTIONS
+        )
    return _create_blob_v2_query_table(db, name)


 async def _blob_query_table_async(db, name, blob_schema):
    if blob_schema == "v1":
-        return await db.create_table(name, _blob_query_data())
+        return await db.create_table(
+            name, _blob_query_data(), storage_options=LEGACY_BLOB_STORAGE_OPTIONS
+        )
    return await _create_blob_v2_query_table_async(db, name)


@@ -275,7 +283,9 @@ async def test_query_to_pandas_kwargs(table, table_async):
 def test_plain_scan_query_to_pandas_blob_modes(tmp_db, blob_mode):
    pytest.importorskip("lance")
    table = tmp_db.create_table(
-        f"test_query_to_pandas_blob_{blob_mode}", _blob_query_data()
+        f"test_query_to_pandas_blob_{blob_mode}",
+        _blob_query_data(),
+        storage_options=LEGACY_BLOB_STORAGE_OPTIONS,
    )

    df = (
@@ -322,7 +332,9 @@ def test_plain_scan_query_to_pandas_blob_mode_does_not_collect_arrow(
 ):
    pytest.importorskip("lance")
    table = tmp_db.create_table(
-        "test_query_to_pandas_blob_no_arrow_collect", _blob_query_data()
+        "test_query_to_pandas_blob_no_arrow_collect",
+        _blob_query_data(),
+        storage_options=LEGACY_BLOB_STORAGE_OPTIONS,
    )
    query = table.search().where("id = 1").select(["id", "blob"])

@@ -347,7 +359,9 @@ def test_plain_scan_query_to_pandas_blob_descriptions_flatten_uses_scanner(
 ):
    pytest.importorskip("lance")
    table = tmp_db.create_table(
-        "test_query_to_pandas_blob_desc_flatten", _blob_query_data()
+        "test_query_to_pandas_blob_desc_flatten",
+        _blob_query_data(),
+        storage_options=LEGACY_BLOB_STORAGE_OPTIONS,
    )
    query = table.search().where("id = 1").select(["id", "blob"])

@@ -365,7 +379,11 @@ def test_plain_scan_query_to_pandas_blob_descriptions_flatten_uses_scanner(
 def test_plain_scan_query_to_pandas_scanner_state(tmp_db):
    pytest.importorskip("lance")
    data = _blob_query_data()
-    table = tmp_db.create_table("test_query_to_pandas_scanner_state", data.slice(0, 2))
+    table = tmp_db.create_table(
+        "test_query_to_pandas_scanner_state",
+        data.slice(0, 2),
+        storage_options=LEGACY_BLOB_STORAGE_OPTIONS,
+    )
    table.add(data.slice(2, 2))

    fragments = table.to_lance().get_fragments()
@@ -400,7 +418,9 @@ def test_plain_scan_query_to_pandas_scanner_state(tmp_db):
 async def test_async_plain_scan_query_to_pandas_blob_projection(tmp_db_async):
    pytest.importorskip("lance")
    table = await tmp_db_async.create_table(
-        "test_async_query_to_pandas_blob_projection", _blob_query_data()
+        "test_async_query_to_pandas_blob_projection",
+        _blob_query_data(),
+        storage_options=LEGACY_BLOB_STORAGE_OPTIONS,
    )

    lazy_df = await (
@@ -452,7 +472,9 @@ async def test_async_plain_scan_query_to_pandas_blob_mode_does_not_collect_arrow
 ):
    pytest.importorskip("lance")
    table = await tmp_db_async.create_table(
-        "test_async_query_to_pandas_blob_no_arrow_collect", _blob_query_data()
+        "test_async_query_to_pandas_blob_no_arrow_collect",
+        _blob_query_data(),
+        storage_options=LEGACY_BLOB_STORAGE_OPTIONS,
    )
    query = table.query().where("id = 1").select(["id", "blob"])

@@ -474,7 +496,11 @@ async def test_async_plain_scan_query_to_pandas_blob_mode_does_not_collect_arrow

 def test_vector_query_to_pandas_blob_mode_requires_native_path(tmp_db):
    pytest.importorskip("lance")
-    table = tmp_db.create_table("test_vector_query_blob_mode", _blob_query_data())
+    table = tmp_db.create_table(
+        "test_vector_query_blob_mode",
+        _blob_query_data(),
+        storage_options=LEGACY_BLOB_STORAGE_OPTIONS,
+    )

    with pytest.raises(RuntimeError, match="Lance native pandas conversion"):
        table.search([1.0, 0.0]).select(["blob", "vector"]).limit(1).to_pandas(
@@ -485,7 +511,9 @@ def test_vector_query_to_pandas_blob_mode_requires_native_path(tmp_db):
 def test_vector_query_to_pandas_blob_descriptions_requires_plain_scan(tmp_db):
    pytest.importorskip("lance")
    table = tmp_db.create_table(
-        "test_vector_query_blob_descriptions", _blob_query_data()
+        "test_vector_query_blob_descriptions",
+        _blob_query_data(),
+        storage_options=LEGACY_BLOB_STORAGE_OPTIONS,
    )

    with pytest.raises(RuntimeError, match="plain scan query"):
@@ -1923,6 +1951,21 @@ 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,24 +479,49 @@ def test_remote_permutation_is_picklable():
                match = re.search(
                    r"_rowoffset\s+in\s+\((.*?)\)", body["filter"], re.IGNORECASE
                )
-                offsets = [int(o.strip()) for o in match.group(1).split(",")]
+                offsets = list(
+                    dict.fromkeys(int(o.strip()) for o in match.group(1).split(","))
+                )
            else:
                offsets = list(range(len(rows)))
-            table = pa.table({"a": [rows[offset] for offset in offsets]})
+            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
+                }
+            )

            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)
+                writer.write_table(table, max_chunksize=2)
        else:
            request.send_response(404)
            request.end_headers()

    with mock_lancedb_connection(handler) as db:
-        permutation = Permutation.identity(db.open_table("test"))
+        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)
        restored = pickle.loads(pickle.dumps(permutation))
-        assert restored.__getitems__([0, 2, 4]) == [{"a": 0}, {"a": 2}, {"a": 4}]
+        assert restored.__getitems__([0, 2, 0, 4]) == [
+            {"a": 0},
+            {"a": 2},
+            {"a": 0},
+            {"a": 4},
+        ]


 def test_create_table_exist_ok():
@@ -795,11 +820,13 @@ def test_table_create_indices():
        scalar_req = received_requests[0]
        assert "name" in scalar_req
        assert scalar_req["name"] == "custom_scalar_idx"
+        assert scalar_req["replace"] is False

        # Check FTS index request has custom name
        fts_req = received_requests[1]
        assert "name" in fts_req
        assert fts_req["name"] == "custom_fts_idx"
+        assert fts_req["replace"] is False
        assert fts_req["block_size"] == 256
        assert fts_req["custom_stop_words"] == ["cloud"]

@@ -807,6 +834,7 @@ def test_table_create_indices():
        vector_req = received_requests[2]
        assert "name" in vector_req
        assert vector_req["name"] == "custom_vector_idx"
+        assert "replace" not in vector_req

        table.wait_for_index(["custom_scalar_idx"], timedelta(seconds=2))
        table.wait_for_index(
@@ -1079,6 +1107,9 @@ def test_remote_create_index_new_api():
            table.create_index("text", config=FTS(block_size=256))
            # IvfRq via new API
            table.create_index("vector", config=IvfRq(distance_type="l2"))
+            table.create_index(
+                "vector", config=IvfPq(distance_type="l2"), replace=False
+            )

        # Legacy index_type="IVF_RQ" routes to IvfRq config under the hood.
        with pytest.warns(DeprecationWarning, match="create_index"):
@@ -1088,15 +1119,17 @@ def test_remote_create_index_new_api():
                num_partitions=8,
            )

-        assert len(received_requests) == 5
+        assert len(received_requests) == 6
        assert [req["column"] for req in received_requests] == [
            "vector",
            "category",
            "text",
            "vector",
            "vector",
+            "vector",
        ]
        assert received_requests[2]["block_size"] == 256
+        assert received_requests[4]["replace"] is False


 def test_table_wait_for_index_timeout():
@@ -2434,7 +2467,7 @@ def test_remote_blob_byte_apis_not_supported_on_old_server():


 def test_remote_connection_jobs_surface():
-    from lancedb.exceptions import JobFailedError
+    from lancedb.exceptions import JobFailedError, JobNotFoundError

    schema = pa.schema([("state", pa.string())])
    batch = pa.record_batch([pa.array(["created", "done"])], schema=schema)
@@ -2442,6 +2475,7 @@ def test_remote_connection_jobs_surface():
    with pa.ipc.new_stream(sink, schema) as writer:
        writer.write_batch(batch)
    events_body = sink.getvalue().to_pybytes()
+    query_events_payloads = []

    def handler(request):
        content_len = int(request.headers.get("Content-Length", 0))
@@ -2479,6 +2513,22 @@ def test_remote_connection_jobs_surface():
            request.end_headers()
            request.wfile.write(json.dumps(rsp).encode())
        elif request.path == "/v1/jobs/describe":
+            if payload["job_id"] == "job-2":
+                request.send_response(200)
+                request.send_header("Content-Type", "application/json")
+                request.end_headers()
+                request.wfile.write(
+                    json.dumps(
+                        dict(
+                            job_id="job-2",
+                            job_type="refresh_column",
+                            job_state="DONE",
+                            creation_ms=2000,
+                            result=dict(rows_assigned=1000000, rows_failed=0),
+                        )
+                    ).encode()
+                )
+                return
            if payload["job_id"] != "job-1":
                request.send_response(404)
                request.end_headers()
@@ -2510,7 +2560,7 @@ def test_remote_connection_jobs_surface():
            request.end_headers()
            request.wfile.write(b'{"job_id": "job-1"}')
        elif request.path == "/v1/jobs/query_events":
-            assert payload["job_id"] == "job-1"
+            query_events_payloads.append(payload)
            request.send_response(200)
            request.send_header("Content-Type", "application/vnd.apache.arrow.stream")
            request.end_headers()
@@ -2526,24 +2576,109 @@ def test_remote_connection_jobs_surface():
        assert jobs[0].table == "t1"
        assert jobs[1].state == "finished"

-        description = db.get_job("job-1")
-        assert description.job_type == "create_index"
-        assert description.state == "failed"
-        assert json.loads(description.spec_json) == {"column": "vec"}
-        assert description.failure.message == "worker died"
-        assert description.failure.retryable is True
-        assert db.get_job("missing") is None
-
        assert db.cancel_job("job-1") is True
        assert db.cancel_job("missing") is False

-        batches = db.job_history("job-1")
-        assert len(batches) == 1
-        assert batches[0].num_rows == 2
-        assert batches[0].column("state").to_pylist() == ["created", "done"]
+        # Opening a job hands back a populated handle; a missing one fails.
+        with pytest.raises(JobNotFoundError, match="missing"):
+            db.open_job("missing")
+        finished = db.open_job("job-2")
+        assert finished.state == "finished"
+        assert finished.result == {"rows_assigned": 1000000, "rows_failed": 0}

-        job = db.job("job-1")
+        job = db.open_job("job-1")
        assert job.id == "job-1"
+        # Opening already populated the handle.
+        assert job.state == "failed"
+        assert job.spec == {"column": "vec"}
+        assert job.failure.message == "worker died"
        assert job.status() == "failed"
        with pytest.raises(JobFailedError, match="worker died"):
            job.wait(timeout=timedelta(seconds=5))
+
+
+def test_remote_job_handle_reports_its_own_detail():
+    schema = pa.schema([("state", pa.string())])
+    batch = pa.record_batch([pa.array(["claim_complete"])], schema=schema)
+    sink = pa.BufferOutputStream()
+    with pa.ipc.new_stream(sink, schema) as writer:
+        writer.write_batch(batch)
+    events_body = sink.getvalue().to_pybytes()
+    event_payloads = []
+
+    def handler(request):
+        content_len = int(request.headers.get("Content-Length", 0))
+        body = request.rfile.read(content_len) if content_len > 0 else b""
+        payload = json.loads(body) if body else {}
+        if request.path == "/v1/jobs/describe":
+            request.send_response(200)
+            request.send_header("Content-Type", "application/json")
+            request.end_headers()
+            request.wfile.write(
+                json.dumps(
+                    dict(
+                        job_id="job-1",
+                        job_type="refresh_column",
+                        job_state="DONE",
+                        creation_ms=2000,
+                        spec=dict(column="vec"),
+                        result=dict(rows_assigned=1000000),
+                    )
+                ).encode()
+            )
+        elif request.path == "/v1/jobs/query_events":
+            event_payloads.append(payload)
+            request.send_response(200)
+            request.send_header("Content-Type", "application/vnd.apache.arrow.stream")
+            request.end_headers()
+            request.wfile.write(events_body)
+        else:
+            request.send_response(404)
+            request.end_headers()
+
+    with mock_lancedb_connection(handler) as db:
+        job = db.open_job("job-1")
+
+        # Opening populates the handle in the same round trip.
+        assert job.state == "finished"
+        job.refresh()
+        assert job.job_type == "refresh_column"
+        assert job.creation_ms == 2000
+        assert job.spec == {"column": "vec"}
+        assert job.result == {"rows_assigned": 1000000}
+        assert job.failure is None
+        # The JSON payloads stay reachable, but as internal APIs.
+        assert json.loads(job._spec_json) == {"column": "vec"}
+        assert json.loads(job._result_json) == {"rows_assigned": 1000000}
+
+        # print() shows everything the handle knows and nothing it does not.
+        # print() lays every known field out on its own line, with the JSON
+        # payloads indented rather than crammed onto one line.
+        assert repr(job) == "\n".join(
+            [
+                "Job(",
+                "    id='job-1',",
+                "    state='finished',",
+                "    job_type='refresh_column',",
+                "    creation_ms=2000,",
+                "    spec={",
+                '        "column": "vec"',
+                "    },",
+                "    result={",
+                '        "rows_assigned": 1000000',
+                "    },",
+                ")",
+            ]
+        )
+        # Nothing it does not know shows up.
+        assert "failure" not in repr(job)
+
+        events = job.events(filter="state = 'claim_complete'", limit=500)
+        assert isinstance(events, pa.Table)
+        assert events.column("state").to_pylist() == ["claim_complete"]
+        # The handle supplies job_id; the caller only narrows the query.
+        assert event_payloads[-1] == {
+            "job_id": "job-1",
+            "limit": 500,
+            "filter": "state = 'claim_complete'",
+        }
@@ -81,7 +81,7 @@ def get_test_table(tmp_path):
        "but his son was mortal",
        "there hasn't been a good battlefield game since 2142",
        "I wish they would make another one",
-        "campains are not as good as they used to be",
+        "campaigns are not as good as they used to be",
        "Multiplayer and open world games have destroyed the single player experience",
        "Maybe the future is console games",
        "I don't know",
--- a/Show More
+++ b/Show More