Bump version: 0.35.0-beta.0 → 0.35.0-beta.1

ci(nodejs): move Windows builds to larger runner and use ThinLTO (#3634 )
The `build - aarch64-pc-windows-msvc` node build job (and, marginally, the x86_64 one) had started hitting `rustc-LLVM ERROR: out of memory` while linking the `lancedb-nodejs` cdylib — most recently surfaced by #3526, which adds the goosefs backend (and its tonic/prost gRPC subtree) to the default node binary. The peak-memory step is the fat-LTO codegen (`lto=fat`, `codegen-units=1` from `.cargo/config.toml`), which merges the whole crate graph into a single LLVM module and runs single-threaded. It therefore neither parallelizes across cores nor fits in the 16 GB of the standard `windows-latest` runner as the dependency graph grows. This PR: - Moves both `*-pc-windows-msvc` node build jobs to `windows-2025-8x-x64` (more memory + cores). - Overrides the release profile to ThinLTO for just these jobs, via `CARGO_PROFILE_RELEASE_LTO=thin` / `CARGO_PROFILE_RELEASE_CODEGEN_UNITS=16` in `pre_build`. ThinLTO parallelizes the cross-module optimization across the runner's cores and keeps peak memory well under the limit. Scoped so Python wheels and Rust release builds keep fat LTO. The larger runner alone would clear the OOM but waste the added cores on the single-threaded fat-LTO tail; ThinLTO is what makes the extra cores actually reduce wall-clock and gives durable memory headroom for future dependency growth. Tradeoff: ThinLTO can leave a small runtime-perf gap vs fat LTO for the node native binary, but it recovers most of it and is a common release configuration. Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-08-26 07:58:31 +00:00 · 2026-07-10 16:12:51 +00:00 · 2026-07-10 08:50:43 -07:00 · 2026-07-10 05:26:01 +00:00 · 2026-07-10 05:25:24 +00:00 · 2026-07-09 15:36:03 -07:00
108 changed files with 7085 additions and 562 deletions
@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "0.31.0-beta.4"
+current_version = "0.32.0-beta.0"
 parse = """(?x)
    (?P<major>0|[1-9]\\d*)\\.
    (?P<minor>0|[1-9]\\d*)\\.
@@ -0,0 +1,21 @@
+# CODEOWNERS
+#
+# These owners will be the default owners for everything in the repo.
+# They will be requested for review when someone opens a pull request.
+#
+# See https://docs.github.com/en/repositories/managing-your-repositorys-settings-and-features/customizing-your-repository/about-code-owners
+
+# Default owners for everything
+* @jackye1995 @wjones127
+
+# Release and publish workflows — changes here can affect supply chain security
+/.github/workflows/ @jackye1995 @wjones127 @Xuanwo
+
+# Remote client and auth — sensitive networking and auth code
+/rust/lancedb/src/remote/ @jackye1995 @wjones127
+
+# Python FFI boundary
+/python/src/ @jackye1995 @wjones127 @AyushExel
+
+# NodeJS FFI boundary
+/nodejs/src/ @jackye1995 @wjones127
@@ -34,15 +34,16 @@ runs:
        maturin-version: "1.12.4"
        command: build
        working-directory: python
-        docker-options: "-e PIP_EXTRA_INDEX_URL='https://pypi.fury.io/lance-format/ https://pypi.fury.io/lancedb/'"
+        docker-options: "-e PIP_EXTRA_INDEX_URL='https://pypi.fury.io/lance-format/ https://pypi.fury.io/lancedb/' -e PROTOC=/usr/local/bin/protoc"
        target: x86_64-unknown-linux-gnu
        manylinux: ${{ inputs.manylinux }}
        args: ${{ inputs.args }}
        before-script-linux: |
          set -e
-          curl -L https://github.com/protocolbuffers/protobuf/releases/download/v24.4/protoc-24.4-linux-$(uname -m).zip > /tmp/protoc.zip \
-            && unzip /tmp/protoc.zip -d /usr/local \
-            && rm /tmp/protoc.zip
+          curl -fsSL https://github.com/protocolbuffers/protobuf/releases/download/v24.4/protoc-24.4-linux-x86_64.zip -o /tmp/protoc.zip
+          unzip /tmp/protoc.zip -d /usr/local
+          rm /tmp/protoc.zip
+          /usr/local/bin/protoc --version
    - name: Build Arm Manylinux Wheel
      if: ${{ inputs.arm-build == 'true' }}
      uses: PyO3/maturin-action@v1
@@ -50,13 +51,14 @@ runs:
        maturin-version: "1.12.4"
        command: build
        working-directory: python
-        docker-options: "-e PIP_EXTRA_INDEX_URL='https://pypi.fury.io/lance-format/ https://pypi.fury.io/lancedb/'"
+        docker-options: "-e PIP_EXTRA_INDEX_URL='https://pypi.fury.io/lance-format/ https://pypi.fury.io/lancedb/' -e PROTOC=/usr/local/bin/protoc"
        target: aarch64-unknown-linux-gnu
        manylinux: ${{ inputs.manylinux }}
        args: ${{ inputs.args }}
        before-script-linux: |
          set -e
-          yum install -y clang \
-            && curl -L https://github.com/protocolbuffers/protobuf/releases/download/v24.4/protoc-24.4-linux-aarch_64.zip > /tmp/protoc.zip \
-            && unzip /tmp/protoc.zip -d /usr/local \
-            && rm /tmp/protoc.zip
+          yum install -y clang
+          curl -fsSL https://github.com/protocolbuffers/protobuf/releases/download/v24.4/protoc-24.4-linux-aarch_64.zip -o /tmp/protoc.zip
+          unzip /tmp/protoc.zip -d /usr/local
+          rm /tmp/protoc.zip
+          /usr/local/bin/protoc --version
@@ -25,7 +25,7 @@ jobs:
    # Only runs on tags that matches the make-release action
    if: startsWith(github.ref, 'refs/tags/v')
    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
      - uses: Swatinem/rust-cache@v2
        with:
          workspaces: rust
@@ -47,7 +47,7 @@ jobs:
      contents: read
      issues: write
    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
      - uses: ./.github/actions/create-failure-issue
        with:
          job-results: ${{ toJSON(needs) }}
@@ -36,14 +36,14 @@ jobs:
          echo "guidelines = ${{ inputs.guidelines }}"

      - name: Checkout Repo
-        uses: actions/checkout@v4
+        uses: actions/checkout@v6
        with:
          ref: ${{ inputs.branch }}
          fetch-depth: 0
          persist-credentials: true

      - name: Set up Node.js
-        uses: actions/setup-node@v4
+        uses: actions/setup-node@v6
        with:
          # pnpm 11 (used by the nodejs install step below) requires
          # Node >= 22.13; use 24 since 22 hits EOL in October.
@@ -82,7 +82,7 @@ jobs:
          cache: maven

      - name: Setup pnpm
-        uses: pnpm/action-setup@v4
+        uses: pnpm/action-setup@v6
        with:
          version: 11.1.1
      - name: Install Node.js dependencies for TypeScript bindings
@@ -30,13 +30,13 @@ jobs:
          echo "tag = ${{ inputs.tag || 'latest' }}"

      - name: Checkout Repo LanceDB
-        uses: actions/checkout@v4
+        uses: actions/checkout@v6
        with:
          fetch-depth: 0
          persist-credentials: true

      - name: Set up Node.js
-        uses: actions/setup-node@v4
+        uses: actions/setup-node@v6
        with:
          node-version: 20

@@ -27,7 +27,7 @@ jobs:
    name: Verify PR title / description conforms to semantic-release
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/setup-node@v4
+      - uses: actions/setup-node@v6
        with:
          node-version: "18"
      # These rules are disabled because Github will always ensure there
@@ -35,7 +35,7 @@ jobs:
    runs-on: ubuntu-24.04
    steps:
      - name: Checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@v6
      - name: Install dependencies needed for ubuntu
        run: |
          sudo apt install -y protobuf-compiler libssl-dev
@@ -53,7 +53,7 @@ jobs:
          python -m pip install --extra-index-url https://pypi.fury.io/lance-format/ --extra-index-url https://pypi.fury.io/lancedb/ -e .
          python -m pip install --extra-index-url https://pypi.fury.io/lance-format/ --extra-index-url https://pypi.fury.io/lancedb/ -r ../docs/requirements.txt
      - name: Set up node
-        uses: actions/setup-node@v4
+        uses: actions/setup-node@v6
        with:
          node-version: 20
          cache: 'npm'
@@ -32,7 +32,7 @@ jobs:
        working-directory: ./java
    steps:
      - name: Checkout repository
-        uses: actions/checkout@v4
+        uses: actions/checkout@v6
      - name: Set up Java 8
        uses: actions/setup-java@v4
        with:
@@ -73,7 +73,7 @@ jobs:
      contents: read
      issues: write
    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
      - uses: ./.github/actions/create-failure-issue
        with:
          job-results: ${{ toJSON(needs) }}
@@ -36,7 +36,7 @@ jobs:
        working-directory: ./java
    steps:
      - name: Checkout repository
-        uses: actions/checkout@v4
+        uses: actions/checkout@v6
      - name: Set up Java 17
        uses: actions/setup-java@v4
        with:
@@ -19,7 +19,7 @@ jobs:
    runs-on: ubuntu-latest
    steps:
      - name: Check out code
-        uses: actions/checkout@v4
+        uses: actions/checkout@v6
      - name: Install license-header-checker
        working-directory: /tmp
        run: |
@@ -49,7 +49,7 @@ jobs:
    steps:
      - name: Output Inputs
        run: echo "${{ toJSON(github.event.inputs) }}"
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
        with:
          fetch-depth: 0
          lfs: true
@@ -38,14 +38,14 @@ jobs:
      CC: gcc-12
      CXX: g++-12
    steps:
-    - uses: actions/checkout@v4
+    - uses: actions/checkout@v6
      with:
        fetch-depth: 0
        lfs: true
-    - uses: pnpm/action-setup@v4
+    - uses: pnpm/action-setup@v6
      with:
        version: 11.1.1
-    - uses: actions/setup-node@v4
+    - 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
@@ -86,14 +86,14 @@ jobs:
        shell: bash
        working-directory: nodejs
    steps:
-    - uses: actions/checkout@v4
+    - uses: actions/checkout@v6
      with:
        fetch-depth: 0
        lfs: true
-    - uses: pnpm/action-setup@v4
+    - uses: pnpm/action-setup@v6
      with:
        version: 11.1.1
-    - uses: actions/setup-node@v4
+    - 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
@@ -130,7 +130,7 @@ jobs:
          echo "Run 'pnpm run docs', fix any warnings, and commit the changes."
          exit 1
        fi
-    - uses: actions/setup-node@v4
+    - uses: actions/setup-node@v6
      name: Setup Node.js ${{ matrix.node-version }} for test
      with:
        node-version: ${{ matrix.node-version }}
@@ -166,14 +166,14 @@ jobs:
        shell: bash
        working-directory: nodejs
    steps:
-    - uses: actions/checkout@v4
+    - uses: actions/checkout@v6
      with:
        fetch-depth: 0
        lfs: true
-    - uses: pnpm/action-setup@v4
+    - uses: pnpm/action-setup@v6
      with:
        version: 11.1.1
-    - uses: actions/setup-node@v4
+    - uses: actions/setup-node@v6
      with:
        # pnpm 11 requires Node >= 22.13; use 24 since 22 hits EOL
        # in October.
@@ -32,7 +32,7 @@ jobs:
    permissions:
      contents: write
    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
        with:
          fetch-depth: 0
          lfs: true
@@ -103,7 +103,7 @@ jobs:
            features: fp16kernels
            pre_build: brew install protobuf
          - target: x86_64-pc-windows-msvc
-            host: windows-latest
+            host: windows-2025-8x-x64
            features: ","
            pre_build: |-
              choco install --no-progress protoc ninja nasm
@@ -111,12 +111,21 @@ jobs:
              # There is an issue where choco doesn't add nasm to the path
              export PATH="$PATH:/c/Program Files/NASM"
              nasm -v
+              # Fat LTO of the cdylib is single-threaded and the peak-memory
+              # step of the build, and had started hitting rustc-LLVM OOM on the
+              # Windows runners. ThinLTO parallelizes it across the runner's
+              # cores and keeps peak memory well under the limit.
+              export CARGO_PROFILE_RELEASE_LTO=thin
+              export CARGO_PROFILE_RELEASE_CODEGEN_UNITS=16
          - target: aarch64-pc-windows-msvc
-            host: windows-latest
+            host: windows-2025-8x-x64
            features: ","
            pre_build: |-
                choco install --no-progress protoc
                rustup target add aarch64-pc-windows-msvc
+                # See ThinLTO note on the x86_64-pc-windows-msvc target above.
+                export CARGO_PROFILE_RELEASE_LTO=thin
+                export CARGO_PROFILE_RELEASE_CODEGEN_UNITS=16
          - target: x86_64-unknown-linux-gnu
            host: ubuntu-latest
            features: fp16kernels
@@ -170,13 +179,13 @@ jobs:
      run:
        working-directory: nodejs
    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
      - name: Setup pnpm
-        uses: pnpm/action-setup@v4
+        uses: pnpm/action-setup@v6
        with:
          version: 11.1.1
      - name: Setup node
-        uses: actions/setup-node@v4
+        uses: actions/setup-node@v6
        with:
          # pnpm 11 requires Node >= 22.13; use 24 since 22 hits EOL
          # in October.
@@ -190,7 +199,7 @@ jobs:
          toolchain: stable
          targets: ${{ matrix.settings.target }}
      - name: Cache cargo
-        uses: actions/cache@v4
+        uses: actions/cache@v5
        with:
          path: |
            ~/.cargo/registry/index/
@@ -244,7 +253,7 @@ jobs:
        if: ${{ !matrix.settings.docker }}
        shell: bash
      - name: Upload artifact
-        uses: actions/upload-artifact@v4
+        uses: actions/upload-artifact@v7
        with:
          name: lancedb-${{ matrix.settings.target }}
          path: nodejs/dist/*.node
@@ -256,7 +265,7 @@ jobs:
        run: pnpm tsc
      - name: Upload Generic Artifacts
        if: ${{ matrix.settings.target == 'aarch64-apple-darwin' }}
-        uses: actions/upload-artifact@v4
+        uses: actions/upload-artifact@v7
        with:
          name: nodejs-dist
          path: |
@@ -287,13 +296,13 @@ jobs:
        shell: bash
        working-directory: nodejs
    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
      - name: Setup pnpm
-        uses: pnpm/action-setup@v4
+        uses: pnpm/action-setup@v6
        with:
          version: 11.1.1
      - name: Setup Node.js 24 for install
-        uses: actions/setup-node@v4
+        uses: actions/setup-node@v6
        with:
          # pnpm 11 requires Node >= 22.13; use 24 since 22 hits EOL
          # in October.
@@ -303,18 +312,18 @@ jobs:
      - name: Install dependencies
        run: pnpm install --frozen-lockfile
      - name: Setup Node.js ${{ matrix.node }} for test
-        uses: actions/setup-node@v4
+        uses: actions/setup-node@v6
        with:
          node-version: ${{ matrix.node }}
      - name: Download artifacts
-        uses: actions/download-artifact@v4
+        uses: actions/download-artifact@v8
        with:
          name: lancedb-${{ matrix.settings.target }}
          path: nodejs/dist/
          # For testing purposes:
          # run-id: 13982782871
          # github-token: ${{ secrets.GITHUB_TOKEN }} # token with actions:read permissions on target repo
-      - uses: actions/download-artifact@v4
+      - uses: actions/download-artifact@v8
        with:
          name: nodejs-dist
          path: nodejs/dist
@@ -339,13 +348,13 @@ jobs:
    needs:
      - test-lancedb
    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
      - name: Setup pnpm
-        uses: pnpm/action-setup@v4
+        uses: pnpm/action-setup@v6
        with:
          version: 11.1.1
      - name: Setup node
-        uses: actions/setup-node@v4
+        uses: actions/setup-node@v6
        with:
          node-version: 24
          cache: pnpm
@@ -353,14 +362,14 @@ jobs:
          registry-url: "https://registry.npmjs.org"
      - name: Install dependencies
        run: pnpm install --frozen-lockfile
-      - uses: actions/download-artifact@v4
+      - uses: actions/download-artifact@v8
        with:
          name: nodejs-dist
          path: nodejs/dist
          # For testing purposes:
          # run-id: 13982782871
          # github-token: ${{ secrets.GITHUB_TOKEN }} # token with actions:read permissions on target repo
-      - uses: actions/download-artifact@v4
+      - uses: actions/download-artifact@v8
        name: Download arch-specific binaries
        with:
          pattern: lancedb-*
@@ -398,7 +407,7 @@ jobs:
      contents: read
      issues: write
    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
      - uses: ./.github/actions/create-failure-issue
        with:
          job-results: ${{ toJSON(needs) }}
@@ -41,7 +41,7 @@ jobs:
        shell: bash
        working-directory: python
    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
        with:
          fetch-depth: 0
          lfs: true
@@ -66,7 +66,7 @@ jobs:
        shell: bash
        working-directory: python
    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
        with:
          fetch-depth: 0
          lfs: true
@@ -95,7 +95,7 @@ jobs:
        shell: bash
        working-directory: python
    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
        with:
          fetch-depth: 0
          lfs: true
@@ -126,7 +126,7 @@ jobs:
        shell: bash
        working-directory: python
    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
        with:
          fetch-depth: 0
          lfs: true
@@ -160,7 +160,7 @@ jobs:
        shell: bash
        working-directory: python
    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
        with:
          fetch-depth: 0
          lfs: true
@@ -189,7 +189,7 @@ jobs:
        shell: bash
        working-directory: python
    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
        with:
          fetch-depth: 0
          lfs: true
@@ -212,7 +212,7 @@ jobs:
        shell: bash
        working-directory: python
    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
        with:
          fetch-depth: 0
          lfs: true
@@ -40,7 +40,7 @@ jobs:
      CC: clang-18
      CXX: clang++-18
    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
        with:
          fetch-depth: 0
          lfs: true
@@ -65,7 +65,7 @@ jobs:
    timeout-minutes: 10
    runs-on: ubuntu-24.04
    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
      - uses: EmbarkStudios/cargo-deny-action@v2
        with:
          command: check advisories bans licenses sources
@@ -78,7 +78,7 @@ jobs:
      CC: clang
      CXX: clang++
    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
      # Building without a lock file often requires the latest Rust version since downstream
      # dependencies may have updated their minimum Rust version.
      - uses: actions-rust-lang/setup-rust-toolchain@v1
@@ -113,7 +113,7 @@ jobs:
      CXX: clang++-18
      GH_TOKEN: ${{ secrets.SOPHON_READ_TOKEN }}
    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
        with:
          fetch-depth: 0
          lfs: true
@@ -152,7 +152,7 @@ jobs:
        shell: bash
        working-directory: rust
    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
        with:
          fetch-depth: 0
          lfs: true
@@ -181,7 +181,7 @@ jobs:
      run:
        working-directory: rust/lancedb
    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
      - name: Set target
        run: rustup target add ${{ matrix.target }}
      - uses: Swatinem/rust-cache@v2
@@ -210,7 +210,7 @@ jobs:
      CC: clang-18
      CXX: clang++-18
    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
        with:
          submodules: true
      - name: Install dependencies
@@ -11,7 +11,7 @@ jobs:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@v6
        with:
          ref: main
          persist-credentials: false
@@ -11,7 +11,7 @@ jobs:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@v6
        with:
          ref: main
          persist-credentials: false
@@ -2039,9 +2039,9 @@ dependencies = [

 [[package]]
 name = "crossbeam-epoch"
-version = "0.9.18"
+version = "0.9.20"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e"
+checksum = "2d6914041f254d6e9176c01941b21115dcfb7089e55135a35411081bd106ef3f"
 dependencies = [
 "crossbeam-utils",
 ]
@@ -3151,6 +3151,12 @@ dependencies = [
 "encoding_rs",
 ]

+[[package]]
+name = "endian-type"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c34f04666d835ff5d62e058c3995147c06f42fe86ff053337632bca83e42702d"
+
 [[package]]
 name = "enum-as-inner"
 version = "0.6.1"
@@ -3423,8 +3429,8 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"

 [[package]]
 name = "fsst"
-version = "9.0.0-beta.10"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.19"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.19#8f0e6d3a7c53438275b134c0ac1afbc80600616e"
 dependencies = [
 "arrow-array",
 "rand 0.9.4",
@@ -3779,6 +3785,30 @@ dependencies = [
 "wasm-bindgen",
 ]

+[[package]]
+name = "goosefs-sdk"
+version = "0.1.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9ae079b88ffe7772d12cfc5c40a5a324babb357893d95b5e3a22ae857f236c5f"
+dependencies = [
+ "async-trait",
+ "bytes",
+ "dashmap",
+ "hostname",
+ "prost",
+ "prost-types",
+ "rand 0.9.4",
+ "reqwest 0.12.28",
+ "serde",
+ "thiserror 2.0.18",
+ "tokio",
+ "tokio-stream",
+ "tonic",
+ "tonic-prost",
+ "tracing",
+ "uuid",
+]
+
 [[package]]
 name = "group"
 version = "0.12.1"
@@ -3990,6 +4020,17 @@ dependencies = [
 "windows-sys 0.61.2",
 ]

+[[package]]
+name = "hostname"
+version = "0.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "617aaa3557aef3810a6369d0a99fac8a080891b68bd9f9812a1eeda0c0730cbd"
+dependencies = [
+ "cfg-if 1.0.4",
+ "libc",
+ "windows-link",
+]
+
 [[package]]
 name = "http"
 version = "0.2.12"
@@ -4150,6 +4191,19 @@ dependencies = [
 "webpki-roots 1.0.7",
 ]

+[[package]]
+name = "hyper-timeout"
+version = "0.5.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2b90d566bffbce6a75bd8b09a05aa8c2cb1fabb6cb348f8840c9e4c90a0d83b0"
+dependencies = [
+ "hyper 1.9.0",
+ "hyper-util",
+ "pin-project-lite",
+ "tokio",
+ "tower-service",
+]
+
 [[package]]
 name = "hyper-util"
 version = "0.1.20"
@@ -4673,7 +4727,7 @@ dependencies = [
 "jiff",
 "nom 8.0.0",
 "num-traits",
- "ordered-float",
+ "ordered-float 5.3.0",
 "rand 0.9.4",
 "serde",
 "serde_json",
@@ -4726,8 +4780,8 @@ checksum = "e037a2e1d8d5fdbd49b16a4ea09d5d6401c1f29eca5ff29d03d3824dba16256a"

 [[package]]
 name = "lance"
-version = "9.0.0-beta.10"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.19"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.19#8f0e6d3a7c53438275b134c0ac1afbc80600616e"
 dependencies = [
 "arc-swap",
 "arrow",
@@ -4801,8 +4855,8 @@ dependencies = [

 [[package]]
 name = "lance-arrow"
-version = "9.0.0-beta.10"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.19"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.19#8f0e6d3a7c53438275b134c0ac1afbc80600616e"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4811,6 +4865,7 @@ dependencies = [
 "arrow-ord",
 "arrow-schema",
 "arrow-select",
+ "bytemuck",
 "bytes",
 "futures",
 "getrandom 0.2.17",
@@ -4823,7 +4878,7 @@ dependencies = [
 [[package]]
 name = "lance-arrow-scalar"
 version = "58.0.0"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.19#8f0e6d3a7c53438275b134c0ac1afbc80600616e"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4837,7 +4892,7 @@ dependencies = [
 [[package]]
 name = "lance-arrow-stats"
 version = "58.0.0"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.19#8f0e6d3a7c53438275b134c0ac1afbc80600616e"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -4846,8 +4901,8 @@ dependencies = [

 [[package]]
 name = "lance-bitpacking"
-version = "9.0.0-beta.10"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.19"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.19#8f0e6d3a7c53438275b134c0ac1afbc80600616e"
 dependencies = [
 "arrayref",
 "crunchy",
@@ -4857,8 +4912,8 @@ dependencies = [

 [[package]]
 name = "lance-core"
-version = "9.0.0-beta.10"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.19"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.19#8f0e6d3a7c53438275b134c0ac1afbc80600616e"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4896,8 +4951,8 @@ dependencies = [

 [[package]]
 name = "lance-datafusion"
-version = "9.0.0-beta.10"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.19"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.19#8f0e6d3a7c53438275b134c0ac1afbc80600616e"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -4927,8 +4982,8 @@ dependencies = [

 [[package]]
 name = "lance-datagen"
-version = "9.0.0-beta.10"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.19"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.19#8f0e6d3a7c53438275b134c0ac1afbc80600616e"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -4945,8 +5000,8 @@ dependencies = [

 [[package]]
 name = "lance-derive"
-version = "9.0.0-beta.10"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.19"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.19#8f0e6d3a7c53438275b134c0ac1afbc80600616e"
 dependencies = [
 "proc-macro2",
 "quote",
@@ -4955,8 +5010,8 @@ dependencies = [

 [[package]]
 name = "lance-encoding"
-version = "9.0.0-beta.10"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.19"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.19#8f0e6d3a7c53438275b134c0ac1afbc80600616e"
 dependencies = [
 "arrow-arith",
 "arrow-array",
@@ -4991,8 +5046,8 @@ dependencies = [

 [[package]]
 name = "lance-file"
-version = "9.0.0-beta.10"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.19"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.19#8f0e6d3a7c53438275b134c0ac1afbc80600616e"
 dependencies = [
 "arrow-arith",
 "arrow-array",
@@ -5022,8 +5077,8 @@ dependencies = [

 [[package]]
 name = "lance-index"
-version = "9.0.0-beta.10"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.19"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.19#8f0e6d3a7c53438275b134c0ac1afbc80600616e"
 dependencies = [
 "arc-swap",
 "arrow",
@@ -5088,8 +5143,8 @@ dependencies = [

 [[package]]
 name = "lance-io"
-version = "9.0.0-beta.10"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.19"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.19#8f0e6d3a7c53438275b134c0ac1afbc80600616e"
 dependencies = [
 "arrow",
 "arrow-arith",
@@ -5113,6 +5168,7 @@ dependencies = [
 "lance-core",
 "lance-namespace",
 "log",
+ "metrics",
 "moka",
 "object_store",
 "object_store_opendal",
@@ -5130,8 +5186,8 @@ dependencies = [

 [[package]]
 name = "lance-linalg"
-version = "9.0.0-beta.10"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.19"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.19#8f0e6d3a7c53438275b134c0ac1afbc80600616e"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -5147,8 +5203,8 @@ dependencies = [

 [[package]]
 name = "lance-namespace"
-version = "9.0.0-beta.10"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.19"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.19#8f0e6d3a7c53438275b134c0ac1afbc80600616e"
 dependencies = [
 "arrow",
 "async-trait",
@@ -5160,8 +5216,8 @@ dependencies = [

 [[package]]
 name = "lance-namespace-impls"
-version = "9.0.0-beta.10"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.19"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.19#8f0e6d3a7c53438275b134c0ac1afbc80600616e"
 dependencies = [
 "arrow",
 "arrow-ipc",
@@ -5184,7 +5240,7 @@ dependencies = [
 "lance-table",
 "log",
 "object_store",
- "quick-xml 0.38.4",
+ "quick-xml 0.40.1",
 "rand 0.9.4",
 "reqwest 0.12.28",
 "roaring",
@@ -5215,8 +5271,8 @@ dependencies = [

 [[package]]
 name = "lance-select"
-version = "9.0.0-beta.10"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.19"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.19#8f0e6d3a7c53438275b134c0ac1afbc80600616e"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -5231,8 +5287,8 @@ dependencies = [

 [[package]]
 name = "lance-table"
-version = "9.0.0-beta.10"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.19"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.19#8f0e6d3a7c53438275b134c0ac1afbc80600616e"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5271,22 +5327,22 @@ dependencies = [

 [[package]]
 name = "lance-testing"
-version = "9.0.0-beta.10"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.19"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.19#8f0e6d3a7c53438275b134c0ac1afbc80600616e"
 dependencies = [
 "arrow-array",
 "arrow-schema",
 "criterion",
 "lance-arrow",
 "num-traits",
- "pprof",
+ "pprof 0.15.0",
 "rand 0.9.4",
 ]

 [[package]]
 name = "lance-tokenizer"
-version = "9.0.0-beta.10"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.19"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.19#8f0e6d3a7c53438275b134c0ac1afbc80600616e"
 dependencies = [
 "icu_segmenter",
 "jieba-rs",
@@ -5299,12 +5355,13 @@ dependencies = [

 [[package]]
 name = "lancedb"
-version = "0.31.0-beta.4"
+version = "0.32.0-beta.0"
 dependencies = [
 "ahash",
 "anyhow",
 "arrow",
 "arrow-array",
+ "arrow-buffer",
 "arrow-cast",
 "arrow-data",
 "arrow-ipc",
@@ -5354,12 +5411,15 @@ dependencies = [
 "lance-testing",
 "lazy_static",
 "log",
+ "metrics",
+ "metrics-util",
 "moka",
 "num-traits",
 "object_store",
 "pin-project",
 "polars",
 "polars-arrow",
+ "pprof 0.14.1",
 "rand 0.9.4",
 "random_word",
 "regex",
@@ -5383,7 +5443,7 @@ dependencies = [

 [[package]]
 name = "lancedb-nodejs"
-version = "0.31.0-beta.4"
+version = "0.32.0-beta.0"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -5408,7 +5468,7 @@ dependencies = [

 [[package]]
 name = "lancedb-python"
-version = "0.34.0-beta.4"
+version = "0.35.0-beta.0"
 dependencies = [
 "arrow",
 "async-trait",
@@ -5810,6 +5870,36 @@ dependencies = [
 "stable_deref_trait",
 ]

+[[package]]
+name = "metrics"
+version = "0.24.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "89550ee9f79e88fef3119de263694973a8adb26c21d75322164fb8c493039fe2"
+dependencies = [
+ "portable-atomic",
+ "rapidhash",
+]
+
+[[package]]
+name = "metrics-util"
+version = "0.19.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b8496cc523d1f94c1385dd8f0f0c2c480b2b8aeccb5b7e4485ad6365523ae376"
+dependencies = [
+ "aho-corasick",
+ "crossbeam-epoch",
+ "crossbeam-utils",
+ "hashbrown 0.15.5",
+ "indexmap 2.14.0",
+ "metrics",
+ "ordered-float 4.6.0",
+ "quanta",
+ "radix_trie",
+ "rand 0.9.4",
+ "rand_xoshiro",
+ "sketches-ddsketch",
+]
+
 [[package]]
 name = "mime"
 version = "0.3.17"
@@ -5951,9 +6041,9 @@ dependencies = [

 [[package]]
 name = "napi"
-version = "3.9.4"
+version = "3.10.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b41bda2ac390efb5e8d22025d925ccc3f3807d8c1bea6d19b36127247c4b8f83"
+checksum = "0c71997d6f7ad4a756966e452426848ac27d3b37a295302d63afbbcce0270f93"
 dependencies = [
 "bitflags 2.11.1",
 "chrono",
@@ -5976,9 +6066,9 @@ checksum = "c9c366d2c8c60b86fa632df75f745509b52f9128f91a6bad4c796e44abb505e1"

 [[package]]
 name = "napi-derive"
-version = "3.5.7"
+version = "3.5.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "61d66f70256ad5aef58659966064471d0ad90e2897bc36a5a5e0389c85aabc1e"
+checksum = "d4ba572deef53e2c386759a8c2014175a62679d74ff83adc205c8bc0e0285727"
 dependencies = [
 "convert_case",
 "ctor 1.0.5",
@@ -5990,9 +6080,9 @@ dependencies = [

 [[package]]
 name = "napi-derive-backend"
-version = "5.0.5"
+version = "5.1.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "81b4b08f15eed7a2a20c3f4c6314013fc3ac890a3afa9892b594485299ebdb2d"
+checksum = "ddd961eb2aa8965e3f29722d754f3a86907eb1984e2fbcbe3fe87b9a02d6bfba"
 dependencies = [
 "convert_case",
 "proc-macro2",
@@ -6025,6 +6115,15 @@ dependencies = [
 "rawpointer",
 ]

+[[package]]
+name = "nibble_vec"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "77a5d83df9f36fe23f0c3648c6bbb8b0298bb5f1939c8f2704431371f4b84d43"
+dependencies = [
+ "smallvec",
+]
+
 [[package]]
 name = "nix"
 version = "0.26.4"
@@ -6340,7 +6439,9 @@ dependencies = [
 "opendal-layer-timeout",
 "opendal-service-azblob",
 "opendal-service-azdls",
+ "opendal-service-cos",
 "opendal-service-gcs",
+ "opendal-service-goosefs",
 "opendal-service-hf",
 "opendal-service-oss",
 "opendal-service-s3",
@@ -6468,6 +6569,23 @@ dependencies = [
 "opendal-core",
 ]

+[[package]]
+name = "opendal-service-cos"
+version = "0.57.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "aa8cafe9729213375c7331019b0cb756ad3e1aff7f45cd32c45eae91ebde8901"
+dependencies = [
+ "bytes",
+ "http 1.4.2",
+ "log",
+ "opendal-core",
+ "quick-xml 0.39.4",
+ "reqsign-core",
+ "reqsign-file-read-tokio",
+ "reqsign-tencent-cos",
+ "serde",
+]
+
 [[package]]
 name = "opendal-service-gcs"
 version = "0.57.0"
@@ -6489,6 +6607,20 @@ dependencies = [
 "tokio",
 ]

+[[package]]
+name = "opendal-service-goosefs"
+version = "0.57.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "69e43048bde419947ba826fbdc2f134d6c03f44ebf48bd33a03b72f9fc45fcb4"
+dependencies = [
+ "bytes",
+ "goosefs-sdk",
+ "log",
+ "opendal-core",
+ "serde",
+ "tokio",
+]
+
 [[package]]
 name = "opendal-service-hf"
 version = "0.57.0"
@@ -6556,6 +6688,15 @@ version = "0.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "04744f49eae99ab78e0d5c0b603ab218f515ea8cfe5a456d7629ad883a3b6e7d"

+[[package]]
+name = "ordered-float"
+version = "4.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7bb71e1b3fa6ca1c61f383464aaf2bb0e2f8e772a1f01d486832464de363b951"
+dependencies = [
+ "num-traits",
+]
+
 [[package]]
 name = "ordered-float"
 version = "5.3.0"
@@ -7325,6 +7466,28 @@ version = "0.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "439ee305def115ba05938db6eb1644ff94165c5ab5e9420d1c1bcedbba909391"

+[[package]]
+name = "pprof"
+version = "0.14.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "afad4d4df7b31280028245f152d5a575083e2abb822d05736f5e47653e77689f"
+dependencies = [
+ "aligned-vec",
+ "backtrace",
+ "cfg-if 1.0.4",
+ "findshlibs",
+ "inferno",
+ "libc",
+ "log",
+ "nix",
+ "once_cell",
+ "smallvec",
+ "spin 0.10.0",
+ "symbolic-demangle",
+ "tempfile",
+ "thiserror 1.0.69",
+]
+
 [[package]]
 name = "pprof"
 version = "0.15.0"
@@ -7574,19 +7737,25 @@ dependencies = [
 ]

 [[package]]
-name = "quick-xml"
-version = "0.26.0"
+name = "quanta"
+version = "0.12.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7f50b1c63b38611e7d4d7f68b82d3ad0cc71a2ad2e7f61fc10f1328d917c93cd"
+checksum = "f3ab5a9d756f0d97bdc89019bd2e4ea098cf9cde50ee7564dde6b81ccc8f06c7"
 dependencies = [
- "memchr",
+ "crossbeam-utils",
+ "libc",
+ "once_cell",
+ "raw-cpuid",
+ "wasi 0.11.1+wasi-snapshot-preview1",
+ "web-sys",
+ "winapi",
 ]

 [[package]]
 name = "quick-xml"
-version = "0.38.4"
+version = "0.26.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b66c2058c55a409d601666cffe35f04333cf1013010882cec174a7467cd4e21c"
+checksum = "7f50b1c63b38611e7d4d7f68b82d3ad0cc71a2ad2e7f61fc10f1328d917c93cd"
 dependencies = [
 "memchr",
 ]
@@ -7601,6 +7770,15 @@ dependencies = [
 "serde",
 ]

+[[package]]
+name = "quick-xml"
+version = "0.40.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2474bd2e5029e7ccb6abb2ba48cf2383a333851dedf495901544281590c7da7f"
+dependencies = [
+ "memchr",
+]
+
 [[package]]
 name = "quinn"
 version = "0.11.9"
@@ -7684,6 +7862,16 @@ version = "0.7.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "dc33ff2d4973d518d823d61aa239014831e521c75da58e3df4840d3f47749d09"

+[[package]]
+name = "radix_trie"
+version = "0.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c069c179fcdc6a2fe24d8d18305cf085fdbd4f922c041943e203685d6a1c58fd"
+dependencies = [
+ "endian-type",
+ "nibble_vec",
+]
+
 [[package]]
 name = "rancor"
 version = "0.1.1"
@@ -7818,6 +8006,15 @@ version = "1.7.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "973443cf09a9c8656b574a866ab68dfa19f0867d0340648c7d2f6a71b8a8ea68"

+[[package]]
+name = "rapidhash"
+version = "4.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5da7e78a036ce858e8d55b7e7dc8ba3a88b78350fd2155d3591bbd966b58589e"
+dependencies = [
+ "rustversion",
+]
+
 [[package]]
 name = "raw-cpuid"
 version = "11.6.0"
@@ -8102,6 +8299,21 @@ dependencies = [
 "tokio",
 ]

+[[package]]
+name = "reqsign-tencent-cos"
+version = "3.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e128f19525861dbded59e1e7c17653a8ed63d573ca04aed708d552dbef5bb32a"
+dependencies = [
+ "anyhow",
+ "http 1.4.2",
+ "log",
+ "percent-encoding",
+ "reqsign-core",
+ "serde",
+ "serde_json",
+]
+
 [[package]]
 name = "reqwest"
 version = "0.12.28"
@@ -8980,6 +9192,12 @@ version = "1.0.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8ee5873ec9cce0195efcb7a4e9507a04cd49aec9c83d0389df45b1ef7ba2e649"

+[[package]]
+name = "sketches-ddsketch"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0c6f73aeb92d671e0cc4dca167e59b2deb6387c375391bc99ee743f326994a2b"
+
 [[package]]
 name = "slab"
 version = "0.4.12"
@@ -9787,6 +10005,45 @@ dependencies = [
 "winnow",
 ]

+[[package]]
+name = "tonic"
+version = "0.14.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ac2a5518c70fa84342385732db33fb3f44bc4cc748936eb5833d2df34d6445ef"
+dependencies = [
+ "async-trait",
+ "base64 0.22.1",
+ "bytes",
+ "h2 0.4.14",
+ "http 1.4.2",
+ "http-body 1.0.1",
+ "http-body-util",
+ "hyper 1.9.0",
+ "hyper-timeout",
+ "hyper-util",
+ "percent-encoding",
+ "pin-project",
+ "socket2 0.6.3",
+ "sync_wrapper",
+ "tokio",
+ "tokio-stream",
+ "tower",
+ "tower-layer",
+ "tower-service",
+ "tracing",
+]
+
+[[package]]
+name = "tonic-prost"
+version = "0.14.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "50849f68853be452acf590cde0b146665b8d507b3b8af17261df47e02c209ea0"
+dependencies = [
+ "bytes",
+ "prost",
+ "tonic",
+]
+
 [[package]]
 name = "tower"
 version = "0.5.3"
@@ -9795,9 +10052,12 @@ checksum = "ebe5ef63511595f1344e2d5cfa636d973292adc0eec1f0ad45fae9f0851ab1d4"
 dependencies = [
 "futures-core",
 "futures-util",
+ "indexmap 2.14.0",
 "pin-project-lite",
+ "slab",
 "sync_wrapper",
 "tokio",
+ "tokio-util",
 "tower-layer",
 "tower-service",
 "tracing",
@@ -13,24 +13,25 @@ categories = ["database-implementations"]
 rust-version = "1.91.0"

 [workspace.dependencies]
-lance = { "version" = "=9.0.0-beta.10", default-features = false, "tag" = "v9.0.0-beta.10", "git" = "https://github.com/lance-format/lance.git" }
-lance-core = { "version" = "=9.0.0-beta.10", "tag" = "v9.0.0-beta.10", "git" = "https://github.com/lance-format/lance.git" }
-lance-datagen = { "version" = "=9.0.0-beta.10", "tag" = "v9.0.0-beta.10", "git" = "https://github.com/lance-format/lance.git" }
-lance-file = { "version" = "=9.0.0-beta.10", "tag" = "v9.0.0-beta.10", "git" = "https://github.com/lance-format/lance.git" }
-lance-io = { "version" = "=9.0.0-beta.10", default-features = false, "tag" = "v9.0.0-beta.10", "git" = "https://github.com/lance-format/lance.git" }
-lance-index = { "version" = "=9.0.0-beta.10", "tag" = "v9.0.0-beta.10", "git" = "https://github.com/lance-format/lance.git" }
-lance-linalg = { "version" = "=9.0.0-beta.10", "tag" = "v9.0.0-beta.10", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace = { "version" = "=9.0.0-beta.10", "tag" = "v9.0.0-beta.10", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace-impls = { "version" = "=9.0.0-beta.10", default-features = false, "tag" = "v9.0.0-beta.10", "git" = "https://github.com/lance-format/lance.git" }
-lance-table = { "version" = "=9.0.0-beta.10", "tag" = "v9.0.0-beta.10", "git" = "https://github.com/lance-format/lance.git" }
-lance-testing = { "version" = "=9.0.0-beta.10", "tag" = "v9.0.0-beta.10", "git" = "https://github.com/lance-format/lance.git" }
-lance-datafusion = { "version" = "=9.0.0-beta.10", "tag" = "v9.0.0-beta.10", "git" = "https://github.com/lance-format/lance.git" }
-lance-encoding = { "version" = "=9.0.0-beta.10", "tag" = "v9.0.0-beta.10", "git" = "https://github.com/lance-format/lance.git" }
-lance-arrow = { "version" = "=9.0.0-beta.10", "tag" = "v9.0.0-beta.10", "git" = "https://github.com/lance-format/lance.git" }
+lance = { "version" = "=9.0.0-beta.19", default-features = false, "tag" = "v9.0.0-beta.19", "git" = "https://github.com/lance-format/lance.git" }
+lance-core = { "version" = "=9.0.0-beta.19", "tag" = "v9.0.0-beta.19", "git" = "https://github.com/lance-format/lance.git" }
+lance-datagen = { "version" = "=9.0.0-beta.19", "tag" = "v9.0.0-beta.19", "git" = "https://github.com/lance-format/lance.git" }
+lance-file = { "version" = "=9.0.0-beta.19", "tag" = "v9.0.0-beta.19", "git" = "https://github.com/lance-format/lance.git" }
+lance-io = { "version" = "=9.0.0-beta.19", default-features = false, "tag" = "v9.0.0-beta.19", "git" = "https://github.com/lance-format/lance.git" }
+lance-index = { "version" = "=9.0.0-beta.19", "tag" = "v9.0.0-beta.19", "git" = "https://github.com/lance-format/lance.git" }
+lance-linalg = { "version" = "=9.0.0-beta.19", "tag" = "v9.0.0-beta.19", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace = { "version" = "=9.0.0-beta.19", "tag" = "v9.0.0-beta.19", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace-impls = { "version" = "=9.0.0-beta.19", default-features = false, "tag" = "v9.0.0-beta.19", "git" = "https://github.com/lance-format/lance.git" }
+lance-table = { "version" = "=9.0.0-beta.19", "tag" = "v9.0.0-beta.19", "git" = "https://github.com/lance-format/lance.git" }
+lance-testing = { "version" = "=9.0.0-beta.19", "tag" = "v9.0.0-beta.19", "git" = "https://github.com/lance-format/lance.git" }
+lance-datafusion = { "version" = "=9.0.0-beta.19", "tag" = "v9.0.0-beta.19", "git" = "https://github.com/lance-format/lance.git" }
+lance-encoding = { "version" = "=9.0.0-beta.19", "tag" = "v9.0.0-beta.19", "git" = "https://github.com/lance-format/lance.git" }
+lance-arrow = { "version" = "=9.0.0-beta.19", "tag" = "v9.0.0-beta.19", "git" = "https://github.com/lance-format/lance.git" }
 ahash = "0.8"
 # Note that this one does not include pyarrow
 arrow = { version = "58.0.0", optional = false }
 arrow-array = "58.0.0"
+arrow-buffer = "58.0.0"
 arrow-data = "58.0.0"
 arrow-ipc = "58.0.0"
 arrow-ord = "58.0.0"
@@ -53,6 +54,8 @@ half = { "version" = "2.7.1", default-features = false, features = [
 ] }
 futures = "0"
 log = "0.4"
+metrics = "0.24"
+metrics-util = "0.19"
 moka = { version = "0.12", features = ["future"] }
 object_store = "0.13.2"
 pin-project = "1.0.7"
@@ -51,18 +51,6 @@ ignore = [
    # https://rustsec.org/advisories/RUSTSEC-2024-0436
    { id = "RUSTSEC-2024-0436", reason = "transitive via datafusion; awaiting ecosystem migration" },

-    # encoding: unmaintained. Reached through lindera-dictionary, which is
-    # required by the native Lindera tokenizer path. Lindera has not migrated
-    # off this crate yet.
-    # https://rustsec.org/advisories/RUSTSEC-2021-0153
-    { id = "RUSTSEC-2021-0153", reason = "transitive via lindera-dictionary for native Lindera tokenizer" },
-
-    # fast-float: unsound and unmaintained. Reached only through polars-arrow
-    # from the optional Polars integration; replacement requires a Polars
-    # dependency upgrade.
-    # https://rustsec.org/advisories/RUSTSEC-2024-0379
-    { id = "RUSTSEC-2024-0379", reason = "transitive via polars-arrow; waiting on Polars migration" },
-
    # tantivy: segfault on malformed input due to missing bounds check.
    # Pulled in via lance for full-text search. We only feed tantivy
    # documents we construct ourselves, not attacker-controlled bytes.
@@ -80,18 +68,6 @@ ignore = [
    # https://rustsec.org/advisories/RUSTSEC-2025-0119
    { id = "RUSTSEC-2025-0119", reason = "transitive via hf-hub/indicatif; cosmetic formatting crate" },

-    # bincode: unmaintained. Reached through lindera and lindera-dictionary,
-    # which are required by the native Lindera tokenizer path. Lindera has not
-    # migrated to another serialization format yet.
-    # https://rustsec.org/advisories/RUSTSEC-2025-0141
-    { id = "RUSTSEC-2025-0141", reason = "transitive via lindera/lindera-dictionary for native Lindera tokenizer" },
-
-    # lru: soundness issue in IterMut. Reached only through aws-sdk-s3 in
-    # LanceDB's dev-dependency graph; LanceDB does not use that iterator
-    # directly. Clearing this requires the AWS SDK chain to update lru.
-    # https://rustsec.org/advisories/RUSTSEC-2026-0002
-    { id = "RUSTSEC-2026-0002", reason = "transitive via aws-sdk-s3 dev-dependency; waiting on AWS SDK lru upgrade" },
-
    # rustls-webpki 0.101.7 (old major line): name-constraint checks for
    # URI / wildcard names. Pulled in only via the legacy rustls 0.21 chain
    # from aws-smithy-http-client. The 0.103 line we actively use is patched.
@@ -108,17 +84,23 @@ ignore = [
    # https://rustsec.org/advisories/RUSTSEC-2026-0104
    { id = "RUSTSEC-2026-0104", reason = "only affects rustls-webpki 0.101 from legacy aws-smithy/rustls 0.21 chain" },

-    # rand 0.8.5: soundness issue only when ThreadRng reseeds inside a custom
-    # logger. Reached through several transitive chains. LanceDB does not use
-    # rand from a custom logger; upgrade once all pinned chains accept 0.8.6+.
-    # https://rustsec.org/advisories/RUSTSEC-2026-0097
-    { id = "RUSTSEC-2026-0097", reason = "transitive rand 0.8.5; LanceDB does not call ThreadRng from custom logging" },
-
    # pyo3 advisories in the Python bindings; tracked pending a patched pyo3 release.
    # https://rustsec.org/advisories/RUSTSEC-2026-0176
    # https://rustsec.org/advisories/RUSTSEC-2026-0177
    { id = "RUSTSEC-2026-0176", reason = "pyo3 in Python bindings; awaiting patched pyo3 release" },
    { id = "RUSTSEC-2026-0177", reason = "pyo3 in Python bindings; awaiting patched pyo3 release" },
+
+    # quick-xml < 0.41.0: quadratic runtime on duplicate attribute names (DoS).
+    # quick-xml < 0.41.0: unbounded namespace-declaration allocation in NsReader (DoS).
+    # Pulled in transitively by inferno (dev-only flame-graph dep), lance-namespace-impls
+    # (git dep from lance), and opendal/reqsign (cloud storage XML parsing). The XML
+    # parsed by opendal/reqsign comes from trusted cloud-storage endpoints (S3, GCS,
+    # Azure), not attacker-controlled input. Clearing requires upstream crates to migrate
+    # to quick-xml >= 0.41.0.
+    # https://rustsec.org/advisories/RUSTSEC-2026-0194
+    # https://rustsec.org/advisories/RUSTSEC-2026-0195
+    { id = "RUSTSEC-2026-0194", reason = "transitive via inferno/lance/opendal; XML from trusted cloud endpoints, not attacker-controlled" },
+    { id = "RUSTSEC-2026-0195", reason = "transitive via inferno/lance/opendal; XML from trusted cloud endpoints, not attacker-controlled" },
 ]

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

@@ -518,6 +518,9 @@ x > 5 OR y = 'test'

 Filtering performance can often be improved by creating a scalar index
 on the filter column(s).
+
+Calling this multiple times combines the filters with a logical AND rather
+than replacing the previous filter.
 ```

 #### Inherited from
@@ -398,6 +398,26 @@ Drop an index from the table.

 ***

+### getLsmWriteSpec()
+
+```ts
+abstract getLsmWriteSpec(): Promise<undefined | LsmWriteSpec>
+```
+
+Read the [LsmWriteSpec](../interfaces/LsmWriteSpec.md) currently installed on this table.
+
+Resolves to `undefined` when the MemWAL LSM write path is not enabled (no
+spec has been set, or it was removed with [Table#unsetLsmWriteSpec](Table.md#unsetlsmwritespec)).
+The returned spec — including its `maintainedIndexes` and
+`writerConfigDefaults` — mirrors what was passed to
+[Table#setLsmWriteSpec](Table.md#setlsmwritespec).
+
+#### Returns
+
+`Promise`&lt;`undefined` \| [`LsmWriteSpec`](../interfaces/LsmWriteSpec.md)&gt;
+
+***
+
 ### indexStats()

 ```ts
@@ -767,6 +767,9 @@ x > 5 OR y = 'test'

 Filtering performance can often be improved by creating a scalar index
 on the filter column(s).
+
+Calling this multiple times combines the filters with a logical AND rather
+than replacing the previous filter.
 ```

 #### Inherited from
@@ -0,0 +1,42 @@
+[**@lancedb/lancedb**](../README.md) • **Docs**
+
+***
+
+[@lancedb/lancedb](../globals.md) / instrumentLanceDbMetrics
+
+# Function: instrumentLanceDbMetrics()
+
+```ts
+function instrumentLanceDbMetrics(meterProvider?): boolean
+```
+
+Register LanceDB metrics as OpenTelemetry observable instruments.
+
+Installs a process-global metrics recorder and creates one observable
+instrument per LanceDB metric (currently object store request counts, bytes,
+latency, errors, and throttles) on the given (or global) `MeterProvider`. The
+configured `MetricReader` then collects them on its own schedule.
+
+Counters and gauges map directly to observable counters/gauges. Because
+OpenTelemetry has no asynchronous histogram instrument, each histogram is
+exported Prometheus-style as cumulative `le` bucket counts (`<name>_bucket`,
+with an `le` attribute) plus `<name>_count` and `<name>_sum`.
+
+Requires `@opentelemetry/api` (a dependency) and, to actually export, an
+OpenTelemetry SDK such as `@opentelemetry/sdk-metrics`.
+
+## Parameters
+
+* **meterProvider?**: `MeterProvider`
+    The provider to register instruments on. Defaults to the
+    global provider from `@opentelemetry/api`.
+
+## Returns
+
+`boolean`
+
+`true` if the recorder is installed and instruments are registered.
+  `false` if a different `metrics` recorder is already installed in this
+  process (only one global recorder is permitted), in which case a warning is
+  emitted and no instruments are created. Calling this more than once is safe;
+  instruments are created only on the first successful call.
@@ -131,6 +131,7 @@
 - [RecordBatchIterator](functions/RecordBatchIterator.md)
 - [connect](functions/connect.md)
 - [connectNamespace](functions/connectNamespace.md)
+- [instrumentLanceDbMetrics](functions/instrumentLanceDbMetrics.md)
 - [makeArrowTable](functions/makeArrowTable.md)
 - [packBits](functions/packBits.md)
 - [permutationBuilder](functions/permutationBuilder.md)
@@ -8,6 +8,14 @@

 ## Properties

+### clumpSize?
+
+```ts
+optional clumpSize: number;
+```
+
+***
+
 ### counts?

 ```ts
@@ -8,7 +8,7 @@
    <parent>
      <groupId>com.lancedb</groupId>
      <artifactId>lancedb-parent</artifactId>
-      <version>0.31.0-beta.4</version>
+      <version>0.32.0-beta.0</version>
      <relativePath>../pom.xml</relativePath>
    </parent>

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

    <groupId>com.lancedb</groupId>
    <artifactId>lancedb-parent</artifactId>
-    <version>0.31.0-beta.4</version>
+    <version>0.32.0-beta.0</version>
    <packaging>pom</packaging>
    <name>${project.artifactId}</name>
    <description>LanceDB Java SDK Parent POM</description>
@@ -28,7 +28,7 @@
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <arrow.version>15.0.0</arrow.version>
-        <lance-core.version>9.0.0-beta.10</lance-core.version>
+        <lance-core.version>9.0.0-beta.19</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.31.0-beta.4"
+version = "0.32.0-beta.0"
 publish = false
 license.workspace = true
 description.workspace = true
@@ -44,6 +44,6 @@ aws-lc-rs = "=1.16.3"
 napi-build = "2.3.1"

 [features]
-default = ["remote", "lancedb/aws", "lancedb/gcs", "lancedb/azure", "lancedb/dynamodb", "lancedb/oss", "lancedb/huggingface"]
+default = ["remote", "lancedb/aws", "lancedb/gcs", "lancedb/azure", "lancedb/dynamodb", "lancedb/oss", "lancedb/huggingface", "lancedb/goosefs", "lancedb/metrics-otel"]
 fp16kernels = ["lancedb/fp16kernels"]
 remote = ["lancedb/remote"]
@@ -0,0 +1,114 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+import {
+  MeterProvider,
+  type MetricData,
+  MetricReader,
+} from "@opentelemetry/sdk-metrics";
+import * as tmp from "tmp";
+import { connect, instrumentLanceDbMetrics } from "../lancedb";
+// snapshotLancedbMetrics is internal plumbing (not part of the public API), so
+// it is imported from the native module rather than the package entry point.
+import { snapshotLancedbMetrics } from "../lancedb/native";
+
+// The metrics recorder is process-global and installed once, so the whole
+// bridge is exercised in a single test to avoid cross-test global-state coupling.
+
+// A minimal pull-based reader whose `collect()` we drive directly, invoking the
+// observable-instrument callbacks. `@opentelemetry/sdk-metrics` ships no
+// in-memory reader, so we subclass the abstract base.
+class TestMetricReader extends MetricReader {
+  protected async onForceFlush(): Promise<void> {
+    // no-op: collection is driven directly via collect()
+  }
+  protected async onShutdown(): Promise<void> {
+    // no-op: nothing to release
+  }
+}
+
+async function metricsByName(
+  reader: TestMetricReader,
+): Promise<Map<string, MetricData>> {
+  const collected = await reader.collect();
+  const result = new Map<string, MetricData>();
+  for (const scope of collected.resourceMetrics.scopeMetrics) {
+    for (const metric of scope.metrics) {
+      result.set(metric.descriptor.name, metric);
+    }
+  }
+  return result;
+}
+
+describe("OpenTelemetry metrics bridge", () => {
+  let tmpDir: tmp.DirResult;
+  beforeEach(() => {
+    tmpDir = tmp.dirSync({ unsafeCleanup: true });
+  });
+  afterEach(() => tmpDir.removeCallback());
+
+  it("snapshot is safe to call regardless of install state", () => {
+    expect(Array.isArray(snapshotLancedbMetrics())).toBe(true);
+  });
+
+  it("exports object store metrics via observable instruments", async () => {
+    const reader = new TestMetricReader();
+    const provider = new MeterProvider({ readers: [reader] });
+    expect(instrumentLanceDbMetrics(provider)).toBe(true);
+
+    // Generate object store activity on the local filesystem (scheme "file").
+    const db = await connect(tmpDir.name);
+    const data = Array.from({ length: 256 }, (_, i) => ({ id: i }));
+    const table = await db.createTable("t", data);
+    expect(await table.countRows()).toBe(256);
+
+    const metrics = await metricsByName(reader);
+
+    const requests = metrics.get("lance_object_store_requests_total");
+    expect(requests).toBeDefined();
+    // biome-ignore lint/suspicious/noExplicitAny: SDK point shape
+    const requestPoints = (requests!.dataPoints as any[]) ?? [];
+    expect(requestPoints.length).toBeGreaterThan(0);
+    for (const p of requestPoints) {
+      // Labelled by `operation` and `base` (the store scheme by default).
+      expect(p.attributes).toHaveProperty("base");
+      expect(p.attributes).toHaveProperty("operation");
+    }
+    const totalRequests = requestPoints.reduce((acc, p) => acc + p.value, 0);
+    expect(totalRequests).toBeGreaterThan(0);
+
+    // Histograms are decomposed into bucket / count / sum observable counters.
+    const bucket = metrics.get(
+      "lance_object_store_request_duration_seconds_bucket",
+    );
+    expect(bucket).toBeDefined();
+    // biome-ignore lint/suspicious/noExplicitAny: SDK point shape
+    const bucketPoints = (bucket!.dataPoints as any[]) ?? [];
+    expect(bucketPoints.length).toBeGreaterThan(0);
+    expect(bucketPoints.every((p) => "le" in p.attributes)).toBe(true);
+    // The implicit +Inf bucket must be present.
+    expect(bucketPoints.some((p) => p.attributes.le === "+Inf")).toBe(true);
+
+    const count = metrics.get(
+      "lance_object_store_request_duration_seconds_count",
+    );
+    expect(count).toBeDefined();
+    // biome-ignore lint/suspicious/noExplicitAny: SDK point shape
+    const countPoints = (count!.dataPoints as any[]) ?? [];
+    expect(countPoints.reduce((acc, p) => acc + p.value, 0)).toBeGreaterThan(0);
+
+    const sum = metrics.get("lance_object_store_request_duration_seconds_sum");
+    expect(sum).toBeDefined();
+    // biome-ignore lint/suspicious/noExplicitAny: SDK point shape
+    const sumPoints = (sum!.dataPoints as any[]) ?? [];
+    expect(sumPoints.reduce((acc, p) => acc + p.value, 0)).toBeGreaterThan(0);
+
+    // Unit handling: only `_sum` keeps the histogram's unit (seconds); `_bucket`
+    // and `_count` observe cumulative counts and are unitless.
+    expect(sum!.descriptor.unit).toBe("s");
+    expect(bucket!.descriptor.unit).toBe("");
+    expect(count!.descriptor.unit).toBe("");
+
+    await provider.shutdown();
+  });
+});
@@ -215,6 +215,20 @@ describe("Query orderBy", () => {
    expect(results[2].score).toBeCloseTo(4.1, 0.001);
  });

+  it("should combine repeated where clauses with AND", async () => {
+    const results = await table
+      .query()
+      .where("score > 1.0")
+      .where("score < 3.0")
+      .orderBy({ columnName: "score" })
+      .toArray();
+    // Only rows matching both predicates should be returned, rather than the
+    // second where() silently replacing the first.
+    expect(results.length).toBe(2);
+    expect(results[0].score).toBeCloseTo(1.2, 0.001);
+    expect(results[1].score).toBeCloseTo(2.8, 0.001);
+  });
+
  it("should support method chaining with limit", async () => {
    const results = await table
      .query()
@@ -2992,6 +2992,56 @@ describe("setLsmWriteSpec / unsetLsmWriteSpec", () => {
      }),
    ).rejects.toThrow();
  });
+
+  it("reads back the installed spec via getLsmWriteSpec", async () => {
+    const conn = await connect(tmpDir.name);
+    const table = await makeTable(conn);
+    await table.setUnenforcedPrimaryKey("id");
+
+    // Nothing installed yet.
+    expect(await table.getLsmWriteSpec()).toBeUndefined();
+
+    // A real scalar index is needed to name it as a maintained index.
+    await table.add([{ id: 1 }, { id: 2 }, { id: 3 }]);
+    await table.createIndex("id");
+    const indexName = (await table.listIndices())[0].name;
+
+    // Bucket spec round-trips, including maintained indexes and writer config
+    // defaults. Lance writer-config keys are canonically snake_case.
+    // biome-ignore lint/style/useNamingConvention: Lance writer-config keys are snake_case
+    const writerConfigDefaults = { durable_write: "false" };
+    await table.setLsmWriteSpec({
+      specType: "bucket",
+      column: "id",
+      numBuckets: 4,
+      maintainedIndexes: [indexName],
+      writerConfigDefaults,
+    });
+    const spec = await table.getLsmWriteSpec();
+    expect(spec).toBeDefined();
+    expect(spec?.specType).toBe("bucket");
+    expect(spec?.column).toBe("id");
+    expect(spec?.numBuckets).toBe(4);
+    expect(spec?.maintainedIndexes).toEqual([indexName]);
+    expect(spec?.writerConfigDefaults).toEqual(writerConfigDefaults);
+
+    // After unset, undefined again.
+    await table.unsetLsmWriteSpec();
+    expect(await table.getLsmWriteSpec()).toBeUndefined();
+
+    // Identity round-trips (column recovered from the schema).
+    await table.setLsmWriteSpec({ specType: "identity", column: "id" });
+    const identity = await table.getLsmWriteSpec();
+    expect(identity?.specType).toBe("identity");
+    expect(identity?.column).toBe("id");
+    await table.unsetLsmWriteSpec();
+
+    // Unsharded round-trips (no routing column).
+    await table.setLsmWriteSpec({ specType: "unsharded" });
+    const unsharded = await table.getLsmWriteSpec();
+    expect(unsharded?.specType).toBe("unsharded");
+    expect(unsharded?.column).toBeFalsy();
+  });
 });

 describe("LSM merge insert", () => {
@@ -20,6 +20,11 @@ import { HeaderProvider } from "./header";
 // Re-export native header provider for use with connectWithHeaderProvider
 export { JsHeaderProvider as NativeJsHeaderProvider } from "./native.js";

+// OpenTelemetry metrics bridge. Only the high-level entry point is public; the
+// underlying recorder/catalog/snapshot functions remain internal plumbing that
+// `otel.ts` consumes from the native module.
+export { instrumentLanceDbMetrics } from "./otel";
+
 export {
  AddColumnsSql,
  ConnectionOptions,
@@ -0,0 +1,137 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+import {
+  type Attributes,
+  type MeterProvider,
+  type ObservableResult,
+  metrics,
+} from "@opentelemetry/api";
+
+import {
+  lancedbMetricsCatalog,
+  registerLancedbMetricsRecorder,
+  snapshotLancedbMetrics,
+} from "./native";
+
+let instrumented = false;
+
+/**
+ * Register LanceDB metrics as OpenTelemetry observable instruments.
+ *
+ * Installs a process-global metrics recorder and creates one observable
+ * instrument per LanceDB metric (currently object store request counts, bytes,
+ * latency, errors, and throttles) on the given (or global) `MeterProvider`. The
+ * configured `MetricReader` then collects them on its own schedule.
+ *
+ * Counters and gauges map directly to observable counters/gauges. Because
+ * OpenTelemetry has no asynchronous histogram instrument, each histogram is
+ * exported Prometheus-style as cumulative `le` bucket counts (`<name>_bucket`,
+ * with an `le` attribute) plus `<name>_count` and `<name>_sum`.
+ *
+ * Requires `@opentelemetry/api` (a dependency) and, to actually export, an
+ * OpenTelemetry SDK such as `@opentelemetry/sdk-metrics`.
+ *
+ * @param meterProvider The provider to register instruments on. Defaults to the
+ *   global provider from `@opentelemetry/api`.
+ * @returns `true` if the recorder is installed and instruments are registered.
+ *   `false` if a different `metrics` recorder is already installed in this
+ *   process (only one global recorder is permitted), in which case a warning is
+ *   emitted and no instruments are created. Calling this more than once is safe;
+ *   instruments are created only on the first successful call.
+ */
+export function instrumentLanceDbMetrics(
+  meterProvider?: MeterProvider,
+): boolean {
+  if (!registerLancedbMetricsRecorder()) {
+    console.warn(
+      "Could not install the LanceDB metrics recorder: another `metrics` " +
+        "recorder is already installed in this process. LanceDB metrics will " +
+        "not be exported via OpenTelemetry.",
+    );
+    return false;
+  }
+
+  if (instrumented) {
+    return true;
+  }
+
+  const provider = meterProvider ?? metrics.getMeterProvider();
+  const meter = provider.getMeter("lancedb");
+
+  const scalarCallback = (metricName: string) => (result: ObservableResult) => {
+    for (const point of snapshotLancedbMetrics()) {
+      if (point.name === metricName && point.value != null) {
+        result.observe(point.value, point.attributes);
+      }
+    }
+  };
+
+  const bucketCallback = (metricName: string) => (result: ObservableResult) => {
+    for (const point of snapshotLancedbMetrics()) {
+      if (point.name !== metricName || point.buckets == null) {
+        continue;
+      }
+      for (const bucket of point.buckets) {
+        const attributes: Attributes = {
+          ...point.attributes,
+          le: bucket.le,
+        };
+        result.observe(bucket.cumulativeCount, attributes);
+      }
+    }
+  };
+
+  const fieldCallback =
+    (metricName: string, field: "count" | "sum") =>
+    (result: ObservableResult) => {
+      for (const point of snapshotLancedbMetrics()) {
+        if (point.name !== metricName) {
+          continue;
+        }
+        const value = point[field];
+        if (value != null) {
+          result.observe(value, point.attributes);
+        }
+      }
+    };
+
+  for (const desc of lancedbMetricsCatalog()) {
+    const unit = desc.unit ?? "";
+    if (desc.kind === "counter") {
+      const counter = meter.createObservableCounter(desc.name, {
+        unit,
+        description: desc.description,
+      });
+      counter.addCallback(scalarCallback(desc.name));
+    } else if (desc.kind === "gauge") {
+      const gauge = meter.createObservableGauge(desc.name, {
+        unit,
+        description: desc.description,
+      });
+      gauge.addCallback(scalarCallback(desc.name));
+    } else if (desc.kind === "histogram") {
+      // `_bucket` and `_count` observe cumulative sample counts, not the
+      // histogram's measured quantity, so they are unitless; only `_sum`
+      // carries the histogram's unit.
+      const bucket = meter.createObservableCounter(`${desc.name}_bucket`, {
+        description: `${desc.description} (cumulative buckets)`,
+      });
+      bucket.addCallback(bucketCallback(desc.name));
+
+      const count = meter.createObservableCounter(`${desc.name}_count`, {
+        description: `${desc.description} (count)`,
+      });
+      count.addCallback(fieldCallback(desc.name, "count"));
+
+      const sum = meter.createObservableCounter(`${desc.name}_sum`, {
+        unit,
+        description: `${desc.description} (sum)`,
+      });
+      sum.addCallback(fieldCallback(desc.name, "sum"));
+    }
+  }
+
+  instrumented = true;
+  return true;
+}
@@ -362,6 +362,9 @@ export class StandardQueryBase<
   *
   * Filtering performance can often be improved by creating a scalar index
   * on the filter column(s).
+   *
+   * Calling this multiple times combines the filters with a logical AND rather
+   * than replacing the previous filter.
   */
  where(predicate: string): this {
    this.doCall((inner: NativeQueryType) => inner.onlyIf(predicate));
@@ -585,6 +585,17 @@ export abstract class Table {
   * @returns {Promise<void>}
   */
  abstract unsetLsmWriteSpec(): Promise<void>;
+  /**
+   * Read the {@link LsmWriteSpec} currently installed on this table.
+   *
+   * Resolves to `undefined` when the MemWAL LSM write path is not enabled (no
+   * spec has been set, or it was removed with {@link Table#unsetLsmWriteSpec}).
+   * The returned spec — including its `maintainedIndexes` and
+   * `writerConfigDefaults` — mirrors what was passed to
+   * {@link Table#setLsmWriteSpec}.
+   * @returns {Promise<LsmWriteSpec | undefined>}
+   */
+  abstract getLsmWriteSpec(): Promise<LsmWriteSpec | undefined>;
  /**
   * Drain and close any cached MemWAL shard writers held for this table.
   *
@@ -1091,6 +1102,15 @@ export class LocalTable extends Table {
    return await this.inner.unsetLsmWriteSpec();
  }

+  async getLsmWriteSpec(): Promise<LsmWriteSpec | undefined> {
+    // The native binding types `specType` as a plain `string`; narrow it back
+    // to the public union. The Rust `From` impl only ever emits one of the
+    // three valid values, so the cast is safe.
+    return ((await this.inner.getLsmWriteSpec()) ?? undefined) as
+      | LsmWriteSpec
+      | undefined;
+  }
+
  async closeLsmWriters(): Promise<void> {
    return await this.inner.closeLsmWriters();
  }
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-darwin-arm64",
-	"version": "0.31.0-beta.4",
+	"version": "0.32.0-beta.0",
 	"os": ["darwin"],
 	"cpu": ["arm64"],
 	"main": "lancedb.darwin-arm64.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-gnu",
-	"version": "0.31.0-beta.4",
+	"version": "0.32.0-beta.0",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-gnu.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-musl",
-	"version": "0.31.0-beta.4",
+	"version": "0.32.0-beta.0",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-musl.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-gnu",
-	"version": "0.31.0-beta.4",
+	"version": "0.32.0-beta.0",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-gnu.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-musl",
-	"version": "0.31.0-beta.4",
+	"version": "0.32.0-beta.0",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-musl.node",
@@ -1,6 +1,6 @@
 {
  "name": "@lancedb/lancedb-win32-arm64-msvc",
-  "version": "0.31.0-beta.4",
+  "version": "0.32.0-beta.0",
  "os": [
    "win32"
  ],
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-win32-x64-msvc",
-	"version": "0.31.0-beta.4",
+	"version": "0.32.0-beta.0",
 	"os": ["win32"],
 	"cpu": ["x64"],
 	"main": "lancedb.win32-x64-msvc.node",
@@ -1,12 +1,12 @@
 {
  "name": "@lancedb/lancedb",
-  "version": "0.31.0-beta.4",
+  "version": "0.32.0-beta.0",
  "lockfileVersion": 3,
  "requires": true,
  "packages": {
    "": {
      "name": "@lancedb/lancedb",
-      "version": "0.31.0-beta.4",
+      "version": "0.32.0-beta.0",
      "cpu": [
        "x64",
        "arm64"
@@ -18,6 +18,7 @@
        "win32"
      ],
      "dependencies": {
+        "@opentelemetry/api": "^1.9.0",
        "reflect-metadata": "^0.2.2"
      },
      "devDependencies": {
@@ -27,6 +28,7 @@
        "@biomejs/biome": "^1.7.3",
        "@jest/globals": "^29.7.0",
        "@napi-rs/cli": "3.7.0",
+        "@opentelemetry/sdk-metrics": "^1.30.0",
        "@types/axios": "^0.14.0",
        "@types/jest": "^29.1.2",
        "@types/node": "22.7.4",
@@ -4148,6 +4150,75 @@
        "@octokit/openapi-types": "^27.0.0"
      }
    },
+    "node_modules/@opentelemetry/api": {
+      "version": "1.9.1",
+      "resolved": "https://registry.npmjs.org/@opentelemetry/api/-/api-1.9.1.tgz",
+      "integrity": "sha512-gLyJlPHPZYdAk1JENA9LeHejZe1Ti77/pTeFm/nMXmQH/HFZlcS/O2XJB+L8fkbrNSqhdtlvjBVjxwUYanNH5Q==",
+      "license": "Apache-2.0",
+      "engines": {
+        "node": ">=8.0.0"
+      }
+    },
+    "node_modules/@opentelemetry/core": {
+      "version": "1.30.1",
+      "resolved": "https://registry.npmjs.org/@opentelemetry/core/-/core-1.30.1.tgz",
+      "integrity": "sha512-OOCM2C/QIURhJMuKaekP3TRBxBKxG/TWWA0TL2J6nXUtDnuCtccy49LUJF8xPFXMX+0LMcxFpCo8M9cGY1W6rQ==",
+      "dev": true,
+      "license": "Apache-2.0",
+      "dependencies": {
+        "@opentelemetry/semantic-conventions": "1.28.0"
+      },
+      "engines": {
+        "node": ">=14"
+      },
+      "peerDependencies": {
+        "@opentelemetry/api": ">=1.0.0 <1.10.0"
+      }
+    },
+    "node_modules/@opentelemetry/resources": {
+      "version": "1.30.1",
+      "resolved": "https://registry.npmjs.org/@opentelemetry/resources/-/resources-1.30.1.tgz",
+      "integrity": "sha512-5UxZqiAgLYGFjS4s9qm5mBVo433u+dSPUFWVWXmLAD4wB65oMCoXaJP1KJa9DIYYMeHu3z4BZcStG3LC593cWA==",
+      "dev": true,
+      "license": "Apache-2.0",
+      "dependencies": {
+        "@opentelemetry/core": "1.30.1",
+        "@opentelemetry/semantic-conventions": "1.28.0"
+      },
+      "engines": {
+        "node": ">=14"
+      },
+      "peerDependencies": {
+        "@opentelemetry/api": ">=1.0.0 <1.10.0"
+      }
+    },
+    "node_modules/@opentelemetry/sdk-metrics": {
+      "version": "1.30.1",
+      "resolved": "https://registry.npmjs.org/@opentelemetry/sdk-metrics/-/sdk-metrics-1.30.1.tgz",
+      "integrity": "sha512-q9zcZ0Okl8jRgmy7eNW3Ku1XSgg3sDLa5evHZpCwjspw7E8Is4K/haRPDJrBcX3YSn/Y7gUvFnByNYEKQNbNog==",
+      "dev": true,
+      "license": "Apache-2.0",
+      "dependencies": {
+        "@opentelemetry/core": "1.30.1",
+        "@opentelemetry/resources": "1.30.1"
+      },
+      "engines": {
+        "node": ">=14"
+      },
+      "peerDependencies": {
+        "@opentelemetry/api": ">=1.3.0 <1.10.0"
+      }
+    },
+    "node_modules/@opentelemetry/semantic-conventions": {
+      "version": "1.28.0",
+      "resolved": "https://registry.npmjs.org/@opentelemetry/semantic-conventions/-/semantic-conventions-1.28.0.tgz",
+      "integrity": "sha512-lp4qAiMTD4sNWW4DbKLBkfiMZ4jbAboJIGOQr5DvciMRI494OapieI9qiODpOt0XBr1LjIDy1xAGAnVs5supTA==",
+      "dev": true,
+      "license": "Apache-2.0",
+      "engines": {
+        "node": ">=14"
+      }
+    },
    "node_modules/@protobufjs/aspromise": {
      "version": "1.1.2",
      "resolved": "https://registry.npmjs.org/@protobufjs/aspromise/-/aspromise-1.1.2.tgz",
@@ -11,7 +11,7 @@
    "ann"
  ],
  "private": false,
-  "version": "0.31.0-beta.4",
+  "version": "0.32.0-beta.0",
  "main": "dist/index.js",
  "exports": {
    ".": "./dist/index.js",
@@ -44,6 +44,7 @@
    "@biomejs/biome": "^1.7.3",
    "@jest/globals": "^29.7.0",
    "@napi-rs/cli": "3.7.0",
+    "@opentelemetry/sdk-metrics": "^1.30.0",
    "@types/axios": "^0.14.0",
    "@types/jest": "^29.1.2",
    "@types/node": "22.7.4",
@@ -92,6 +93,7 @@
    "version": "napi version"
  },
  "dependencies": {
+    "@opentelemetry/api": "^1.9.0",
    "reflect-metadata": "^0.2.2"
  },
  "optionalDependencies": {
@@ -8,6 +8,9 @@ importers:

  .:
    dependencies:
+      '@opentelemetry/api':
+        specifier: ^1.9.0
+        version: 1.9.1
      apache-arrow:
        specifier: '>=15.0.0 <=18.1.0'
        version: 18.1.0
@@ -33,6 +36,9 @@ importers:
      '@napi-rs/cli':
        specifier: 3.7.0
        version: 3.7.0(@emnapi/core@1.10.0)(@emnapi/runtime@1.10.0)(@types/node@22.7.4)
+      '@opentelemetry/sdk-metrics':
+        specifier: ^1.30.0
+        version: 1.30.1(@opentelemetry/api@1.9.1)
      '@types/axios':
        specifier: ^0.14.0
        version: 0.14.4
@@ -1307,6 +1313,32 @@ packages:
  '@octokit/types@16.0.0':
    resolution: {integrity: sha512-sKq+9r1Mm4efXW1FCk7hFSeJo4QKreL/tTbR0rz/qx/r1Oa2VV83LTA/H/MuCOX7uCIJmQVRKBcbmWoySjAnSg==}

+  '@opentelemetry/api@1.9.1':
+    resolution: {integrity: sha512-gLyJlPHPZYdAk1JENA9LeHejZe1Ti77/pTeFm/nMXmQH/HFZlcS/O2XJB+L8fkbrNSqhdtlvjBVjxwUYanNH5Q==}
+    engines: {node: '>=8.0.0'}
+
+  '@opentelemetry/core@1.30.1':
+    resolution: {integrity: sha512-OOCM2C/QIURhJMuKaekP3TRBxBKxG/TWWA0TL2J6nXUtDnuCtccy49LUJF8xPFXMX+0LMcxFpCo8M9cGY1W6rQ==}
+    engines: {node: '>=14'}
+    peerDependencies:
+      '@opentelemetry/api': '>=1.0.0 <1.10.0'
+
+  '@opentelemetry/resources@1.30.1':
+    resolution: {integrity: sha512-5UxZqiAgLYGFjS4s9qm5mBVo433u+dSPUFWVWXmLAD4wB65oMCoXaJP1KJa9DIYYMeHu3z4BZcStG3LC593cWA==}
+    engines: {node: '>=14'}
+    peerDependencies:
+      '@opentelemetry/api': '>=1.0.0 <1.10.0'
+
+  '@opentelemetry/sdk-metrics@1.30.1':
+    resolution: {integrity: sha512-q9zcZ0Okl8jRgmy7eNW3Ku1XSgg3sDLa5evHZpCwjspw7E8Is4K/haRPDJrBcX3YSn/Y7gUvFnByNYEKQNbNog==}
+    engines: {node: '>=14'}
+    peerDependencies:
+      '@opentelemetry/api': '>=1.3.0 <1.10.0'
+
+  '@opentelemetry/semantic-conventions@1.28.0':
+    resolution: {integrity: sha512-lp4qAiMTD4sNWW4DbKLBkfiMZ4jbAboJIGOQr5DvciMRI494OapieI9qiODpOt0XBr1LjIDy1xAGAnVs5supTA==}
+    engines: {node: '>=14'}
+
  '@protobufjs/aspromise@1.1.2':
    resolution: {integrity: sha512-j+gKExEuLmKwvz3OgROXtrJ2UG2x8Ch2YZUxahh+s1F2HZ+wAceUNLkvy6zKCPVRkU++ZWQrdxsUeQXmcg4uoQ==}

@@ -4925,6 +4957,27 @@ snapshots:
    dependencies:
      '@octokit/openapi-types': 27.0.0

+  '@opentelemetry/api@1.9.1': {}
+
+  '@opentelemetry/core@1.30.1(@opentelemetry/api@1.9.1)':
+    dependencies:
+      '@opentelemetry/api': 1.9.1
+      '@opentelemetry/semantic-conventions': 1.28.0
+
+  '@opentelemetry/resources@1.30.1(@opentelemetry/api@1.9.1)':
+    dependencies:
+      '@opentelemetry/api': 1.9.1
+      '@opentelemetry/core': 1.30.1(@opentelemetry/api@1.9.1)
+      '@opentelemetry/semantic-conventions': 1.28.0
+
+  '@opentelemetry/sdk-metrics@1.30.1(@opentelemetry/api@1.9.1)':
+    dependencies:
+      '@opentelemetry/api': 1.9.1
+      '@opentelemetry/core': 1.30.1(@opentelemetry/api@1.9.1)
+      '@opentelemetry/resources': 1.30.1(@opentelemetry/api@1.9.1)
+
+  '@opentelemetry/semantic-conventions@1.28.0': {}
+
  '@protobufjs/aspromise@1.1.2':
    optional: true

@@ -12,6 +12,7 @@ mod header;
 mod index;
 mod iterator;
 pub mod merge;
+pub mod otel;
 pub mod permutation;
 mod query;
 pub mod remote;
@@ -3,7 +3,7 @@

 use std::time::Duration;

-use lancedb::{arrow::IntoArrow, ipc::ipc_file_to_batches, table::merge::MergeInsertBuilder};
+use lancedb::{ipc::ipc_file_to_batches, table::merge::MergeInsertBuilder};
 use napi::bindgen_prelude::*;
 use napi_derive::napi;

@@ -66,11 +66,9 @@ impl NativeMergeInsertBuilder {

    #[napi(catch_unwind)]
    pub async fn execute(&self, buf: Buffer) -> napi::Result<MergeResult> {
-        let data = ipc_file_to_batches(buf.to_vec())
-            .and_then(IntoArrow::into_arrow)
-            .map_err(|e| {
-                napi::Error::from_reason(format!("Failed to read IPC file: {}", convert_error(&e)))
-            })?;
+        let data = ipc_file_to_batches(buf.to_vec()).map_err(|e| {
+            napi::Error::from_reason(format!("Failed to read IPC file: {}", convert_error(&e)))
+        })?;

        let this = self.clone();

@@ -0,0 +1,119 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+//! Node.js bindings over [`lancedb::metrics_otel`].
+//!
+//! The aggregation, catalog, and histogram bucketing all live in the LanceDB
+//! core crate; this module only converts the core snapshot types into napi
+//! objects and exposes the three entry points to JavaScript, where
+//! `lancedb/otel.ts` bridges them into the user's OpenTelemetry `MeterProvider`.
+
+use std::collections::HashMap;
+
+use lancedb::metrics_otel::{MetricPoint as CoreMetricPoint, MetricValue};
+use napi_derive::napi;
+
+/// One cumulative histogram bucket: all samples with value `<= le`.
+#[napi(object)]
+pub struct MetricBucket {
+    /// The inclusive upper bound of the bucket, or `"+Inf"` for the final bucket.
+    pub le: String,
+    /// Cumulative number of samples less than or equal to `le`.
+    pub cumulative_count: f64,
+}
+
+/// One aggregated metric data point. For counters and gauges only `value` is
+/// set; for histograms `buckets` (cumulative `le` counts), `count`, and `sum`
+/// are set.
+#[napi(object)]
+pub struct MetricPoint {
+    pub name: String,
+    pub kind: String,
+    pub attributes: HashMap<String, String>,
+    pub value: Option<f64>,
+    pub buckets: Option<Vec<MetricBucket>>,
+    pub count: Option<f64>,
+    pub sum: Option<f64>,
+}
+
+impl From<CoreMetricPoint> for MetricPoint {
+    fn from(point: CoreMetricPoint) -> Self {
+        let kind = point.kind.as_str().to_string();
+        let (value, buckets, count, sum) = match point.value {
+            MetricValue::Scalar(v) => (Some(v), None, None, None),
+            MetricValue::Histogram {
+                buckets,
+                count,
+                sum,
+            } => (
+                None,
+                Some(
+                    buckets
+                        .into_iter()
+                        // Counts stay well within the f64-exact integer range
+                        // (2^53), so this cast is lossless in practice and keeps
+                        // the values plain JS numbers for OpenTelemetry.
+                        .map(|(le, cumulative_count)| MetricBucket {
+                            le,
+                            cumulative_count: cumulative_count as f64,
+                        })
+                        .collect(),
+                ),
+                Some(count as f64),
+                Some(sum),
+            ),
+        };
+        Self {
+            name: point.name,
+            kind,
+            attributes: point.attributes,
+            value,
+            buckets,
+            count,
+            sum,
+        }
+    }
+}
+
+/// A described metric, used by the JavaScript layer to create instruments up front.
+#[napi(object)]
+pub struct MetricDescription {
+    pub name: String,
+    pub kind: String,
+    pub unit: Option<String>,
+    pub description: String,
+}
+
+/// Install the LanceDB metrics recorder as the process-global `metrics` recorder.
+///
+/// Returns `true` if the recorder is installed (now or previously). Returns
+/// `false` if a *different* recorder is already installed — `metrics` allows
+/// only one global recorder per process, so LanceDB cannot coexist with another.
+#[napi]
+pub fn register_lancedb_metrics_recorder() -> bool {
+    lancedb::metrics_otel::register_metrics_recorder()
+}
+
+/// The catalog of described LanceDB metrics. Empty until the recorder is installed.
+#[napi]
+pub fn lancedb_metrics_catalog() -> Vec<MetricDescription> {
+    lancedb::metrics_otel::metrics_catalog()
+        .into_iter()
+        .map(|desc| MetricDescription {
+            name: desc.name,
+            kind: desc.kind.as_str().to_string(),
+            unit: desc.unit,
+            description: desc.description,
+        })
+        .collect()
+}
+
+/// A point-in-time snapshot of every recorded metric. Empty until the recorder
+/// is installed.
+#[napi]
+pub fn snapshot_lancedb_metrics() -> Vec<MetricPoint> {
+    lancedb::metrics_otel::snapshot_metrics()
+        .into_iter()
+        .map(MetricPoint::from)
+        .collect()
+}
@@ -16,6 +16,7 @@ pub struct SplitRandomOptions {
    pub counts: Option<Vec<i64>>,
    pub fixed: Option<i64>,
    pub seed: Option<i64>,
+    pub clump_size: Option<i64>,
    pub split_names: Option<Vec<String>>,
 }

@@ -125,10 +126,15 @@ impl PermutationBuilder {
        };

        let seed = options.seed.map(|s| s as u64);
+        let clump_size = options.clump_size.map(|c| c as u64);

        self.modify(|builder| {
            builder.with_split_strategy(
-                SplitStrategy::Random { seed, sizes },
+                SplitStrategy::Random {
+                    seed,
+                    sizes,
+                    clump_size,
+                },
                options.split_names.clone(),
            )
        })
@@ -411,6 +411,16 @@ impl Table {
            .default_error()
    }

+    #[napi(catch_unwind)]
+    pub async fn get_lsm_write_spec(&self) -> napi::Result<Option<LsmWriteSpec>> {
+        let spec = self
+            .inner_ref()?
+            .get_lsm_write_spec()
+            .await
+            .default_error()?;
+        Ok(spec.map(LsmWriteSpec::from))
+    }
+
    #[napi(catch_unwind)]
    pub async fn close_lsm_writers(&self) -> napi::Result<()> {
        self.inner_ref()?.close_lsm_writers().await.default_error()
@@ -728,6 +738,47 @@ impl TryFrom<LsmWriteSpec> for lancedb::table::LsmWriteSpec {
    }
 }

+impl From<lancedb::table::LsmWriteSpec> for LsmWriteSpec {
+    fn from(spec: lancedb::table::LsmWriteSpec) -> Self {
+        use lancedb::table::LsmWriteSpec as Native;
+        match spec {
+            Native::Bucket {
+                column,
+                num_buckets,
+                maintained_indexes,
+                writer_config_defaults,
+            } => Self {
+                spec_type: "bucket".to_string(),
+                column: Some(column),
+                num_buckets: Some(num_buckets),
+                maintained_indexes: Some(maintained_indexes),
+                writer_config_defaults: Some(writer_config_defaults),
+            },
+            Native::Identity {
+                column,
+                maintained_indexes,
+                writer_config_defaults,
+            } => Self {
+                spec_type: "identity".to_string(),
+                column: Some(column),
+                num_buckets: None,
+                maintained_indexes: Some(maintained_indexes),
+                writer_config_defaults: Some(writer_config_defaults),
+            },
+            Native::Unsharded {
+                maintained_indexes,
+                writer_config_defaults,
+            } => Self {
+                spec_type: "unsharded".to_string(),
+                column: None,
+                num_buckets: None,
+                maintained_indexes: Some(maintained_indexes),
+                writer_config_defaults: Some(writer_config_defaults),
+            },
+        }
+    }
+}
+
 /// Statistics about a compaction operation.
 #[napi(object)]
 #[derive(Clone, Debug)]
@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "0.34.0-beta.4"
+current_version = "0.35.0-beta.1"
 parse = """(?x)
    (?P<major>0|[1-9]\\d*)\\.
    (?P<minor>0|[1-9]\\d*)\\.
@@ -1,6 +1,6 @@
 [package]
 name = "lancedb-python"
-version = "0.34.0-beta.4"
+version = "0.35.0-beta.1"
 publish = false
 edition.workspace = true
 description = "Python bindings for LanceDB"
@@ -47,6 +47,6 @@ pyo3-build-config = { version = "0.28", features = [
 ] }

 [features]
-default = ["remote",  "lancedb/aws", "lancedb/gcs", "lancedb/azure", "lancedb/dynamodb", "lancedb/oss", "lancedb/huggingface"]
+default = ["remote", "lancedb/aws", "lancedb/gcs", "lancedb/azure", "lancedb/dynamodb", "lancedb/oss", "lancedb/huggingface", "lancedb/cos", "lancedb/goosefs", "lancedb/metrics-otel"]
 fp16kernels = ["lancedb/fp16kernels"]
 remote = ["lancedb/remote"]
@@ -0,0 +1,135 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+"""Benchmark for StreamingDataset throughput.
+
+Sweeps read_batch_size from 1 to 16384 to show how amortising the per-request
+overhead scales.  Each row at each chunk size is timed via the real
+StreamingDataset so the numbers reflect production code.
+
+Run with:
+    cd python
+    uv run --extra tests benchmarks/bench_streaming_dataloader.py
+
+Optional env vars:
+    BENCH_NUM_ROWS   — total rows in the table (default 49152 = 24 × 2048)
+    BENCH_NUM_SPLITS — number of splits (default 24)
+    BENCH_STEPS      — round-robin cycles to time per chunk size (default 100)
+    BENCH_ROW_BYTES  — approximate bytes per row padded with a binary column
+                       (default 4096, mimics a small embedding/image patch)
+"""
+
+import os
+import time
+import tempfile
+
+import pyarrow as pa
+import lancedb
+
+from lancedb.streaming import StreamingDataset
+
+# ---------------------------------------------------------------------------
+# Configuration
+# ---------------------------------------------------------------------------
+
+NUM_SPLITS = int(os.environ.get("BENCH_NUM_SPLITS", 24))
+# Default: 2048 rows per split so every chunk size up to 16Ki has ≥1 full
+# chunk (except 16Ki itself which gets a single full-split fetch — still valid).
+NUM_ROWS = int(os.environ.get("BENCH_NUM_ROWS", NUM_SPLITS * 2048))
+STEPS = int(os.environ.get("BENCH_STEPS", 100))
+ROW_BYTES = int(os.environ.get("BENCH_ROW_BYTES", 4096))
+
+assert NUM_ROWS % NUM_SPLITS == 0, "NUM_ROWS must be divisible by NUM_SPLITS"
+
+CHUNK_SIZES = [1, 4, 16, 64, 256, 1024, 4096, 16384]
+
+
+# ---------------------------------------------------------------------------
+# Table helpers
+# ---------------------------------------------------------------------------
+
+
+def make_table(db_path: str) -> lancedb.table.Table:
+    db = lancedb.connect(db_path)
+    payload = b"x" * ROW_BYTES
+    data = pa.table(
+        {
+            "id": pa.array(range(NUM_ROWS), type=pa.int32()),
+            "payload": pa.array([payload] * NUM_ROWS, type=pa.large_binary()),
+        }
+    )
+    return db.create_table("bench", data, mode="overwrite")
+
+
+# ---------------------------------------------------------------------------
+# Timing
+# ---------------------------------------------------------------------------
+
+
+def bench_chunk(table, chunk_size: int, steps: int) -> tuple[int, float]:
+    """Return (rows_drained, elapsed_seconds) for one timed run."""
+    total_rows = steps * NUM_SPLITS
+    ds = StreamingDataset(
+        table, num_splits=NUM_SPLITS, shuffle_seed=42, read_batch_size=chunk_size
+    )
+    count = 0
+    t0 = time.perf_counter()
+    for _ in ds:
+        count += 1
+        if count >= total_rows:
+            break
+    return count, time.perf_counter() - t0
+
+
+# ---------------------------------------------------------------------------
+# Main
+# ---------------------------------------------------------------------------
+
+
+def main() -> None:
+    rows_per_split = NUM_ROWS // NUM_SPLITS
+    print("Benchmark config:")
+    print(
+        f"  NUM_ROWS={NUM_ROWS}  NUM_SPLITS={NUM_SPLITS}  "
+        f"rows/split={rows_per_split}  STEPS={STEPS}  ROW_BYTES={ROW_BYTES}"
+    )
+    print(f"  ~{NUM_ROWS * ROW_BYTES / 1024 / 1024:.1f} MB total table size")
+    print()
+
+    with tempfile.TemporaryDirectory() as tmp:
+        print("Creating table...", flush=True)
+        table = make_table(tmp)
+
+        cols = (
+            f"{'chunk':>6}  {'rows':>6}  {'elapsed':>8}  {'rows/s':>10}  {'ms/step':>9}"
+        )
+        print(f"\n{cols}")
+        print("-" * 52)
+
+        for chunk in CHUNK_SIZES:
+            # Warm-up pass (one step's worth of rows)
+            warmup_ds = StreamingDataset(
+                table, num_splits=NUM_SPLITS, shuffle_seed=42, read_batch_size=chunk
+            )
+            warmup_count = 0
+            for _ in warmup_ds:
+                warmup_count += 1
+                if warmup_count >= NUM_SPLITS:
+                    break
+
+            drained, elapsed = bench_chunk(table, chunk, STEPS)
+            rows_per_sec = drained / elapsed if elapsed > 0 else float("inf")
+            ms_per_step = elapsed / STEPS * 1000
+
+            print(
+                f"{chunk:>6}  {drained:>6}  {elapsed:>7.3f}s  "
+                f"{rows_per_sec:>10.0f}  {ms_per_step:>8.1f}ms"
+            )
+
+    print()
+    print("Done.")
+
+
+if __name__ == "__main__":
+    main()
@@ -47,6 +47,10 @@ repository = "https://github.com/lancedb/lancedb"
 pylance = [
    "pylance>=5.0.0b5",
 ]
+# A library only needs the OpenTelemetry API; the application supplies and
+# configures the SDK (the actual exporter/reader). See
+# https://opentelemetry.io/docs/languages/python/instrumentation/
+otel = ["opentelemetry-api"]
 tests = [
    "aiohttp>=3.9.0",
    "boto3>=1.28.57",
@@ -61,6 +65,7 @@ tests = [
    "pylance>=5.0.0b5",
    "requests>=2.31.0",
    "datafusion>=52,<53",
+    "opentelemetry-sdk>=1.30.0",
 ]
 dev = [
    "ruff>=0.3.0",
@@ -149,8 +149,14 @@ def connect(

    For object storage, use a URI prefix:

-    >>> db = lancedb.connect("s3://my-bucket/lancedb",
-    ...                      storage_options={"aws_access_key_id": "***"})
+    >>> db = lancedb.connect(  # doctest: +SKIP
+    ...     "s3://my-bucket/lancedb",
+    ...     storage_options={
+    ...         "aws_access_key_id": "***",
+    ...         "aws_secret_access_key": "***",
+    ...         "aws_region": "us-east-1",
+    ...     },
+    ... )

    For tests and temporary data, use an in-memory database:

@@ -1,4 +1,5 @@
-from datetime import datetime, timedelta
+from datetime import date, datetime, timedelta
+from decimal import Decimal
 from typing import Dict, List, Optional, Tuple, Any, TypedDict, Union, Literal

 import pyarrow as pa
@@ -29,6 +30,25 @@ IvfHnswPq: type[HnswPq] = HnswPq
 IvfHnswSq: type[HnswSq] = HnswSq
 IvfHnswFlat: type[HnswFlat] = HnswFlat

+class MetricPoint:
+    name: str
+    kind: str
+    attributes: Dict[str, str]
+    value: Optional[float]
+    buckets: Optional[List[Tuple[str, int]]]
+    count: Optional[int]
+    sum: Optional[float]
+
+class MetricDescription:
+    name: str
+    kind: str
+    unit: Optional[str]
+    description: str
+
+def register_lancedb_metrics_recorder() -> bool: ...
+def lancedb_metrics_catalog() -> List[MetricDescription]: ...
+def snapshot_lancedb_metrics() -> List[MetricPoint]: ...
+
 class PyExpr:
    """A type-safe DataFusion expression node (Rust-side handle)."""

@@ -53,7 +73,9 @@ class PyExpr:
    def to_sql(self) -> str: ...

 def expr_col(name: str) -> PyExpr: ...
-def expr_lit(value: Union[bool, int, float, str, bytes]) -> PyExpr: ...
+def expr_lit(
+    value: Union[bool, int, float, str, bytes, date, datetime, Decimal],
+) -> PyExpr: ...
 def expr_func(name: str, args: List[PyExpr]) -> PyExpr: ...

 class Session:
@@ -226,6 +248,7 @@ class Table:
    async def set_unenforced_primary_key(self, columns: List[str]) -> None: ...
    async def set_lsm_write_spec(self, spec: LsmWriteSpec) -> None: ...
    async def unset_lsm_write_spec(self) -> None: ...
+    async def get_lsm_write_spec(self) -> Optional[LsmWriteSpec]: ...
    async def close_lsm_writers(self) -> None: ...
    @property
    def tags(self) -> Tags: ...
@@ -282,6 +305,23 @@ async def connect(
    namespace_client_properties: Optional[Dict[str, str]] = None,
    oauth_config: Optional[Any] = None,
 ) -> Connection: ...
+def connect_namespace(
+    namespace_client_impl: str,
+    namespace_client_properties: Dict[str, str],
+    read_consistency_interval: Optional[float] = None,
+    storage_options: Optional[Dict[str, str]] = None,
+    session: Optional[Session] = None,
+    namespace_client_pushdown_operations: Optional[List[str]] = None,
+) -> Connection: ...
+def connect_namespace_client(
+    namespace_client: Any,
+    read_consistency_interval: Optional[float] = None,
+    storage_options: Optional[Dict[str, str]] = None,
+    session: Optional[Session] = None,
+    namespace_client_pushdown_operations: Optional[List[str]] = None,
+    namespace_client_impl: Optional[str] = None,
+    namespace_client_properties: Optional[Dict[str, str]] = None,
+) -> Connection: ...

 class RecordBatchStream:
    @property
@@ -2,7 +2,7 @@
 # SPDX-FileCopyrightText: Copyright The LanceDB Authors

 from functools import cached_property
-from typing import TYPE_CHECKING, List, Optional, Sequence, Union
+from typing import TYPE_CHECKING, Any, List, Optional, Sequence, Union

 import numpy as np

@@ -56,6 +56,16 @@ class OllamaEmbeddings(TextEmbeddingFunction):
        embeddings = self._compute_embedding(texts)
        return list(embeddings)

+    def __getstate__(self) -> dict[str, Any]:
+        state = super().__getstate__()
+        state["__dict__"] = {
+            k: v for k, v in state["__dict__"].items() if k != "_ollama_client"
+        }
+        return state
+
+    def __setstate__(self, state: dict[str, Any]) -> None:
+        super().__setstate__(state)
+
    @cached_property
    def _ollama_client(self) -> "ollama.Client":
        ollama = attempt_import_or_raise("ollama")
@@ -19,6 +19,8 @@ operators::

 from __future__ import annotations

+from datetime import date, datetime
+from decimal import Decimal
 from typing import Iterable, Union

 import pyarrow as pa
@@ -63,7 +65,7 @@ def _coerce(value: "ExprLike") -> "Expr":


 # Type alias used in annotations.
-ExprLike = Union["Expr", bool, int, float, str, bytes]
+ExprLike = Union["Expr", bool, int, float, str, bytes, date, datetime, Decimal]


 class Expr:
@@ -118,10 +120,18 @@ class Expr:
        """Logical AND (``expr_a & expr_b``)."""
        return Expr(self._inner.and_(_coerce(other)._inner))

+    def __rand__(self, other: ExprLike) -> "Expr":
+        """Right-hand logical AND (``True & expr``)."""
+        return Expr(_coerce(other)._inner.and_(self._inner))
+
    def __or__(self, other: "Expr") -> "Expr":
        """Logical OR (``expr_a | expr_b``)."""
        return Expr(self._inner.or_(_coerce(other)._inner))

+    def __ror__(self, other: ExprLike) -> "Expr":
+        """Right-hand logical OR (``False | expr``)."""
+        return Expr(_coerce(other)._inner.or_(self._inner))
+
    def __invert__(self) -> "Expr":
        """Logical NOT (``~expr``)."""
        return Expr(self._inner.not_())
@@ -266,13 +276,14 @@ def col(name: str) -> Expr:
    return Expr(expr_col(name))


-def lit(value: Union[bool, int, float, str, bytes]) -> Expr:
+def lit(value: Union[bool, int, float, str, bytes, date, datetime, Decimal]) -> Expr:
    """Create a literal (constant) value expression.

    Parameters
    ----------
    value:
-        A Python ``bool``, ``int``, ``float``, ``str``, or ``bytes``.
+        A Python ``bool``, ``int``, ``float``, ``str``, ``bytes``, ``date``,
+        ``datetime``, or ``Decimal``.

    Examples
    --------
@@ -280,6 +291,9 @@ def lit(value: Union[bool, int, float, str, bytes]) -> Expr:
    >>> col("price") * lit(1.1)
    Expr((price * 1.1))
    """
+    if not isinstance(value, (bool, int, float, str, bytes, date, datetime, Decimal)):
+        raise TypeError(f"Unsupported literal type: {type(value).__name__}")
+
    return Expr(expr_lit(value))


@@ -51,6 +51,15 @@ class LanceMergeInsertBuilder(object):
        If there are multiple matches then the behavior is undefined.
        Currently this causes multiple copies of the row to be created
        but that behavior is subject to change.
+
+        Parameters
+        ----------
+        where: Optional[str], default None
+            An optional filter to limit which rows are updated. Column
+            references in this expression must be prefixed with "target."
+            to refer to the existing table data. For example, to only
+            update rows where the existing color is red, use:
+            ``where="target.color = 'red'"``
        """
        self._when_matched_update_all = True
        self._when_matched_update_all_condition = where
@@ -38,15 +38,13 @@ from lance_namespace_urllib3_client.models.query_table_request_vector import (
    QueryTableRequestVector,
 )
 from lance_namespace_urllib3_client.models.string_fts_query import StringFtsQuery
-from lance_namespace.errors import TableNotFoundError
-from lancedb._lancedb import connect_namespace_client as _connect_namespace_client
+from lance_namespace.errors import NamespaceNotEmptyError, TableNotFoundError
+from lancedb._lancedb import (
+    connect_namespace as _connect_namespace,
+    connect_namespace_client as _connect_namespace_client,
+)
 from lancedb.background_loop import LOOP
 from lancedb.db import AsyncConnection, DBConnection
-from lancedb.namespace_utils import (
-    _normalize_create_namespace_mode,
-    _normalize_drop_namespace_mode,
-    _normalize_drop_namespace_behavior,
-)
 from lance_namespace import (
    LanceNamespace,
    connect as namespace_connect,
@@ -55,13 +53,6 @@ from lance_namespace import (
    DropNamespaceResponse,
    ListNamespacesResponse,
    ListTablesResponse,
-    ListTablesRequest,
-    DescribeNamespaceRequest,
-    DropTableRequest,
-    RenameTableRequest,
-    ListNamespacesRequest,
-    CreateNamespaceRequest,
-    DropNamespaceRequest,
 )
 from lancedb.table import AsyncTable, LanceTable, Table
 from lancedb.util import validate_table_name
@@ -386,6 +377,10 @@ def _builds_namespace_natively(
    return namespace_client_impl == "rest" and bool(namespace_client_properties)


+def _supports_native_namespace(namespace_client_impl: str) -> bool:
+    return namespace_client_impl in {"dir", "rest"}
+
+
 class LanceNamespaceDBConnection(DBConnection):
    """
    A LanceDB connection that uses a namespace for table management.
@@ -396,7 +391,7 @@ class LanceNamespaceDBConnection(DBConnection):

    def __init__(
        self,
-        namespace_client: LanceNamespace,
+        namespace_client: Optional[LanceNamespace] = None,
        *,
        read_consistency_interval: Optional[timedelta] = None,
        storage_options: Optional[Dict[str, str]] = None,
@@ -404,6 +399,7 @@ class LanceNamespaceDBConnection(DBConnection):
        namespace_client_pushdown_operations: Optional[List[str]] = None,
        namespace_client_impl: Optional[str] = None,
        namespace_client_properties: Optional[Dict[str, str]] = None,
+        _inner: Optional[AsyncConnection] = None,
    ):
        """
        Initialize a namespace-based LanceDB connection.
@@ -445,30 +441,36 @@ class LanceNamespaceDBConnection(DBConnection):
        )
        self._namespace_client_impl = namespace_client_impl
        self._namespace_client_properties = namespace_client_properties
-        # When the namespace client is built natively (see Rust
-        # ``build_namespace_natively``), the underlying Rust table performs
-        # QueryTable pushdown through the read-freshness context provider, which
-        # the pure-Python ``query_table`` path bypasses.
-        self._route_pushdown_to_rust = _builds_namespace_natively(
+        # When the namespace connection or client is built natively in Rust, the
+        # underlying Rust table performs QueryTable pushdown through the
+        # read-freshness context provider, which the pure-Python ``query_table``
+        # path bypasses.
+        self._route_pushdown_to_rust = _inner is not None or _builds_namespace_natively(
            namespace_client_impl, namespace_client_properties
        )
-        self._inner = AsyncConnection(
-            _connect_namespace_client(
-                namespace_client,
-                read_consistency_interval=(
-                    read_consistency_interval.total_seconds()
-                    if read_consistency_interval is not None
-                    else None
-                ),
-                storage_options=self.storage_options or None,
-                session=session,
-                namespace_client_pushdown_operations=(
-                    list(self._namespace_client_pushdown_operations)
-                ),
-                namespace_client_impl=namespace_client_impl,
-                namespace_client_properties=namespace_client_properties,
+        if _inner is not None:
+            self._inner = _inner
+        else:
+            if namespace_client is None:
+                raise ValueError("namespace_client is required without a native _inner")
+            self._inner = AsyncConnection(
+                _connect_namespace_client(
+                    namespace_client,
+                    read_consistency_interval=(
+                        read_consistency_interval.total_seconds()
+                        if read_consistency_interval is not None
+                        else None
+                    ),
+                    storage_options=self.storage_options or None,
+                    session=session,
+                    namespace_client_pushdown_operations=(
+                        list(self._namespace_client_pushdown_operations)
+                    ),
+                    namespace_client_impl=namespace_client_impl,
+                    namespace_client_properties=namespace_client_properties,
+                )
            )
-        )
+        self._uri = self._inner.uri

    @override
    def serialize(self) -> str:
@@ -514,11 +516,11 @@ class LanceNamespaceDBConnection(DBConnection):
        )
        if namespace_path is None:
            namespace_path = []
-        request = ListTablesRequest(
-            id=namespace_path, page_token=page_token, limit=limit
+        return LOOP.run(
+            self._inner.table_names(
+                namespace_path=namespace_path, start_after=page_token, limit=limit
+            )
        )
-        response = self._namespace_client.list_tables(request)
-        return response.tables if response.tables else []

    @override
    def create_table(
@@ -589,8 +591,8 @@ class LanceNamespaceDBConnection(DBConnection):
                    index_cache_size=index_cache_size,
                )
            )
-        except RuntimeError as e:
-            if "Table not found" in str(e):
+        except (RuntimeError, ValueError) as e:
+            if "Table not found" in str(e) or "was not found" in str(e):
                table_id = namespace_path + [name]
                raise TableNotFoundError(f"Table not found: {'$'.join(table_id)}")
            raise
@@ -612,12 +614,9 @@ class LanceNamespaceDBConnection(DBConnection):

    @override
    def drop_table(self, name: str, namespace_path: Optional[List[str]] = None):
-        # Use namespace drop_table directly
        if namespace_path is None:
            namespace_path = []
-        table_id = namespace_path + [name]
-        request = DropTableRequest(id=table_id)
-        self._namespace_client.drop_table(request)
+        LOOP.run(self._inner.drop_table(name, namespace_path=namespace_path))

    @override
    def rename_table(
@@ -631,14 +630,19 @@ class LanceNamespaceDBConnection(DBConnection):
            cur_namespace_path = []
        if new_namespace_path is None:
            new_namespace_path = []
-        cur_table_id = cur_namespace_path + [cur_name]
-        new_namespace_id = new_namespace_path if new_namespace_path else None
-        request = RenameTableRequest(
-            id=cur_table_id,
-            new_table_name=new_name,
-            new_namespace_id=new_namespace_id,
-        )
-        self._namespace_client.rename_table(request)
+        try:
+            LOOP.run(
+                self._inner.rename_table(
+                    cur_name,
+                    new_name,
+                    cur_namespace_path=cur_namespace_path,
+                    new_namespace_path=new_namespace_path,
+                )
+            )
+        except RuntimeError as e:
+            if "rename_table not implemented" in str(e):
+                raise NotImplementedError("rename_table not implemented") from e
+            raise

    @override
    def drop_database(self):
@@ -650,8 +654,7 @@ class LanceNamespaceDBConnection(DBConnection):
    def drop_all_tables(self, namespace_path: Optional[List[str]] = None):
        if namespace_path is None:
            namespace_path = []
-        for table_name in self.table_names(namespace_path=namespace_path):
-            self.drop_table(table_name, namespace_path=namespace_path)
+        LOOP.run(self._inner.drop_all_tables(namespace_path=namespace_path))

    @override
    def list_namespaces(
@@ -681,13 +684,10 @@ class LanceNamespaceDBConnection(DBConnection):
        """
        if namespace_path is None:
            namespace_path = []
-        request = ListNamespacesRequest(
-            id=namespace_path, page_token=page_token, limit=limit
-        )
-        response = self._namespace_client.list_namespaces(request)
-        return ListNamespacesResponse(
-            namespaces=response.namespaces if response.namespaces else [],
-            page_token=response.page_token,
+        return LOOP.run(
+            self._inner.list_namespaces(
+                namespace_path=namespace_path, page_token=page_token, limit=limit
+            )
        )

    @override
@@ -715,14 +715,12 @@ class LanceNamespaceDBConnection(DBConnection):
        CreateNamespaceResponse
            Response containing the properties of the created namespace.
        """
-        request = CreateNamespaceRequest(
-            id=namespace_path,
-            mode=_normalize_create_namespace_mode(mode),
-            properties=properties,
-        )
-        response = self._namespace_client.create_namespace(request)
-        return CreateNamespaceResponse(
-            properties=response.properties if hasattr(response, "properties") else None
+        return LOOP.run(
+            self._inner.create_namespace(
+                namespace_path=namespace_path,
+                mode=mode,
+                properties=properties,
+            )
        )

    @override
@@ -750,20 +748,18 @@ class LanceNamespaceDBConnection(DBConnection):
        DropNamespaceResponse
            Response containing properties and transaction_id if applicable.
        """
-        request = DropNamespaceRequest(
-            id=namespace_path,
-            mode=_normalize_drop_namespace_mode(mode),
-            behavior=_normalize_drop_namespace_behavior(behavior),
-        )
-        response = self._namespace_client.drop_namespace(request)
-        return DropNamespaceResponse(
-            properties=(
-                response.properties if hasattr(response, "properties") else None
-            ),
-            transaction_id=(
-                response.transaction_id if hasattr(response, "transaction_id") else None
-            ),
-        )
+        try:
+            return LOOP.run(
+                self._inner.drop_namespace(
+                    namespace_path=namespace_path,
+                    mode=mode,
+                    behavior=behavior,
+                )
+            )
+        except RuntimeError as e:
+            if "Namespace not empty" in str(e):
+                raise NamespaceNotEmptyError(str(e)) from e
+            raise

    @override
    def describe_namespace(
@@ -782,11 +778,7 @@ class LanceNamespaceDBConnection(DBConnection):
        DescribeNamespaceResponse
            Response containing the namespace properties.
        """
-        request = DescribeNamespaceRequest(id=namespace_path)
-        response = self._namespace_client.describe_namespace(request)
-        return DescribeNamespaceResponse(
-            properties=response.properties if hasattr(response, "properties") else None
-        )
+        return LOOP.run(self._inner.describe_namespace(namespace_path))

    @override
    def list_tables(
@@ -816,13 +808,10 @@ class LanceNamespaceDBConnection(DBConnection):
        """
        if namespace_path is None:
            namespace_path = []
-        request = ListTablesRequest(
-            id=namespace_path, page_token=page_token, limit=limit
-        )
-        response = self._namespace_client.list_tables(request)
-        return ListTablesResponse(
-            tables=response.tables if response.tables else [],
-            page_token=response.page_token,
+        return LOOP.run(
+            self._inner.list_tables(
+                namespace_path=namespace_path, page_token=page_token, limit=limit
+            )
        )

    def _lance_table_from_uri(
@@ -878,6 +867,18 @@ class LanceNamespaceDBConnection(DBConnection):
        LanceNamespace
            The namespace client for this connection.
        """
+        if self._namespace_client is None:
+            if (
+                self._namespace_client_impl is None
+                or self._namespace_client_properties is None
+            ):
+                raise ValueError(
+                    "Cannot construct a Python namespace client without "
+                    "namespace implementation properties"
+                )
+            self._namespace_client = namespace_connect(
+                self._namespace_client_impl, self._namespace_client_properties
+            )
        return self._namespace_client


@@ -891,7 +892,7 @@ class AsyncLanceNamespaceDBConnection:

    def __init__(
        self,
-        namespace_client: LanceNamespace,
+        namespace_client: Optional[LanceNamespace] = None,
        *,
        read_consistency_interval: Optional[timedelta] = None,
        storage_options: Optional[Dict[str, str]] = None,
@@ -899,6 +900,7 @@ class AsyncLanceNamespaceDBConnection:
        namespace_client_pushdown_operations: Optional[List[str]] = None,
        namespace_client_impl: Optional[str] = None,
        namespace_client_properties: Optional[Dict[str, str]] = None,
+        _inner: Optional[AsyncConnection] = None,
    ):
        """
        Initialize an async namespace-based LanceDB connection.
@@ -940,29 +942,35 @@ class AsyncLanceNamespaceDBConnection:
        )
        self._namespace_client_impl = namespace_client_impl
        self._namespace_client_properties = namespace_client_properties
-        # See LanceNamespaceDBConnection: when built natively the Rust table runs
-        # QueryTable pushdown through the read-freshness provider, so defer to it
-        # rather than the urllib3 client (which omits x-lancedb-min-timestamp).
-        self._route_pushdown_to_rust = _builds_namespace_natively(
+        # See LanceNamespaceDBConnection: when Rust owns the namespace
+        # connection/client, its table performs QueryTable pushdown through the
+        # read-freshness provider, so defer to it rather than the urllib3 client
+        # path (which omits x-lancedb-min-timestamp).
+        self._route_pushdown_to_rust = _inner is not None or _builds_namespace_natively(
            namespace_client_impl, namespace_client_properties
        )
-        self._inner = AsyncConnection(
-            _connect_namespace_client(
-                namespace_client,
-                read_consistency_interval=(
-                    read_consistency_interval.total_seconds()
-                    if read_consistency_interval is not None
-                    else None
-                ),
-                storage_options=self.storage_options or None,
-                session=session,
-                namespace_client_pushdown_operations=(
-                    list(self._namespace_client_pushdown_operations)
-                ),
-                namespace_client_impl=namespace_client_impl,
-                namespace_client_properties=namespace_client_properties,
+        if _inner is not None:
+            self._inner = _inner
+        else:
+            if namespace_client is None:
+                raise ValueError("namespace_client is required without a native _inner")
+            self._inner = AsyncConnection(
+                _connect_namespace_client(
+                    namespace_client,
+                    read_consistency_interval=(
+                        read_consistency_interval.total_seconds()
+                        if read_consistency_interval is not None
+                        else None
+                    ),
+                    storage_options=self.storage_options or None,
+                    session=session,
+                    namespace_client_pushdown_operations=(
+                        list(self._namespace_client_pushdown_operations)
+                    ),
+                    namespace_client_impl=namespace_client_impl,
+                    namespace_client_properties=namespace_client_properties,
+                )
            )
-        )

    async def table_names(
        self,
@@ -986,11 +994,9 @@ class AsyncLanceNamespaceDBConnection:
        )
        if namespace_path is None:
            namespace_path = []
-        request = ListTablesRequest(
-            id=namespace_path, page_token=page_token, limit=limit
+        return await self._inner.table_names(
+            namespace_path=namespace_path, start_after=page_token, limit=limit
        )
-        response = self._namespace_client.list_tables(request)
-        return response.tables if response.tables else []

    async def create_table(
        self,
@@ -1053,8 +1059,8 @@ class AsyncLanceNamespaceDBConnection:
                storage_options=storage_options,
                index_cache_size=index_cache_size,
            )
-        except RuntimeError as e:
-            if "Table not found" in str(e):
+        except (RuntimeError, ValueError) as e:
+            if "Table not found" in str(e) or "was not found" in str(e):
                table_id = namespace_path + [name]
                raise TableNotFoundError(f"Table not found: {'$'.join(table_id)}")
            raise
@@ -1075,9 +1081,7 @@ class AsyncLanceNamespaceDBConnection:
        """Drop a table from the namespace."""
        if namespace_path is None:
            namespace_path = []
-        table_id = namespace_path + [name]
-        request = DropTableRequest(id=table_id)
-        self._namespace_client.drop_table(request)
+        await self._inner.drop_table(name, namespace_path=namespace_path)

    async def rename_table(
        self,
@@ -1091,14 +1095,17 @@ class AsyncLanceNamespaceDBConnection:
            cur_namespace_path = []
        if new_namespace_path is None:
            new_namespace_path = []
-        cur_table_id = cur_namespace_path + [cur_name]
-        new_namespace_id = new_namespace_path if new_namespace_path else None
-        request = RenameTableRequest(
-            id=cur_table_id,
-            new_table_name=new_name,
-            new_namespace_id=new_namespace_id,
-        )
-        self._namespace_client.rename_table(request)
+        try:
+            await self._inner.rename_table(
+                cur_name,
+                new_name,
+                cur_namespace_path=cur_namespace_path,
+                new_namespace_path=new_namespace_path,
+            )
+        except RuntimeError as e:
+            if "rename_table not implemented" in str(e):
+                raise NotImplementedError("rename_table not implemented") from e
+            raise

    async def drop_database(self):
        """Deprecated method."""
@@ -1110,9 +1117,7 @@ class AsyncLanceNamespaceDBConnection:
        """Drop all tables in the namespace."""
        if namespace_path is None:
            namespace_path = []
-        table_names = await self.table_names(namespace_path=namespace_path)
-        for table_name in table_names:
-            await self.drop_table(table_name, namespace_path=namespace_path)
+        await self._inner.drop_all_tables(namespace_path=namespace_path)

    async def list_namespaces(
        self,
@@ -1141,13 +1146,8 @@ class AsyncLanceNamespaceDBConnection:
        """
        if namespace_path is None:
            namespace_path = []
-        request = ListNamespacesRequest(
-            id=namespace_path, page_token=page_token, limit=limit
-        )
-        response = self._namespace_client.list_namespaces(request)
-        return ListNamespacesResponse(
-            namespaces=response.namespaces if response.namespaces else [],
-            page_token=response.page_token,
+        return await self._inner.list_namespaces(
+            namespace_path=namespace_path, page_token=page_token, limit=limit
        )

    async def create_namespace(
@@ -1174,15 +1174,11 @@ class AsyncLanceNamespaceDBConnection:
        CreateNamespaceResponse
            Response containing the properties of the created namespace.
        """
-        request = CreateNamespaceRequest(
-            id=namespace_path,
-            mode=_normalize_create_namespace_mode(mode),
+        return await self._inner.create_namespace(
+            namespace_path=namespace_path,
+            mode=mode,
            properties=properties,
        )
-        response = self._namespace_client.create_namespace(request)
-        return CreateNamespaceResponse(
-            properties=response.properties if hasattr(response, "properties") else None
-        )

    async def drop_namespace(
        self,
@@ -1208,20 +1204,16 @@ class AsyncLanceNamespaceDBConnection:
        DropNamespaceResponse
            Response containing properties and transaction_id if applicable.
        """
-        request = DropNamespaceRequest(
-            id=namespace_path,
-            mode=_normalize_drop_namespace_mode(mode),
-            behavior=_normalize_drop_namespace_behavior(behavior),
-        )
-        response = self._namespace_client.drop_namespace(request)
-        return DropNamespaceResponse(
-            properties=(
-                response.properties if hasattr(response, "properties") else None
-            ),
-            transaction_id=(
-                response.transaction_id if hasattr(response, "transaction_id") else None
-            ),
-        )
+        try:
+            return await self._inner.drop_namespace(
+                namespace_path=namespace_path,
+                mode=mode,
+                behavior=behavior,
+            )
+        except RuntimeError as e:
+            if "Namespace not empty" in str(e):
+                raise NamespaceNotEmptyError(str(e)) from e
+            raise

    async def describe_namespace(
        self, namespace_path: List[str]
@@ -1239,11 +1231,7 @@ class AsyncLanceNamespaceDBConnection:
        DescribeNamespaceResponse
            Response containing the namespace properties.
        """
-        request = DescribeNamespaceRequest(id=namespace_path)
-        response = self._namespace_client.describe_namespace(request)
-        return DescribeNamespaceResponse(
-            properties=response.properties if hasattr(response, "properties") else None
-        )
+        return await self._inner.describe_namespace(namespace_path)

    async def list_tables(
        self,
@@ -1272,13 +1260,8 @@ class AsyncLanceNamespaceDBConnection:
        """
        if namespace_path is None:
            namespace_path = []
-        request = ListTablesRequest(
-            id=namespace_path, page_token=page_token, limit=limit
-        )
-        response = self._namespace_client.list_tables(request)
-        return ListTablesResponse(
-            tables=response.tables if response.tables else [],
-            page_token=response.page_token,
+        return await self._inner.list_tables(
+            namespace_path=namespace_path, page_token=page_token, limit=limit
        )

    async def namespace_client(self) -> LanceNamespace:
@@ -1292,6 +1275,18 @@ class AsyncLanceNamespaceDBConnection:
        LanceNamespace
            The namespace client for this connection.
        """
+        if self._namespace_client is None:
+            if (
+                self._namespace_client_impl is None
+                or self._namespace_client_properties is None
+            ):
+                raise ValueError(
+                    "Cannot construct a Python namespace client without "
+                    "namespace implementation properties"
+                )
+            self._namespace_client = namespace_connect(
+                self._namespace_client_impl, self._namespace_client_properties
+            )
        return self._namespace_client


@@ -1342,6 +1337,32 @@ def connect_namespace(
    LanceNamespaceDBConnection
        A namespace-based connection to LanceDB
    """
+    if _supports_native_namespace(namespace_client_impl):
+        inner = AsyncConnection(
+            _connect_namespace(
+                namespace_client_impl,
+                namespace_client_properties,
+                read_consistency_interval=(
+                    read_consistency_interval.total_seconds()
+                    if read_consistency_interval is not None
+                    else None
+                ),
+                storage_options=storage_options,
+                session=session,
+                namespace_client_pushdown_operations=namespace_client_pushdown_operations,
+            )
+        )
+        return LanceNamespaceDBConnection(
+            namespace_client=None,
+            read_consistency_interval=read_consistency_interval,
+            storage_options=storage_options,
+            session=session,
+            namespace_client_pushdown_operations=namespace_client_pushdown_operations,
+            namespace_client_impl=namespace_client_impl,
+            namespace_client_properties=namespace_client_properties,
+            _inner=inner,
+        )
+
    namespace_client = namespace_connect(
        namespace_client_impl, namespace_client_properties
    )
@@ -1417,6 +1438,32 @@ def connect_namespace_async(
    ...     tables = await db.table_names()
    ...     table = await db.create_table("my_table", schema=schema)
    """
+    if _supports_native_namespace(namespace_client_impl):
+        inner = AsyncConnection(
+            _connect_namespace(
+                namespace_client_impl,
+                namespace_client_properties,
+                read_consistency_interval=(
+                    read_consistency_interval.total_seconds()
+                    if read_consistency_interval is not None
+                    else None
+                ),
+                storage_options=storage_options,
+                session=session,
+                namespace_client_pushdown_operations=namespace_client_pushdown_operations,
+            )
+        )
+        return AsyncLanceNamespaceDBConnection(
+            namespace_client=None,
+            read_consistency_interval=read_consistency_interval,
+            storage_options=storage_options,
+            session=session,
+            namespace_client_pushdown_operations=namespace_client_pushdown_operations,
+            namespace_client_impl=namespace_client_impl,
+            namespace_client_properties=namespace_client_properties,
+            _inner=inner,
+        )
+
    namespace_client = namespace_connect(
        namespace_client_impl, namespace_client_properties
    )
@@ -0,0 +1,170 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+"""Bridge LanceDB's internal metrics into OpenTelemetry.
+
+LanceDB (through Lance core) publishes metrics (currently object store request
+counts, bytes, latency, errors, and throttles) through the Rust ``metrics``
+facade. This module installs a process-global recorder that aggregates them and
+registers OpenTelemetry observable instruments that report the aggregated values
+into the user's ``MeterProvider``.
+
+The bridge is generic: every metric LanceDB describes is surfaced automatically,
+with no per-metric Python code. Histograms have no asynchronous OpenTelemetry
+instrument, so each is exported Prometheus-style as cumulative ``le`` buckets
+plus ``_count`` and ``_sum`` observable counters.
+"""
+
+from __future__ import annotations
+
+import warnings
+from typing import TYPE_CHECKING, Optional
+
+from ._lancedb import (
+    lancedb_metrics_catalog,
+    register_lancedb_metrics_recorder,
+    snapshot_lancedb_metrics,
+)
+
+if TYPE_CHECKING:
+    from opentelemetry.metrics import MeterProvider
+
+_INSTRUMENTED = False
+
+
+def instrument_lancedb_metrics(
+    meter_provider: Optional["MeterProvider"] = None,
+) -> bool:
+    """Register LanceDB metrics as OpenTelemetry observable instruments.
+
+    Installs a process-global metrics recorder and creates one observable
+    instrument per LanceDB metric on the given (or global) ``MeterProvider``. The
+    user's configured ``MetricReader`` then collects them on its own schedule.
+
+    Counters and gauges map directly to observable counters/gauges. Each
+    histogram is exported as cumulative ``le`` bucket counts (``<name>_bucket``,
+    with an ``le`` attribute) plus ``<name>_count`` and ``<name>_sum``.
+
+    Parameters
+    ----------
+    meter_provider : opentelemetry.metrics.MeterProvider, optional
+        The provider to register instruments on. Defaults to the global provider
+        from ``opentelemetry.metrics.get_meter_provider()``.
+
+    Returns
+    -------
+    bool
+        ``True`` if the recorder is installed and instruments are registered.
+        ``False`` if a different ``metrics`` recorder is already installed in
+        this process (``metrics`` permits only one global recorder), in which
+        case a warning is emitted and no instruments are created.
+
+    Notes
+    -----
+    Requires the OpenTelemetry API (``pip install lancedb[otel]``) and, to
+    actually export, an OpenTelemetry SDK (``pip install opentelemetry-sdk``)
+    configured by the application. Calling this more than once is safe;
+    instruments are created only on the first successful call.
+    """
+    global _INSTRUMENTED
+
+    try:
+        from opentelemetry.metrics import Observation, get_meter_provider
+    except ImportError as exc:
+        raise ImportError(
+            "instrument_lancedb_metrics requires the OpenTelemetry API/SDK. "
+            "Install it with `pip install lancedb[otel]` or "
+            "`pip install opentelemetry-sdk`."
+        ) from exc
+
+    if not register_lancedb_metrics_recorder():
+        warnings.warn(
+            "Could not install the LanceDB metrics recorder: another `metrics` "
+            "recorder is already installed in this process. LanceDB metrics will "
+            "not be exported via OpenTelemetry.",
+            stacklevel=2,
+        )
+        return False
+
+    if _INSTRUMENTED:
+        return True
+
+    provider = meter_provider or get_meter_provider()
+    meter = provider.get_meter("lancedb")
+
+    def scalar_callback(metric_name: str):
+        def callback(_options):
+            return [
+                Observation(point.value, point.attributes)
+                for point in snapshot_lancedb_metrics()
+                if point.name == metric_name and point.value is not None
+            ]
+
+        return callback
+
+    def bucket_callback(metric_name: str):
+        def callback(_options):
+            observations = []
+            for point in snapshot_lancedb_metrics():
+                if point.name != metric_name or point.buckets is None:
+                    continue
+                for le, cumulative in point.buckets:
+                    attributes = dict(point.attributes)
+                    attributes["le"] = le
+                    observations.append(Observation(cumulative, attributes))
+            return observations
+
+        return callback
+
+    def field_callback(metric_name: str, field: str):
+        def callback(_options):
+            observations = []
+            for point in snapshot_lancedb_metrics():
+                if point.name != metric_name:
+                    continue
+                value = getattr(point, field)
+                if value is not None:
+                    observations.append(Observation(value, point.attributes))
+            return observations
+
+        return callback
+
+    for desc in lancedb_metrics_catalog():
+        unit = desc.unit or ""
+        if desc.kind == "counter":
+            meter.create_observable_counter(
+                desc.name,
+                callbacks=[scalar_callback(desc.name)],
+                unit=unit,
+                description=desc.description,
+            )
+        elif desc.kind == "gauge":
+            meter.create_observable_gauge(
+                desc.name,
+                callbacks=[scalar_callback(desc.name)],
+                unit=unit,
+                description=desc.description,
+            )
+        elif desc.kind == "histogram":
+            # `_bucket` and `_count` observe cumulative sample counts, not the
+            # histogram's measured quantity, so they are unitless; only `_sum`
+            # carries the histogram's unit.
+            meter.create_observable_counter(
+                f"{desc.name}_bucket",
+                callbacks=[bucket_callback(desc.name)],
+                description=f"{desc.description} (cumulative buckets)",
+            )
+            meter.create_observable_counter(
+                f"{desc.name}_count",
+                callbacks=[field_callback(desc.name, "count")],
+                description=f"{desc.description} (count)",
+            )
+            meter.create_observable_counter(
+                f"{desc.name}_sum",
+                callbacks=[field_callback(desc.name, "sum")],
+                unit=unit,
+                description=f"{desc.description} (sum)",
+            )
+
+    _INSTRUMENTED = True
+    return True
@@ -11,7 +11,7 @@ import pyarrow as pa
 from ._lancedb import async_permutation_builder, PermutationReader
 from .table import LanceTable, Table
 from .background_loop import LOOP
-from .util import batch_to_tensor, batch_to_tensor_rows
+from .util import batch_to_tensor, batch_to_tensor_dict, batch_to_tensor_rows
 from typing import Any, Callable, Iterator, Literal, Optional, TYPE_CHECKING, Union

 if TYPE_CHECKING:
@@ -65,6 +65,7 @@ class PermutationBuilder:
        counts: Optional[list[int]] = None,
        fixed: Optional[int] = None,
        seed: Optional[int] = None,
+        clump_size: Optional[int] = None,
        split_names: Optional[list[str]] = None,
    ) -> "PermutationBuilder":
        """
@@ -87,6 +88,9 @@ class PermutationBuilder:
        Rows will be randomly assigned to splits.  The optional seed can be provided to
        make the assignment deterministic.

+        If clump_size is provided, rows are shuffled as contiguous groups of that size,
+        preserving I/O locality while still randomising the split assignment.
+
        The optional split_names can be provided to name the splits.  If not provided,
        the splits can only be referenced by their index.
        """
@@ -95,6 +99,7 @@ class PermutationBuilder:
            counts=counts,
            fixed=fixed,
            seed=seed,
+            clump_size=clump_size,
            split_names=split_names,
        )
        return self
@@ -941,6 +946,7 @@ class Permutation:
            "pandas",
            "arrow",
            "torch",
+            "torch_row",
            "torch_col",
            "polars",
        ],
@@ -956,15 +962,19 @@ class Permutation:
        - "python_col" - the batch will be a dict of lists (one entry per column)
        - "pandas" - the batch will be a pandas DataFrame
        - "arrow" - the batch will be a pyarrow RecordBatch
-        - "torch" - the batch will be a list of tensors, one per row
+        - "torch" - the batch will be a list of per-row dicts mapping column
+          name to a 0-D torch tensor. Works with the default
+          ``torch.utils.data.DataLoader`` collate, which stacks the per-row
+          dicts back into a dict of batched tensors.
+        - "torch_row" - the batch will be a list of tensors, one per row
        - "torch_col" - the batch will be a 2D torch tensor (first dim indexes columns)
        - "polars" - the batch will be a polars DataFrame

        Conversion may or may not involve a data copy.  Lance uses Arrow internally
        and so it is able to zero-copy to the arrow and polars formats.

-        Conversion to torch_col will be zero-copy but will only support a subset of data
-        types (numeric types).
+        Conversion to torch and torch_col will be zero-copy but will only support a
+        subset of data types (numeric types).

        Conversion to numpy and/or pandas will typically be zero-copy for numeric
        types.  Conversion of strings, lists, and structs will require creating python
@@ -985,6 +995,8 @@ class Permutation:
        elif format == "arrow":
            return self.with_transform(Transforms.arrow2arrow)
        elif format == "torch":
+            return self.with_transform(batch_to_tensor_dict)
+        elif format == "torch_row":
            return self.with_transform(batch_to_tensor_rows)
        elif format == "torch_col":
            return self.with_transform(batch_to_tensor)
@@ -119,6 +119,27 @@ def _filter_to_sql(filter: Optional[Union[str, Expr]]) -> Optional[str]:
    return filter


+def _combine_where(
+    existing: Optional[Union[str, Expr]], new: Union[str, Expr]
+) -> Union[str, Expr]:
+    """Combine a new filter with an existing one using a logical AND.
+
+    Calling ``where`` more than once composes the filters with AND instead of
+    replacing the previous filter. Two :class:`~lancedb.expr.Expr` filters are
+    combined as an expression; otherwise both filters are lowered to SQL strings
+    and combined as SQL.
+    """
+    if existing is None:
+        return new
+    existing_is_expr = isinstance(existing, Expr)
+    new_is_expr = isinstance(new, Expr)
+    if existing_is_expr and new_is_expr:
+        return existing & new
+    existing_sql = existing.to_sql() if existing_is_expr else existing
+    new_sql = new.to_sql() if new_is_expr else new
+    return f"({existing_sql}) AND ({new_sql})"
+
+
 def _projection_to_scanner_kwargs(
    columns: Optional[
        Union[
@@ -1148,8 +1169,13 @@ class LanceQueryBuilder(ABC):
        -------
        LanceQueryBuilder
            The LanceQueryBuilder object.
+
+        Notes
+        -----
+        Calling this multiple times combines the filters with a logical AND
+        rather than replacing the previous filter.
        """
-        self._where = where
+        self._where = _combine_where(self._where, where)
        self._postfilter = not prefilter
        return self

@@ -1693,8 +1719,13 @@ class LanceVectorQueryBuilder(LanceQueryBuilder):
        -------
        LanceQueryBuilder
            The LanceQueryBuilder object.
+
+        Notes
+        -----
+        Calling this multiple times combines the filters with a logical AND
+        rather than replacing the previous filter.
        """
-        self._where = where
+        self._where = _combine_where(self._where, where)
        if prefilter is not None:
            self._postfilter = not prefilter
        return self
@@ -2894,6 +2925,9 @@ class AsyncStandardQuery(AsyncQueryBase):

        Filtering performance can often be improved by creating a scalar index
        on the filter column(s).
+
+        Calling this multiple times combines the filters with a logical AND
+        rather than replacing the previous filter.
        """
        if isinstance(predicate, Expr):
            self._inner.where_expr(predicate._inner)
@@ -912,6 +912,10 @@ class RemoteTable(Table):
        """Not supported on LanceDB Cloud."""
        return LOOP.run(self._table.unset_lsm_write_spec())

+    def get_lsm_write_spec(self) -> Optional["LsmWriteSpec"]:
+        """Read the installed LsmWriteSpec, or ``None``."""
+        return LOOP.run(self._table.get_lsm_write_spec())
+
    def close_lsm_writers(self) -> None:
        """No-op on LanceDB Cloud (no local shard writers)."""
        return LOOP.run(self._table.close_lsm_writers())
@@ -156,9 +156,16 @@ class MRRReranker(Reranker):
                reciprocal_rank = 1.0 / rank
                mrr_score_map[result_id].append(reciprocal_rank)

+        # MRR averages the reciprocal rank across *all* ranking systems, treating
+        # a system in which a document does not appear as a reciprocal rank of 0.
+        # We therefore divide by the total number of systems, not by the number of
+        # systems the document happens to appear in -- otherwise a document found
+        # by a single ranking would outrank one ranked highly by every system,
+        # defeating the purpose of fusing the rankings.
+        num_systems = len(vector_results)
        final_mrr_scores = {}
        for result_id, reciprocal_ranks in mrr_score_map.items():
-            mean_rr = np.mean(reciprocal_ranks)
+            mean_rr = float(np.sum(reciprocal_ranks)) / num_systems
            final_mrr_scores[result_id] = mean_rr

        combined = pa.concat_tables(vector_results, **self._concat_tables_args)
@@ -0,0 +1,607 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+"""Elastic streaming dataloader for PyTorch.
+
+Provides StreamingDataset, a PyTorch IterableDataset that guarantees:
+
+- **Elastic determinism**: for a fixed (num_splits, shuffle_seed, epoch) the set
+  of samples that forms each global training step is identical regardless of
+  world_size or num_workers.
+- **Resumability**: state_dict / load_state_dict capture per-split consumption
+  counts so training can resume from an exact mid-epoch position even when the
+  distributed topology changes between runs.
+"""
+
+import ctypes
+import logging
+import os
+import random
+import threading
+import time
+from collections import deque
+from concurrent.futures import ThreadPoolExecutor
+from multiprocessing import RawArray
+from typing import Any, Callable, Iterator, Optional
+
+from torch.utils.data import IterableDataset, get_worker_info
+
+from .permutation import (
+    Permutation,
+    Transforms,
+    permutation_builder,
+    _table_from_pickle_state,
+    _table_to_pickle_state,
+)
+
+logger = logging.getLogger(__name__)
+
+# Multiplier used to combine shuffle_seed and epoch into a single permutation
+# seed.  Chosen to be a large prime so different (seed, epoch) pairs produce
+# distinct seeds for any practically encountered epoch count.
+_EPOCH_PRIME = 100003
+
+DEFAULT_READ_BATCH_SIZE = 64
+DEFAULT_PREFETCH_BATCHES = 4
+
+
+class StreamingDataset(IterableDataset):
+    """An elastic, resumable PyTorch IterableDataset backed by a LanceDB table.
+
+    The table is partitioned into ``num_splits`` fixed splits using a
+    deterministic random shuffle controlled by ``shuffle_seed`` and ``epoch``.
+    Each rank is assigned a contiguous block of splits, and within a rank each
+    DataLoader worker is assigned a contiguous sub-block.  Samples are yielded
+    by round-robining over the assigned splits, one sample per split per cycle.
+
+    Internally ``__iter__`` runs a two-stage pipeline:
+
+    - **Stage 1 (I/O)**: one thread pool with ``num_splits * prefetch_batches``
+      workers fetches raw ``RecordBatch`` objects from LanceDB in parallel
+      across all splits and places them in a per-split raw-batch queue.
+    - **Stage 2 (transform)**: a second thread pool with ``os.cpu_count()``
+      workers picks up raw batches, applies the transform, and places the
+      results in a per-split cooked-row queue.
+
+    The main thread round-robins over the cooked queues, yielding one row per
+    split per cycle.
+
+    Parameters
+    ----------
+    table:
+        LanceDB table to stream from.
+    num_splits:
+        Number of fixed splits to partition the table into.  Must be divisible
+        by ``world_size``.  When used with DataLoader workers it must also be
+        divisible by ``world_size * num_workers``.  Defaults to ``world_size``.
+        If the row count (after any ``filter``) is not evenly divisible by
+        ``num_splits``, the surplus rows — at most ``num_splits - 1`` per epoch
+        — are silently dropped to keep all splits the same length.
+    shuffle:
+        Whether to randomly assign rows to splits.  When ``True`` (the
+        default) rows are shuffled using ``shuffle_seed`` and ``epoch``.
+        When ``False`` rows are divided into splits sequentially in storage
+        order, which can be useful for deterministic debugging or evaluation.
+    shuffle_seed:
+        Base seed for the random permutation.  Combined with ``epoch`` so
+        each epoch produces a different ordering.  Pass ``None`` to generate
+        a random seed at construction time.
+    epoch:
+        Current training epoch.  Combined with ``shuffle_seed`` so that each
+        epoch produces a different sample ordering.
+    rank:
+        This process's rank in the distributed training group.
+    world_size:
+        Total number of processes in the distributed training group.
+    read_batch_size:
+        Number of rows fetched from each split in a single ``take_offsets``
+        call.  Larger values amortise per-request overhead (critical on object
+        storage) at the cost of higher memory usage per split buffer.  Defaults
+        to ``DEFAULT_READ_BATCH_SIZE`` (64).
+    prefetch_batches:
+        Number of I/O batches to keep in flight per split.  Higher values
+        overlap storage latency with transform and training compute at the cost
+        of more memory and threads.  Defaults to ``DEFAULT_PREFETCH_BATCHES``
+        (4).
+    columns:
+        Optional list of column names to read.  When set, only those columns
+        are fetched from storage; all others are omitted.  ``None`` (the
+        default) reads every column.
+    shuffle_clump_size:
+        When set, rows are shuffled in contiguous groups of this size rather
+        than individually.  Larger clumps improve I/O locality (important on
+        object storage) at the cost of reduced randomness.  ``None`` (the
+        default) shuffles rows individually.
+    filter:
+        Optional SQL filter expression (e.g. ``"label = 'dog'"``).  Only rows
+        that satisfy the predicate are included in the permutation.  The filter
+        is applied during permutation construction so split sizes reflect the
+        filtered row count.
+    transform:
+        Optional callable applied to each ``pyarrow.RecordBatch`` before rows
+        are yielded.  Receives one batch at a time and must return an iterable
+        whose length equals the number of rows in the batch.  When ``None``
+        (the default) rows are returned as plain Python dicts.
+    worker_info_override:
+        If set, used in place of ``torch.utils.data.get_worker_info()`` to
+        determine the DataLoader worker assignment.  Intended for unit tests
+        that need to simulate multiple workers without spawning real processes.
+        If both this and the real worker info are non-None a warning is logged
+        and the override takes precedence.
+    """
+
+    def __init__(
+        self,
+        table,
+        *,
+        num_splits: Optional[int] = None,
+        shuffle: bool = True,
+        shuffle_seed: Optional[int] = 0,
+        epoch: int = 0,
+        rank: int = 0,
+        world_size: int = 1,
+        read_batch_size: int = DEFAULT_READ_BATCH_SIZE,
+        prefetch_batches: int = DEFAULT_PREFETCH_BATCHES,
+        columns: Optional[list[str]] = None,
+        shuffle_clump_size: Optional[int] = None,
+        filter: Optional[str] = None,
+        transform: Optional[Callable] = None,
+        connection_factory: Optional[Callable[[str], Any]] = None,
+        worker_info_override=None,
+    ):
+        super().__init__()
+        if num_splits is None:
+            num_splits = world_size
+        if shuffle_seed is None:
+            shuffle_seed = random.randrange(2**32)
+        if num_splits % world_size != 0:
+            raise ValueError(
+                f"num_splits ({num_splits}) must be divisible by "
+                f"world_size ({world_size})"
+            )
+
+        self._table = table
+        self._num_splits = num_splits
+        self._shuffle = shuffle
+        self._shuffle_seed = shuffle_seed
+        self._epoch = epoch
+        self._rank = rank
+        self._world_size = world_size
+        self._read_batch_size = read_batch_size
+        self._prefetch_batches = prefetch_batches
+        self._columns = columns
+        self._shuffle_clump_size = shuffle_clump_size
+        self._filter = filter
+        self._transform = transform
+        self._connection_factory = connection_factory
+        self._worker_info_override = worker_info_override
+
+        # Live references to pipeline state, set only while __iter__ is running
+        # in the same process.  Used by the observability properties when the
+        # DataLoader runs with num_workers=0.
+        self._raw_batches_ref: Optional[list[deque]] = None
+        self._cooked_ref: Optional[list[deque]] = None
+        self._fetch_head_ref: Optional[list[int]] = None
+        self._split_sizes_ref: Optional[list[int]] = None
+        self._local_consumed_ref: Optional[list[int]] = None
+
+        # Shared-memory counters written by __iter__ (which may run in a
+        # DataLoader worker process) and read by the observability properties
+        # in the main process.  RawArray is picklable via the forkserver
+        # reduction protocol so it survives the dataset pickle round-trip.
+        # Layout: [unscanned_rows, raw_rows, cooked_rows, consumed_rows,
+        #          bytes_loaded, fetch_time_us, transform_time_us]
+        self._worker_stats: RawArray = RawArray(ctypes.c_int64, 7)
+
+        # Cumulative bytes of Arrow buffer data fetched across all iterations.
+        self._bytes_loaded: int = 0
+        # Cumulative seconds spent in LanceDB I/O and in transform functions.
+        self._fetch_time: float = 0.0
+        self._transform_time: float = 0.0
+
+        # Number of samples each split has already been consumed.  At global
+        # step boundaries all splits have consumed this many samples, so a
+        # single scalar captures the topology-independent checkpoint state.
+        self._resume_offset: int = 0
+
+        # Build the permutation table once, deterministically.
+        builder = permutation_builder(table)
+        if filter is not None:
+            builder = builder.filter(filter)
+        if shuffle:
+            perm_seed = shuffle_seed + epoch * _EPOCH_PRIME
+            self._perm_table = builder.split_random(
+                fixed=num_splits, seed=perm_seed, clump_size=shuffle_clump_size
+            ).execute()
+        else:
+            self._perm_table = builder.split_sequential(fixed=num_splits).execute()
+
+        # Contiguous block of global split indices assigned to this rank.
+        splits_per_rank = num_splits // world_size
+        rank_start = rank * splits_per_rank
+        self._rank_splits: list[int] = list(
+            range(rank_start, rank_start + splits_per_rank)
+        )
+
+    def _resolve_my_splits(self) -> list[int]:
+        """Return the split indices this instance should read in __iter__."""
+        torch_worker_info = get_worker_info()
+        if self._worker_info_override is not None:
+            if torch_worker_info is not None:
+                logger.warning(
+                    "worker_info_override is set but get_worker_info() also returned a "
+                    "non-None value; ignoring the real torch worker info and using the "
+                    "override instead.  This may lead to duplicated or incorrect data "
+                    "from the dataset."
+                )
+            worker_info = self._worker_info_override
+        else:
+            worker_info = torch_worker_info
+
+        if worker_info is None:
+            return self._rank_splits
+
+        num_workers: int = worker_info.num_workers
+        worker_id: int = worker_info.id
+        n_rank_splits = len(self._rank_splits)
+        if n_rank_splits % num_workers != 0:
+            raise ValueError(
+                f"Number of rank splits ({n_rank_splits}) must be divisible by "
+                f"num_workers ({num_workers})"
+            )
+        splits_per_worker = n_rank_splits // num_workers
+        start = worker_id * splits_per_worker
+        return self._rank_splits[start : start + splits_per_worker]
+
+    def __iter__(self) -> Iterator[dict[str, Any]]:
+        if self._raw_batches_ref is not None:
+            raise RuntimeError(
+                "StreamingDataset does not support concurrent iteration. "
+                "Only one active iterator per dataset instance is allowed."
+            )
+        my_splits = self._resolve_my_splits()
+        if not my_splits:
+            return
+
+        # Set identity transform on each Permutation so __getitems__ returns
+        # the raw RecordBatch.  Stage 2 applies the real transform.
+        permutations: list[Permutation] = []
+        for split_idx in my_splits:
+            perm = Permutation.from_tables(
+                self._table, self._perm_table, split=split_idx
+            )
+            if self._columns is not None:
+                perm = perm.select_columns(self._columns)
+            perm = perm.with_transform(lambda batch: batch)
+            if self._resume_offset > 0:
+                perm = perm.with_skip(self._resume_offset)
+            permutations.append(perm)
+
+        n = len(permutations)
+        split_sizes = [perm.num_rows for perm in permutations]
+        initial_offset = self._resume_offset
+        local_consumed = [0] * n
+
+        batch_size = self._read_batch_size
+        max_prefetch = self._prefetch_batches
+        cpu_workers = os.cpu_count() or 1
+        final_transform = (
+            self._transform if self._transform is not None else Transforms.arrow2python
+        )
+
+        # Per-split pipeline state.
+        fetch_head = [0] * n
+        io_pending = [deque() for _ in range(n)]  # Future[RecordBatch]
+        raw_batches = [deque() for _ in range(n)]  # RecordBatch — fetched, awaiting tx
+        tx_pending = [deque() for _ in range(n)]  # Future[list[Any]]
+        cooked = [deque() for _ in range(n)]  # rows ready to yield
+
+        # Limit simultaneous transforms to cpu_workers across all splits.
+        tx_semaphore = threading.Semaphore(cpu_workers)
+
+        # ── Stage 1 helpers ───────────────────────────────────────────────────
+
+        def _io_call(perm, indices):
+            t0 = time.perf_counter()
+            batch = perm.__getitems__(indices)
+            self._bytes_loaded += batch.nbytes
+            self._fetch_time += time.perf_counter() - t0
+            return batch
+
+        def _submit_io(i: int) -> None:
+            remaining = split_sizes[i] - fetch_head[i]
+            if remaining <= 0:
+                return
+            fetch = min(batch_size, remaining)
+            start = fetch_head[i]
+            fetch_head[i] += fetch
+            perm_i = permutations[i]
+            indices = list(range(start, start + fetch))
+            io_pending[i].append(io_pool.submit(_io_call, perm_i, indices))
+
+        def _fill_io(i: int) -> None:
+            while len(io_pending[i]) < max_prefetch and fetch_head[i] < split_sizes[i]:
+                _submit_io(i)
+
+        def _drain_io(i: int) -> None:
+            """Move completed I/O futures into raw_batches non-blockingly."""
+            while io_pending[i] and io_pending[i][0].done():
+                raw_batches[i].append(io_pending[i].popleft().result())
+
+        # ── Stage 2 helpers ───────────────────────────────────────────────────
+
+        def _tx_call_guarded(batch):
+            try:
+                t0 = time.perf_counter()
+                result = final_transform(batch)
+                self._transform_time += time.perf_counter() - t0
+                return result
+            finally:
+                tx_semaphore.release()
+
+        def _try_submit_tx(i: int) -> None:
+            """Submit transforms for raw_batches[i] up to available capacity."""
+            while raw_batches[i] and tx_semaphore.acquire(blocking=False):
+                batch = raw_batches[i].popleft()
+                tx_pending[i].append(tx_pool.submit(_tx_call_guarded, batch))
+
+        def _drain_tx(i: int) -> None:
+            """Move completed transform futures into cooked non-blockingly."""
+            while tx_pending[i] and tx_pending[i][0].done():
+                cooked[i].extend(tx_pending[i].popleft().result())
+
+        # ── Combined advance ──────────────────────────────────────────────────
+
+        def _advance(i: int) -> None:
+            """Non-blocking pipeline pump for split i."""
+            _drain_io(i)
+            _drain_tx(i)
+            _try_submit_tx(i)
+            _fill_io(i)
+
+        def _ensure_cooked(i: int) -> None:
+            """Ensure cooked[i] has at least one row, blocking if necessary."""
+            _advance(i)
+            while not cooked[i]:
+                if tx_pending[i]:
+                    # Wait for the oldest in-flight transform.
+                    cooked[i].extend(tx_pending[i].popleft().result())
+                    _advance(i)
+                elif raw_batches[i]:
+                    # Acquire a transform slot (may block briefly if all
+                    # cpu_workers are busy with other splits).
+                    tx_semaphore.acquire()
+                    batch = raw_batches[i].popleft()
+                    tx_pending[i].append(tx_pool.submit(_tx_call_guarded, batch))
+                elif io_pending[i]:
+                    # Block on the oldest in-flight I/O fetch.
+                    raw_batches[i].append(io_pending[i].popleft().result())
+                    _advance(i)
+                else:
+                    break  # split exhausted
+
+        # ── Main loop ─────────────────────────────────────────────────────────
+
+        with ThreadPoolExecutor(max_workers=n * max_prefetch) as io_pool:
+            with ThreadPoolExecutor(max_workers=cpu_workers) as tx_pool:
+                self._raw_batches_ref = raw_batches
+                self._cooked_ref = cooked
+                self._fetch_head_ref = fetch_head
+                self._split_sizes_ref = split_sizes
+                self._local_consumed_ref = local_consumed
+                try:
+                    for i in range(n):
+                        _fill_io(i)
+
+                    while True:
+                        # Stop when any split is exhausted (all exhaust
+                        # simultaneously: equal split sizes + round-robin).
+                        if any(local_consumed[i] >= split_sizes[i] for i in range(n)):
+                            break
+
+                        for i in range(n):
+                            _ensure_cooked(i)
+                            row = cooked[i].popleft()
+                            local_consumed[i] += 1
+                            _advance(i)
+
+                            # After the last split in each cycle: update the
+                            # global offset and refresh the shared-memory stats
+                            # so the main process can observe pipeline depth
+                            # even when __iter__ runs in a worker process.
+                            if i == n - 1:
+                                self._resume_offset = initial_offset + local_consumed[i]
+                                ws = self._worker_stats
+                                ws[0] = sum(
+                                    split_sizes[j] - fetch_head[j] for j in range(n)
+                                )
+                                ws[1] = sum(
+                                    batch.num_rows for q in raw_batches for batch in q
+                                )
+                                ws[2] = sum(len(q) for q in cooked)
+                                ws[3] = sum(local_consumed)
+                                ws[4] = self._bytes_loaded
+                                ws[5] = int(self._fetch_time * 1_000_000)
+                                ws[6] = int(self._transform_time * 1_000_000)
+
+                            yield row
+                finally:
+                    self._raw_batches_ref = None
+                    self._cooked_ref = None
+                    self._fetch_head_ref = None
+                    self._split_sizes_ref = None
+                    self._local_consumed_ref = None
+
+    @property
+    def bytes_loaded(self) -> int:
+        """Cumulative bytes of raw Arrow buffer data fetched from storage.
+
+        Measured on the ``RecordBatch`` before any transform is applied, so
+        the value reflects actual I/O rather than the size of transformed
+        output.  Accumulates across multiple iterations of the same dataset
+        instance and is never reset automatically.
+        """
+        if self._raw_batches_ref is not None:
+            return self._bytes_loaded
+        return int(self._worker_stats[4])
+
+    @property
+    def fetch_time(self) -> float:
+        """Cumulative seconds spent waiting for data from LanceDB.
+
+        Measured per batch in the Stage 1 I/O threads as the total elapsed
+        time of the ``take_offsets`` call.  Accumulates across all splits and
+        all iterations.
+        """
+        if self._raw_batches_ref is not None:
+            return self._fetch_time
+        return self._worker_stats[5] / 1_000_000
+
+    @property
+    def transform_time(self) -> float:
+        """Cumulative seconds spent applying the transform.
+
+        Measured per batch in the Stage 2 transform threads as the elapsed
+        time inside the transform callable (or the default ``arrow2python``
+        conversion when no transform is set).  Accumulates across all splits
+        and all iterations.
+        """
+        if self._raw_batches_ref is not None:
+            return self._transform_time
+        return self._worker_stats[6] / 1_000_000
+
+    @property
+    def raw_queue_depth(self) -> int:
+        """Number of raw rows waiting for a transform thread across all splits.
+
+        A persistently non-zero value means Stage 2 (transform) is the
+        bottleneck: I/O is completing faster than transforms can consume
+        batches.  Returns 0 when not iterating.
+        """
+        if self._raw_batches_ref is not None:
+            return sum(batch.num_rows for q in self._raw_batches_ref for batch in q)
+        return int(self._worker_stats[1])
+
+    @property
+    def prefetch_queue_depth(self) -> int:
+        """Number of rows transformed and ready to yield across all splits.
+
+        Counts rows whose transform has completed and are sitting in memory
+        waiting for the main thread — rows that can be handed off with no
+        I/O or CPU wait.  Returns 0 when not iterating.
+        """
+        if self._cooked_ref is not None:
+            return sum(len(q) for q in self._cooked_ref)
+        return int(self._worker_stats[2])
+
+    @property
+    def unscanned_rows(self) -> int:
+        """Number of rows not yet submitted to the I/O stage across all splits.
+
+        Decreases as the I/O stage submits fetch requests.  When this reaches
+        zero all data has been requested from storage (though it may not have
+        arrived yet).  Returns 0 when not iterating.
+        """
+        if self._fetch_head_ref is not None:
+            return sum(
+                size - head
+                for size, head in zip(self._split_sizes_ref, self._fetch_head_ref)
+            )
+        return int(self._worker_stats[0])
+
+    @property
+    def consumed_rows(self) -> int:
+        """Number of rows already yielded to the caller across all splits.
+
+        Monotonically increases throughout iteration.  Returns 0 when not
+        iterating.
+        """
+        if self._local_consumed_ref is not None:
+            return sum(self._local_consumed_ref)
+        return int(self._worker_stats[3])
+
+    def __getstate__(self):
+        """Support pickling for multi-worker DataLoader (forkserver / spawn).
+
+        The live LanceDB table object contains non-picklable connection state
+        (sockets, Rust-backed PyO3 objects).  If a ``connection_factory`` was
+        supplied only the table name is serialised; the factory is called in
+        the worker to reopen the connection without embedding any credentials.
+        Without a factory the table's own picklable reopen state is captured
+        via ``_table_to_pickle_state`` (mirrors the ``Permutation`` approach).
+        """
+        state = self.__dict__.copy()
+        # _table: replace with reconnect info (credentials must not be embedded).
+        state["_table_name"] = self._table.name
+        if self._connection_factory is not None:
+            state["_table"] = None
+        else:
+            state["_table"] = _table_to_pickle_state(self._table)
+        # _perm_table: always in-memory; serialise as Arrow data (mirrors
+        # how Permutation.__getstate__ handles its permutation_table).
+        state["_perm_table"] = (
+            self._perm_table.name,
+            self._perm_table.to_arrow(),
+        )
+        for key in (
+            "_raw_batches_ref",
+            "_cooked_ref",
+            "_fetch_head_ref",
+            "_split_sizes_ref",
+            "_local_consumed_ref",
+        ):
+            state[key] = None
+        return state
+
+    def __setstate__(self, state):
+        """Reconnect to LanceDB after unpickling in a worker process."""
+        from . import connect as _connect
+
+        table_name = state.pop("_table_name")
+        table_state = state.pop("_table")
+        perm_name, perm_data = state.pop("_perm_table")
+        self.__dict__.update(state)
+        if self._connection_factory is not None:
+            self._table = self._connection_factory(table_name)
+        else:
+            self._table = _table_from_pickle_state(table_state)
+        self._perm_table = _connect("memory://").create_table(perm_name, perm_data)
+
+    def state_dict(self) -> dict:
+        """Snapshot the dataset's consumption state.
+
+        The returned dict is topology-independent: at global step boundaries
+        every split has been consumed the same number of times (by the
+        round-robin design), so the per-split count is a single uniform value
+        that is identical across all ranks and DataLoader workers.
+        """
+        return {
+            "shuffle_seed": self._shuffle_seed,
+            "num_splits": self._num_splits,
+            "epoch": self._epoch,
+            "samples_consumed_per_split": [self._resume_offset] * self._num_splits,
+        }
+
+    def load_state_dict(self, state: dict) -> None:
+        """Resume from a previously snapshotted state.
+
+        Raises ``ValueError`` if ``num_splits`` or ``shuffle_seed`` differ
+        from the checkpoint, since a different split structure or shuffle order
+        makes mid-epoch resumption meaningless.
+        """
+        if state["num_splits"] != self._num_splits:
+            raise ValueError(
+                f"num_splits mismatch: checkpoint has {state['num_splits']}, "
+                f"current dataset has {self._num_splits}"
+            )
+        if state["shuffle_seed"] != self._shuffle_seed:
+            raise ValueError(
+                f"shuffle_seed mismatch: checkpoint has {state['shuffle_seed']}, "
+                f"current dataset has {self._shuffle_seed}"
+            )
+        consumed = state["samples_consumed_per_split"]
+        # All entries are equal at step boundaries; use the first.
+        if isinstance(consumed, list):
+            self._resume_offset = consumed[0] if consumed else 0
+        else:
+            self._resume_offset = int(consumed)
@@ -651,6 +651,16 @@ def _append_vector_columns(
                    col_data = func.compute_source_embeddings_with_retry(
                        batch[conf.source_column]
                    )
+                    # Replace vectors with wrong length (including empty lists
+                    # returned for inputs like empty strings) with None so that
+                    # _handle_bad_vectors can process them according to the
+                    # on_bad_vectors policy instead of crashing when PyArrow
+                    # tries to cast them into a fixed-size list array.
+                    expected_ndims = conf.function.ndims()
+                    col_data = [
+                        v if v is not None and len(v) == expected_ndims else None
+                        for v in col_data
+                    ]
                    if no_vector_column:
                        batch = batch.append_column(
                            schema.field(vector_column),
@@ -2142,12 +2152,19 @@ class LanceTable(Table):

        branch = self.current_branch()
        version = None if branch is not None else self.version
-        if self._namespace_client is not None:
+        namespace_client = self._namespace_client
+        if namespace_client is None:
+            conn_uri = getattr(self._conn, "uri", "")
+            if get_uri_scheme(conn_uri) == "namespace":
+                namespace_client = self._conn.namespace_client()
+                self._namespace_client = namespace_client
+
+        if namespace_client is not None:
            table_id = self._namespace_path + [self.name]
            ds = lance.dataset(
                version=version,
                storage_options=self._conn.storage_options,
-                namespace_client=self._namespace_client,
+                namespace_client=namespace_client,
                table_id=table_id,
                **kwargs,
            )
@@ -3742,6 +3759,11 @@ class LanceTable(Table):
        [`AsyncTable.unset_lsm_write_spec`][lancedb.AsyncTable.unset_lsm_write_spec]."""
        return LOOP.run(self._table.unset_lsm_write_spec())

+    def get_lsm_write_spec(self) -> Optional["LsmWriteSpec"]:
+        """Read the installed LsmWriteSpec, or ``None``. See
+        [`AsyncTable.get_lsm_write_spec`][lancedb.AsyncTable.get_lsm_write_spec]."""
+        return LOOP.run(self._table.get_lsm_write_spec())
+
    def close_lsm_writers(self) -> None:
        """Close cached MemWAL shard writers. See
        [`AsyncTable.close_lsm_writers`][lancedb.AsyncTable.close_lsm_writers]."""
@@ -4013,7 +4035,16 @@ def _handle_bad_vector_column(
        dim = _infer_vector_dim(vec_arr)
        if dim is None:
            return data
-    has_wrong_dim = pc.not_equal(pc.list_value_length(vec_arr), dim)
+
+    is_null = pc.is_null(vec_arr)
+    # pc.list_value_length returns null for null list entries, so
+    # pc.not_equal(null, dim) also returns null. Use or_kleene so that
+    # True OR null = True (Kleene three-valued logic), ensuring null vectors
+    # are counted as wrong-dim.
+    has_wrong_dim = pc.or_kleene(
+        is_null,
+        pc.not_equal(pc.list_value_length(vec_arr), dim),
+    )

    has_bad_vectors = pc.any(has_nan).as_py() or pc.any(has_wrong_dim).as_py()

@@ -4391,6 +4422,17 @@ class AsyncTable:
        """
        await self._inner.unset_lsm_write_spec()

+    async def get_lsm_write_spec(self) -> Optional["LsmWriteSpec"]:
+        """Read the LsmWriteSpec currently installed on this table.
+
+        Returns ``None`` when the MemWAL LSM write path is not enabled (no
+        spec has been set, or it was removed with `unset_lsm_write_spec`).
+        The returned spec — including its ``maintained_indexes`` and
+        ``writer_config_defaults`` — mirrors what was passed to
+        `set_lsm_write_spec`.
+        """
+        return await self._inner.get_lsm_write_spec()
+
    async def close_lsm_writers(self) -> None:
        """Drain and close any cached MemWAL shard writers for this table.

@@ -177,7 +177,10 @@ def flatten_columns(tbl: pa.Table, flatten: Optional[Union[int, bool]] = None):
                continue
            else:
                break
-    elif isinstance(flatten, int):
+    # `bool` is a subclass of `int`, so guard against it explicitly: `flatten=False`
+    # (and `None`) must mean "do not flatten" rather than falling into the integer
+    # branch and raising on the `flatten <= 0` check.
+    elif isinstance(flatten, int) and not isinstance(flatten, bool):
        if flatten <= 0:
            raise ValueError(
                "Please specify a positive integer for flatten or the boolean "
@@ -515,3 +518,38 @@ def batch_to_tensor_rows(batch: pa.RecordBatch):
    stacked = torch.tensor(numpy.column_stack(columns))
    rows = list(stacked.unbind(dim=0))
    return rows
+
+
+def batch_to_tensor_dict(batch: pa.RecordBatch):
+    """
+    Convert a PyArrow RecordBatch to a list of per-row dicts of PyTorch Tensors.
+
+    Each column is first converted to a 1-D tensor (zero-copy via DLPack), then
+    sliced per row. The result is a list whose length is ``batch.num_rows`` and
+    whose items are dicts keyed by column name. This shape composes directly
+    with PyTorch's default ``DataLoader`` collate, which stacks the per-row
+    dicts back into a dict of batched tensors.
+
+    Fails if torch is not installed.
+    Fails if a column's data type is not supported by PyTorch.
+
+    Parameters
+    ----------
+    batch : pa.RecordBatch
+        The record batch to convert.
+
+    Returns
+    -------
+    list[dict[str, torch.Tensor]]
+        One per-row dict per row in the batch. Each dict maps column name to a
+        0-D tensor view into the column.
+    """
+    torch = attempt_import_or_raise("torch", "torch")
+    tensors = {
+        name: torch.from_dlpack(col)
+        for name, col in zip(batch.schema.names, batch.columns)
+    }
+    return [
+        {name: tensor[i] for name, tensor in tensors.items()}
+        for i in range(batch.num_rows)
+    ]
@@ -2,6 +2,7 @@
 # SPDX-FileCopyrightText: Copyright The LanceDB Authors

 import os
+import pickle
 from typing import List, Optional, Union
 from unittest.mock import MagicMock, patch

@@ -242,6 +243,49 @@ def test_embedding_with_bad_results(tmp_path):
    assert tbl["vector"].null_count == 1


+def test_embedding_with_empty_output_vectors(tmp_path):
+    """Regression test for issue #1672.
+
+    When an embedding function returns an empty list (e.g. for empty-string
+    inputs), _append_vector_columns used to crash because PyArrow cannot cast
+    [] into a fixed-size list element.  The fix replaces wrong-length vectors
+    with None before building the Arrow array so that _handle_bad_vectors can
+    process them normally.
+    """
+
+    @register("empty-vec-embedding")
+    class EmptyVecEmbeddingFunction(TextEmbeddingFunction):
+        def ndims(self):
+            return 128
+
+        def generate_embeddings(self, texts: Union[List[str], np.ndarray]) -> list:
+            # Simulate a model that returns an empty list for blank inputs
+            return [
+                [] if text.strip() == "" else np.random.randn(self.ndims()).tolist()
+                for text in texts
+            ]
+
+    db = lancedb.connect(tmp_path)
+    registry = EmbeddingFunctionRegistry.get_instance()
+    model = registry.get("empty-vec-embedding").create()
+
+    class Schema(LanceModel):
+        text: str = model.SourceField()
+        vector: Vector(model.ndims()) = model.VectorField()
+
+    table = db.create_table("test_empty_vec", schema=Schema, mode="overwrite")
+
+    # Should not crash; the row with the empty string should be dropped
+    table.add(
+        [{"text": "hello world"}, {"text": ""}, {"text": "foo"}],
+        on_bad_vectors="drop",
+    )
+
+    assert len(table) == 2
+    texts = table.to_arrow()["text"].to_pylist()
+    assert "" not in texts
+
+
 def test_with_existing_vectors(tmp_path):
    @register("mock-embedding")
    class MockEmbeddingFunction(TextEmbeddingFunction):
@@ -548,6 +592,22 @@ def test_openai_no_retry_on_401(mock_sleep):
    assert mock_sleep.call_count == 0


+def test_ollama_embeddings_pickle():
+    """OllamaEmbeddings must pickle even after the cached client is created."""
+    registry = get_registry()
+    model = registry.get("ollama").create(name="nomic-embed-text")
+
+    # Simulate accessing the cached client, which stores it on the instance.
+    model.__dict__["_ollama_client"] = MagicMock()
+
+    pickled = pickle.dumps(model)
+    restored = pickle.loads(pickled)
+
+    assert restored.name == "nomic-embed-text"
+    assert restored.host == "http://localhost:11434"
+    assert "_ollama_client" not in restored.__dict__
+
+
 def test_url_retrieve_downloads_image():
    """
    Embedding functions like open-clip, siglip, and jinaai use url_retrieve()
@@ -3,10 +3,14 @@

 """Tests for the type-safe expression builder API."""

-import pytest
+from datetime import date, datetime, timedelta, timezone
+from decimal import Decimal
+
 import pyarrow as pa
+import pytest
+
 import lancedb
-from lancedb.expr import Expr, col, lit, func
+from lancedb.expr import Expr, col, func, lit


 # ── unit tests for Expr construction ─────────────────────────────────────────
@@ -54,6 +58,28 @@ class TestExprConstruction:
        with pytest.raises(Exception):
            func("not_a_real_function", col("x"))

+    def test_lit_date(self):
+        e = lit(date(2024, 1, 1))
+        assert isinstance(e, Expr)
+
+    def test_lit_datetime(self):
+        # Naive datetime
+        e = lit(datetime(2024, 1, 1, 10, 0))
+        assert isinstance(e, Expr)
+
+    def test_lit_datetime_tz(self):
+        # Timezone-aware datetime
+        tz = timezone(timedelta(hours=5))
+        dt = datetime(2024, 1, 1, 10, 0, tzinfo=tz)
+        e = lit(dt)
+        assert isinstance(e, Expr)
+
+    def test_lit_decimal_precision(self):
+        # High precision Decimal that would be rounded if converted to float
+        d = Decimal("1.234567890123456789")
+        e = lit(d)
+        assert isinstance(e, Expr)
+

 class TestExprOperators:
    def test_eq_operator(self):
@@ -142,6 +168,20 @@ class TestExprOperators:
        assert isinstance(e, Expr)
        assert e.to_sql() == "(name = 'alice')"

+    def test_reflexive_comparisons(self):
+        # 10 < col("age") swaps to col("age") > 10
+        assert (10 < col("age")).to_sql() == "(age > 10)"
+        assert (10 <= col("age")).to_sql() == "(age >= 10)"
+        assert (10 > col("age")).to_sql() == "(age < 10)"
+        assert (10 >= col("age")).to_sql() == "(age <= 10)"
+        assert (10 == col("age")).to_sql() == "(age = 10)"
+        assert (10 != col("age")).to_sql() == "(age <> 10)"
+
+    def test_reflexive_logical(self):
+        # True & Expr calls Expr.__rand__(True)
+        assert (True & (col("age") > 18)).to_sql() == "(true AND (age > 18))"
+        assert (False | (col("age") > 18)).to_sql() == "(false OR (age > 18))"
+

 class TestExprBytesLiteral:
    def test_bytes_to_sql(self):
@@ -282,6 +322,40 @@ class TestExprRepr:
            {e: 1}


+class TestExprReflexive:
+    def test_reflexive_eq(self):
+        e = 1 == col("x")
+        assert isinstance(e, Expr)
+        assert e.to_sql() == "(x = 1)"
+
+    def test_reflexive_ne(self):
+        e = 1 != col("x")
+        assert isinstance(e, Expr)
+        assert e.to_sql() == "(x <> 1)"
+
+    def test_reflexive_lt(self):
+        # 1 < x  =>  (x > 1)
+        e = 1 < col("x")
+        assert isinstance(e, Expr)
+        assert e.to_sql() == "(x > 1)"
+
+    def test_reflexive_gt(self):
+        # 1 > x  =>  (x < 1)
+        e = 1 > col("x")
+        assert isinstance(e, Expr)
+        assert e.to_sql() == "(x < 1)"
+
+    def test_reflexive_and(self):
+        e = True & col("active")
+        assert isinstance(e, Expr)
+        assert e.to_sql() == "(true AND active)"
+
+    def test_reflexive_or(self):
+        e = False | col("inactive")
+        assert isinstance(e, Expr)
+        assert e.to_sql() == "(false OR inactive)"
+
+
 # ── integration tests: end-to-end query against a real table ─────────────────


@@ -432,6 +506,72 @@ class TestColNamingIntegration:
        assert sorted(result["upper_name"].to_pylist()) == ["ALICE", "BOB", "CHARLIE"]


+@pytest.fixture
+def type_check_table(tmp_path):
+    """Fixture that creates a table with Date32 and Decimal128 columns."""
+    db = lancedb.connect(str(tmp_path))
+    schema = pa.schema(
+        [
+            ("date", pa.date32()),
+            ("decimal", pa.decimal128(10, 2)),
+            ("binary", pa.binary()),
+        ]
+    )
+    data = pa.table(
+        {
+            "date": [date(2024, 1, 1), date(2024, 1, 2)],
+            "decimal": [Decimal("10.50"), Decimal("20.75")],
+            "binary": [b"\x01", b"\x02"],
+        },
+        schema=schema,
+    )
+    return db.create_table("extended_types", data)
+
+
+class TestExtendedTypeIntegration:
+    """Integration tests verifying that typed literals work correctly in filters."""
+
+    def test_date_integration(self, type_check_table):
+        """Verify that Date32 literals are correctly parsed and filtered."""
+        result = (
+            type_check_table.search()
+            .where(col("date") == lit(date(2024, 1, 1)))
+            .to_arrow()
+        )
+        assert result.num_rows == 1
+        assert result["date"][0].as_py() == date(2024, 1, 1)
+
+    def test_decimal_integration(self, tmp_path):
+        """A Decimal literal must retain full 128-bit precision in a filter.
+
+        1.234567890123456789 and 1.234567890123456790 differ only in the last
+        digit and are indistinguishable once truncated to f64.  The filter
+        therefore returns the single expected row only if ``lit(Decimal)``
+        produces a true ``Decimal128`` scalar rather than being coerced to f64.
+        """
+        low = Decimal("1.234567890123456789")
+        high = Decimal("1.234567890123456790")
+
+        db = lancedb.connect(str(tmp_path / "decimal_precision"))
+        schema = pa.schema([("val", pa.decimal128(19, 18))])
+        table = db.create_table(
+            "decimal_precision",
+            pa.table({"val": [low, high]}, schema=schema),
+        )
+
+        result = table.search().where(col("val") < lit(high)).to_arrow()
+        assert result.num_rows == 1
+        assert result["val"][0].as_py() == low
+
+    def test_binary_integration(self, type_check_table):
+        """Verify that Binary literals are correctly filtered."""
+        result = (
+            type_check_table.search().where(col("binary") == lit(b"\x01")).to_arrow()
+        )
+        assert result.num_rows == 1
+        assert result["binary"][0].as_py() == b"\x01"
+
+
 # ── bytes / binary column integration tests ───────────────────────────────────


@@ -11,6 +11,7 @@ import lancedb
 import pyarrow as pa
 import pytest
 from lancedb._lancedb import LsmWriteSpec
+from lancedb.index import BTree

 SCHEMA = pa.schema(
    [
@@ -136,3 +137,76 @@ def test_lsm_write_spec_identity_and_writer_config_defaults():
    s = s.with_writer_config_defaults({"durable_write": "false"})
    assert s.writer_config_defaults == {"durable_write": "false"}
    assert "durable_write" in repr(s)
+
+
+def test_get_lsm_write_spec(tmp_path):
+    _db, table = _make_table(tmp_path)
+    table.set_unenforced_primary_key("id")
+
+    # None when nothing is installed.
+    assert table.get_lsm_write_spec() is None
+
+    # A real scalar index is needed to name it as a maintained index.
+    table.create_index("id", config=BTree())
+    idx_name = table.list_indices()[0].name
+
+    # Bucket spec round-trips, including maintained indexes and writer config
+    # defaults.
+    table.set_lsm_write_spec(
+        LsmWriteSpec.bucket("id", 4)
+        .with_maintained_indexes([idx_name])
+        .with_writer_config_defaults({"durable_write": "false"})
+    )
+    spec = table.get_lsm_write_spec()
+    assert spec is not None
+    assert spec.spec_type == "bucket"
+    assert spec.column == "id"
+    assert spec.num_buckets == 4
+    assert spec.maintained_indexes == [idx_name]
+    assert spec.writer_config_defaults == {"durable_write": "false"}
+
+    # After unset, None again.
+    table.unset_lsm_write_spec()
+    assert table.get_lsm_write_spec() is None
+
+    # Identity round-trips (column recovered from the schema).
+    table.set_lsm_write_spec(LsmWriteSpec.identity("id"))
+    spec = table.get_lsm_write_spec()
+    assert spec.spec_type == "identity"
+    assert spec.column == "id"
+    table.unset_lsm_write_spec()
+
+    # Unsharded round-trips (no routing column).
+    table.set_lsm_write_spec(LsmWriteSpec.unsharded())
+    spec = table.get_lsm_write_spec()
+    assert spec.spec_type == "unsharded"
+    assert spec.column is None
+
+
+@pytest.mark.asyncio
+async def test_async_get_lsm_write_spec(tmp_path):
+    db = await lancedb.connect_async(
+        tmp_path, read_consistency_interval=timedelta(seconds=0)
+    )
+    table = await db.create_table(
+        "t",
+        pa.RecordBatchReader.from_batches(SCHEMA, [_batch(["seed"], [0])]),
+    )
+
+    assert await table.get_lsm_write_spec() is None
+
+    # A real scalar index is needed to name it as a maintained index.
+    await table.create_index("id", config=BTree())
+    idx_name = (await table.list_indices())[0].name
+
+    await table.set_lsm_write_spec(
+        LsmWriteSpec.bucket("id", 8).with_maintained_indexes([idx_name])
+    )
+    spec = await table.get_lsm_write_spec()
+    assert spec is not None
+    assert spec.spec_type == "bucket"
+    assert spec.column == "id"
+    assert spec.num_buckets == 8
+    assert spec.maintained_indexes == [idx_name]
+    await table.unset_lsm_write_spec()
+    assert await table.get_lsm_write_spec() is None
@@ -5,6 +5,7 @@

 import tempfile
 import shutil
+import importlib
 import pytest
 import pyarrow as pa
 import lancedb
@@ -103,6 +104,40 @@ class TestNamespaceConnection:
        assert isinstance(db, lancedb.LanceNamespaceDBConnection)
        assert len(list(db.table_names())) == 0

+    def test_sync_builtin_namespace_uses_rust_without_python_client(self, monkeypatch):
+        """Built-in sync namespace connections should not construct or call the
+        Python namespace client for normal namespace/table management."""
+        namespace_module = importlib.import_module("lancedb.namespace")
+
+        def fail_namespace_connect(*args, **kwargs):
+            raise AssertionError("Python namespace client should not be constructed")
+
+        monkeypatch.setattr(
+            namespace_module, "namespace_connect", fail_namespace_connect
+        )
+
+        db = lancedb.connect_namespace("dir", {"root": self.temp_dir})
+        assert isinstance(db, lancedb.LanceNamespaceDBConnection)
+        assert db._namespace_client is None
+        assert db._route_pushdown_to_rust is True
+
+        db.create_namespace(["test_ns"])
+        assert "test_ns" in db.list_namespaces().namespaces
+
+        schema = pa.schema([pa.field("id", pa.int64())])
+        table = db.create_table("test_table", schema=schema, namespace_path=["test_ns"])
+        assert table.namespace == ["test_ns"]
+        assert "test_table" in db.table_names(namespace_path=["test_ns"])
+        assert "test_table" in db.list_tables(namespace_path=["test_ns"]).tables
+
+        opened = db.open_table("test_table", namespace_path=["test_ns"])
+        assert opened.namespace == ["test_ns"]
+
+        db.drop_table("test_table", namespace_path=["test_ns"])
+        assert db.list_tables(namespace_path=["test_ns"]).tables == []
+        db.drop_namespace(["test_ns"])
+        assert "test_ns" not in db.list_namespaces().namespaces
+
    def test_create_table_through_namespace(self):
        """Test creating a table through namespace."""
        db = lancedb.connect_namespace("dir", {"root": self.temp_dir})
@@ -564,6 +599,61 @@ class TestAsyncNamespaceConnection:
        table_names = await db.table_names()
        assert len(list(table_names)) == 0

+    async def test_async_builtin_namespace_uses_rust_without_python_client(
+        self, monkeypatch
+    ):
+        """Built-in async namespace connections should not construct or call the
+        Python namespace client for normal namespace/table management."""
+        namespace_module = importlib.import_module("lancedb.namespace")
+
+        def fail_namespace_connect(*args, **kwargs):
+            raise AssertionError("Python namespace client should not be constructed")
+
+        monkeypatch.setattr(
+            namespace_module, "namespace_connect", fail_namespace_connect
+        )
+
+        db = lancedb.connect_namespace_async("dir", {"root": self.temp_dir})
+        assert isinstance(db, lancedb.AsyncLanceNamespaceDBConnection)
+        assert db._namespace_client is None
+        assert db._route_pushdown_to_rust is True
+
+        await db.create_namespace(["test_ns"])
+        assert "test_ns" in (await db.list_namespaces()).namespaces
+
+        schema = pa.schema([pa.field("id", pa.int64())])
+        table = await db.create_table(
+            "test_table", schema=schema, namespace_path=["test_ns"]
+        )
+        assert table._namespace_path == ["test_ns"]
+        assert table._namespace_client is None
+        assert table._route_pushdown_to_rust is True
+        assert "test_table" in await db.table_names(namespace_path=["test_ns"])
+        assert "test_table" in (await db.list_tables(namespace_path=["test_ns"])).tables
+
+        opened = await db.open_table("test_table", namespace_path=["test_ns"])
+        assert opened._namespace_path == ["test_ns"]
+
+        await db.drop_table("test_table", namespace_path=["test_ns"])
+        assert (await db.list_tables(namespace_path=["test_ns"])).tables == []
+        await db.drop_namespace(["test_ns"])
+        assert "test_ns" not in (await db.list_namespaces()).namespaces
+
+    async def test_async_namespace_client_is_lazy(self):
+        """namespace_client() should still return the backing client on demand."""
+        pytest.importorskip("lance")
+        from lance.namespace import DirectoryNamespace
+
+        db = lancedb.connect_namespace_async("dir", {"root": self.temp_dir})
+        assert db._namespace_client is None
+
+        ns_client = await db.namespace_client()
+
+        assert isinstance(ns_client, DirectoryNamespace)
+        namespace_id = ns_client.namespace_id().replace("\\\\", "\\")
+        assert str(self.temp_dir) in namespace_id
+        assert db._namespace_client is ns_client
+
    # Async connect via namespace helper is not enabled yet.

    async def test_create_table_async(self):
@@ -818,10 +908,11 @@ class TestPushdownOperations:
        )
        assert db._route_pushdown_to_rust is True

-    def test_route_pushdown_to_rust_false_for_dir(self):
-        """A non-native (dir) connection keeps the Python pushdown path."""
+    def test_route_pushdown_to_rust_for_native_dir(self):
+        """The sync dir connection is natively built and defers QueryTable
+        pushdown to Rust."""
        db = lancedb.connect_namespace("dir", {"root": self.temp_dir})
-        assert db._route_pushdown_to_rust is False
+        assert db._route_pushdown_to_rust is True

    def test_async_route_pushdown_to_rust_for_native_rest(self):
        """The async connection must not silently bypass the read-freshness fix:
@@ -834,10 +925,11 @@ class TestPushdownOperations:
        )
        assert db._route_pushdown_to_rust is True

-    def test_async_route_pushdown_to_rust_false_for_dir(self):
-        """The async non-native (dir) connection keeps the Python pushdown path."""
+    def test_async_route_pushdown_to_rust_for_native_dir(self):
+        """The async dir connection is natively built and defers QueryTable
+        pushdown to Rust."""
        db = lancedb.connect_namespace_async("dir", {"root": self.temp_dir})
-        assert db._route_pushdown_to_rust is False
+        assert db._route_pushdown_to_rust is True

    def test_lance_table_to_arrow_uses_query_pushdown(self):
        namespace_client = _NamespaceClient()
@@ -935,14 +935,41 @@ def test_transform_fn(mem_db):
    try:
        import torch

-        torch_result = list(
-            permutation.with_format("torch").iter(10, skip_last_batch=False)
+        # "torch" returns a list of per-row dicts. Default DataLoader collate
+        # stacks the per-row dicts back into a dict of batched tensors.
+        torch_perm = permutation.with_format("torch")
+        torch_batch = list(torch_perm.iter(10, skip_last_batch=False))[0]
+        assert isinstance(torch_batch, list)
+        assert len(torch_batch) == 10
+        assert isinstance(torch_batch[0], dict)
+        assert set(torch_batch[0].keys()) == {"id", "value"}
+        assert isinstance(torch_batch[0]["id"], torch.Tensor)
+        assert torch_batch[0]["id"].dtype == torch.int64
+
+        rows = torch_perm.__getitems__([0, 1, 2])
+        assert isinstance(rows, list)
+        assert len(rows) == 3
+        assert isinstance(rows[0], dict)
+        assert set(rows[0].keys()) == {"id", "value"}
+        assert isinstance(rows[0]["id"], torch.Tensor)
+
+        # "torch_row" returns a list of tensors, one per row.
+        torch_rows = list(
+            permutation.with_format("torch_row").iter(10, skip_last_batch=False)
        )[0]
-        assert isinstance(torch_result, list)
-        assert len(torch_result) == 10
-        assert isinstance(torch_result[0], torch.Tensor)
-        assert torch_result[0].shape == (2,)
-        assert torch_result[0].dtype == torch.int64
+        assert isinstance(torch_rows, list)
+        assert len(torch_rows) == 10
+        assert isinstance(torch_rows[0], torch.Tensor)
+        assert torch_rows[0].shape == (2,)
+        assert torch_rows[0].dtype == torch.int64
+
+        # "torch_col" stacks columns into a single 2D tensor.
+        torch_col = list(
+            permutation.with_format("torch_col").iter(10, skip_last_batch=False)
+        )[0]
+        assert isinstance(torch_col, torch.Tensor)
+        assert torch_col.shape == (2, 10)
+        assert torch_col.dtype == torch.int64
    except ImportError:
        # Skip check if torch is not installed
        pass
@@ -502,6 +502,61 @@ def test_with_row_id(table: lancedb.table.Table):
    assert rs["_rowid"].to_pylist() == [0, 1]


+def test_where_repeated_combines_with_and(table: lancedb.table.Table):
+    # Calling where() more than once should AND the filters together instead of
+    # silently replacing the previous one (regression test for #2649).
+    builder = table.search().where("id >= 1").where("id < 2")
+    assert builder._where == "(id >= 1) AND (id < 2)"
+
+    ids = [row["id"] for row in builder.limit(10).to_list()]
+    assert ids == [1]
+
+
+def test_where_repeated_combines_expr(table: lancedb.table.Table):
+    from lancedb.expr import col, lit
+
+    builder = table.search().where(col("id") >= lit(1)).where(col("id") < lit(2))
+    ids = [row["id"] for row in builder.limit(10).to_list()]
+    assert ids == [1]
+
+
+def test_where_mixed_filter_kinds_combines(table: lancedb.table.Table):
+    # Mixing a SQL string filter with an expression filter lowers the
+    # expression to SQL and combines them as SQL strings.
+    from lancedb.expr import col, lit
+
+    builder = table.search().where("id >= 1").where(col("id") < lit(2))
+    ids = [row["id"] for row in builder.limit(10).to_list()]
+    assert ids == [1]
+
+
+@pytest.mark.asyncio
+async def test_where_repeated_combines_with_and_async(table_async: AsyncTable):
+    ids = [
+        row["id"]
+        for row in (
+            await table_async.query().where("id >= 1").where("id < 2").to_list()
+        )
+    ]
+    assert ids == [1]
+
+
+@pytest.mark.asyncio
+async def test_where_mixed_filter_kinds_combines_async(table_async: AsyncTable):
+    from lancedb.expr import col, lit
+
+    ids = [
+        row["id"]
+        for row in (
+            await table_async.query()
+            .where("id >= 1")
+            .where(col("id") < lit(2))
+            .to_list()
+        )
+    ]
+    assert ids == [1]
+
+
 def test_distance_range(table: lancedb.table.Table):
    q = [0, 0]
    rs = table.search(q).to_arrow()
@@ -412,10 +412,12 @@ def test_remote_permutation_is_picklable():
            content_len = int(request.headers.get("Content-Length"))
            body = json.loads(request.rfile.read(content_len))
            if "filter" in body:
-                match = re.search(r"_rowoffset in \((.*?)\)", body["filter"])
-                offsets = [int(offset.strip()) for offset in match.group(1).split(",")]
+                match = re.search(
+                    r"_rowoffset\s+in\s+\((.*?)\)", body["filter"], re.IGNORECASE
+                )
+                offsets = [int(o.strip()) for o in match.group(1).split(",")]
            else:
-                offsets = rows
+                offsets = list(range(len(rows)))
            table = pa.table({"a": [rows[offset] for offset in offsets]})

            request.send_response(200)
@@ -350,6 +350,38 @@ def test_mrr_reranker_empty_input():
        reranker.rerank_multivector([])


+def test_mrr_multivector_rewards_consensus():
+    # Reciprocal ranks must be averaged across *all* ranking systems, treating a
+    # missing system as 0. A document ranked first by every system must outrank a
+    # document ranked first by only one of them.
+    reranker = MRRReranker()
+
+    def ranking(row_ids):
+        return pa.table({"_rowid": pa.array(row_ids, type=pa.int64())})
+
+    # Doc 1 is rank 1 in only the first system; doc 2 is rank 1 in two systems
+    # and rank 2 in the third (strong cross-system consensus).
+    rs1 = ranking([1, 2, 3])
+    rs2 = ranking([2, 3, 4])
+    rs3 = ranking([2, 5, 6])
+
+    result = reranker.rerank_multivector([rs1, rs2, rs3])
+    scores = {
+        row_id: score
+        for row_id, score in zip(
+            result["_rowid"].to_pylist(),
+            result["_relevance_score"].to_pylist(),
+        )
+    }
+
+    # sum of reciprocal ranks / number of systems
+    assert scores[1] == pytest.approx(1.0 / 3)
+    assert scores[2] == pytest.approx((0.5 + 1.0 + 1.0) / 3)
+    assert scores[2] > scores[1]
+    # The consensus document ranks first overall.
+    assert result["_rowid"].to_pylist()[0] == 2
+
+
 def test_rrf_reranker_distance():
    data = pa.table(
        {
@@ -7,7 +7,8 @@ Tests for S3 bucket names containing dots.
 Related issue: https://github.com/lancedb/lancedb/issues/1898

 These tests validate the early error checking for S3 bucket names with dots.
-No actual S3 connection is made - validation happens before connection.
+When validation succeeds, eager namespace initialization may still fail later
+because these tests intentionally do not provide real S3 credentials.
 """

 import pytest
@@ -20,6 +21,13 @@ BUCKET_WITH_DOTS_AND_AWS_REGION = ("s3://my.bucket.name", {"aws_region": "us-eas
 BUCKET_WITHOUT_DOTS = "s3://my-bucket/path"


+def assert_not_rejected_for_bucket_dots(connect):
+    try:
+        connect()
+    except ValueError as err:
+        assert "contains dots" not in str(err)
+
+
 class TestS3BucketWithDotsSync:
    """Tests for connect()."""

@@ -27,19 +35,22 @@ class TestS3BucketWithDotsSync:
        with pytest.raises(ValueError, match="contains dots"):
            lancedb.connect(BUCKET_WITH_DOTS)

-    def test_bucket_with_dots_and_region_passes(self):
+    def test_bucket_with_dots_and_region_is_not_rejected(self):
        uri, opts = BUCKET_WITH_DOTS_AND_REGION
-        db = lancedb.connect(uri, storage_options=opts)
-        assert db is not None
+        assert_not_rejected_for_bucket_dots(
+            lambda: lancedb.connect(uri, storage_options=opts)
+        )

-    def test_bucket_with_dots_and_aws_region_passes(self):
+    def test_bucket_with_dots_and_aws_region_is_not_rejected(self):
        uri, opts = BUCKET_WITH_DOTS_AND_AWS_REGION
-        db = lancedb.connect(uri, storage_options=opts)
-        assert db is not None
+        assert_not_rejected_for_bucket_dots(
+            lambda: lancedb.connect(uri, storage_options=opts)
+        )

-    def test_bucket_without_dots_passes(self):
-        db = lancedb.connect(BUCKET_WITHOUT_DOTS)
-        assert db is not None
+    def test_bucket_without_dots_is_not_rejected(self):
+        assert_not_rejected_for_bucket_dots(
+            lambda: lancedb.connect(BUCKET_WITHOUT_DOTS)
+        )


 class TestS3BucketWithDotsAsync:
@@ -51,18 +62,24 @@ class TestS3BucketWithDotsAsync:
            await lancedb.connect_async(BUCKET_WITH_DOTS)

    @pytest.mark.asyncio
-    async def test_bucket_with_dots_and_region_passes(self):
+    async def test_bucket_with_dots_and_region_is_not_rejected(self):
        uri, opts = BUCKET_WITH_DOTS_AND_REGION
-        db = await lancedb.connect_async(uri, storage_options=opts)
-        assert db is not None
+        try:
+            await lancedb.connect_async(uri, storage_options=opts)
+        except ValueError as err:
+            assert "contains dots" not in str(err)

    @pytest.mark.asyncio
-    async def test_bucket_with_dots_and_aws_region_passes(self):
+    async def test_bucket_with_dots_and_aws_region_is_not_rejected(self):
        uri, opts = BUCKET_WITH_DOTS_AND_AWS_REGION
-        db = await lancedb.connect_async(uri, storage_options=opts)
-        assert db is not None
+        try:
+            await lancedb.connect_async(uri, storage_options=opts)
+        except ValueError as err:
+            assert "contains dots" not in str(err)

    @pytest.mark.asyncio
-    async def test_bucket_without_dots_passes(self):
-        db = await lancedb.connect_async(BUCKET_WITHOUT_DOTS)
-        assert db is not None
+    async def test_bucket_without_dots_is_not_rejected(self):
+        try:
+            await lancedb.connect_async(BUCKET_WITHOUT_DOTS)
+        except ValueError as err:
+            assert "contains dots" not in str(err)
@@ -1137,6 +1137,16 @@ def test_namespace_open_table_with_branch_version(tmp_path):
    assert db.open_table("t", namespace_path=["ns1"], branch="exp").count_rows() == 3


+def test_namespace_root_table_to_lance_uses_namespace_client(tmp_path):
+    pytest.importorskip("lance")  # "dir" impl is lance.namespace.DirectoryNamespace
+    db = lancedb.connect_namespace("dir", {"root": str(tmp_path)})
+    table = db.create_table("t", [{"i": 0}])
+
+    assert table._namespace_client is None
+    assert table.to_lance().count_rows() == 1
+    assert table._namespace_client is not None
+
+
@pytest.mark.asyncio
 async def test_async_namespace_open_table_with_branch_version(tmp_path):
    pytest.importorskip("lance")  # "dir" impl is lance.namespace.DirectoryNamespace
@@ -148,15 +148,47 @@ def test_permutation_dataloader(mem_db):
    for batch in dataloader:
        assert batch["a"].size(0) == 10

-    permutation = permutation.with_format("torch")
-    dataloader = torch.utils.data.DataLoader(permutation, batch_size=10, shuffle=True)
+    # "torch" produces a list of per-row dicts per batch. The default
+    # DataLoader collate stacks the per-row dicts back into a batched dict.
+    torch_perm = permutation.with_format("torch")
+    batch = next(torch_perm.iter(10, skip_last_batch=False))
+    assert isinstance(batch, list)
+    assert len(batch) == 10
+    assert isinstance(batch[0], dict)
+    assert isinstance(batch[0]["a"], torch.Tensor)
+    rows = torch_perm.__getitems__([0, 1, 2])
+    assert isinstance(rows, list)
+    assert len(rows) == 3
+    assert isinstance(rows[0], dict)
+    assert isinstance(rows[0]["a"], torch.Tensor)
+    dataloader = torch.utils.data.DataLoader(torch_perm, batch_size=10, shuffle=True)
+    for batch in dataloader:
+        assert isinstance(batch, dict)
+        assert batch["a"].shape == (10,)
+    # Spawn-based workers exercise the pickle round-trip path: the new
+    # transform-as-list shape must survive pickling so workers produce the
+    # same per-row dicts the parent does.
+    spawn_loader = torch.utils.data.DataLoader(
+        torch_perm,
+        batch_size=10,
+        num_workers=2,
+        multiprocessing_context="spawn",
+    )
+    for batch in spawn_loader:
+        assert isinstance(batch, dict)
+        assert batch["a"].shape == (10,)
+
+    # "torch_row" returns a list of row tensors. Works with the default
+    # DataLoader collate (stacks rows into 2D).
+    row_perm = permutation.with_format("torch_row")
+    dataloader = torch.utils.data.DataLoader(row_perm, batch_size=10, shuffle=True)
    for batch in dataloader:
        assert batch.size(0) == 10
        assert batch.size(1) == 1

-    permutation = permutation.with_format("torch_col")
+    col_perm = permutation.with_format("torch_col")
    dataloader = torch.utils.data.DataLoader(
-        permutation, collate_fn=lambda x: x, batch_size=10, shuffle=True
+        col_perm, collate_fn=lambda x: x, batch_size=10, shuffle=True
    )
    for batch in dataloader:
        assert batch.size(0) == 1
@@ -25,10 +25,42 @@ import pandas as pd
 import polars as pl
 import pytest
 import lancedb
-from lancedb.util import get_uri_scheme, join_uri, value_to_sql
+from lancedb.util import flatten_columns, get_uri_scheme, join_uri, value_to_sql
 from utils import exception_output


+def _struct_table() -> pa.Table:
+    return pa.table(
+        {
+            "id": [1, 2],
+            "nested": pa.array([{"a": 1, "b": 2}, {"a": 3, "b": 4}]),
+        }
+    )
+
+
+def test_flatten_columns():
+    tbl = _struct_table()
+
+    # None / False mean "do not flatten": the struct column is preserved.
+    # `False` is a regression guard: because bool is a subclass of int it used
+    # to fall into the integer branch and raise ValueError (see issue).
+    for no_flatten in (None, False):
+        result = flatten_columns(tbl, no_flatten)
+        assert result.column_names == ["id", "nested"]
+
+    # True flattens all nested levels.
+    flattened = flatten_columns(tbl, True)
+    assert flattened.column_names == ["id", "nested.a", "nested.b"]
+
+    # A positive integer flattens up to that depth.
+    flattened = flatten_columns(tbl, 1)
+    assert flattened.column_names == ["id", "nested.a", "nested.b"]
+
+    # Non-positive integers are still rejected.
+    with pytest.raises(ValueError):
+        flatten_columns(tbl, 0)
+
+
 def test_normalize_uri():
    uris = [
        "relative/path",
@@ -655,6 +655,46 @@ pub fn connect_namespace_client(
    )))
 }

+#[pyfunction]
+#[pyo3(signature = (
+    namespace_client_impl,
+    namespace_client_properties,
+    read_consistency_interval=None,
+    storage_options=None,
+    session=None,
+    namespace_client_pushdown_operations=None,
+))]
+#[allow(clippy::too_many_arguments)]
+pub fn connect_namespace(
+    namespace_client_impl: String,
+    namespace_client_properties: HashMap<String, String>,
+    read_consistency_interval: Option<f64>,
+    storage_options: Option<HashMap<String, String>>,
+    session: Option<crate::session::Session>,
+    namespace_client_pushdown_operations: Option<Vec<String>>,
+) -> PyResult<Connection> {
+    let read_consistency_interval = read_consistency_interval.map(Duration::from_secs_f64);
+    let namespace_client_pushdown_operations =
+        parse_namespace_client_pushdown_operations(namespace_client_pushdown_operations)?;
+
+    let mut builder =
+        lancedb::connect_namespace(&namespace_client_impl, namespace_client_properties)
+            .pushdown_operations(namespace_client_pushdown_operations);
+    if let Some(storage_options) = storage_options {
+        builder = builder.storage_options(storage_options);
+    }
+    if let Some(read_consistency_interval) = read_consistency_interval {
+        builder = builder.read_consistency_interval(read_consistency_interval);
+    }
+    if let Some(session) = session {
+        builder = builder.session(session.inner.clone());
+    }
+
+    Ok(Connection::new(
+        crate::runtime::block_on(builder.execute()).infer_error()?,
+    ))
+}
+
 /// Whether to build the namespace natively (from impl + properties) instead of
 /// wrapping a pre-built client. Native construction is required for the
 /// read-freshness provider to be installed
@@ -7,12 +7,14 @@
 //! build type-safe filter / projection expressions that map directly to
 //! DataFusion [`Expr`] nodes, bypassing SQL string parsing.

+use std::ops::{Add, Div, Mul, Not, Sub};
+
 use arrow::{datatypes::DataType, pyarrow::PyArrowType};
 use datafusion_common::ScalarValue;
 use lancedb::expr::{
    DfExpr, col as ldb_col, contains, expr_cast, is_in, lit as df_lit, lower, upper,
 };
-use pyo3::types::PyBytes;
+use pyo3::types::{PyBytes, PyDate, PyDateTime};
 use pyo3::{Bound, PyAny, PyResult, exceptions::PyValueError, prelude::*, pyfunction};

 /// A type-safe DataFusion expression.
@@ -63,30 +65,30 @@ impl PyExpr {
        Self(self.0.clone().or(other.0.clone()))
    }

+    /// Logical NOT.
    fn not_(&self) -> Self {
-        use std::ops::Not;
        Self(self.0.clone().not())
    }

    // ── arithmetic ───────────────────────────────────────────────────────────

+    /// Add expressions.
    fn add(&self, other: &Self) -> Self {
-        use std::ops::Add;
        Self(self.0.clone().add(other.0.clone()))
    }

+    /// Subtract expressions.
    fn sub(&self, other: &Self) -> Self {
-        use std::ops::Sub;
        Self(self.0.clone().sub(other.0.clone()))
    }

+    /// Multiply expressions.
    fn mul(&self, other: &Self) -> Self {
-        use std::ops::Mul;
        Self(self.0.clone().mul(other.0.clone()))
    }

+    /// Divide expressions.
    fn div(&self, other: &Self) -> Self {
-        use std::ops::Div;
        Self(self.0.clone().div(other.0.clone()))
    }

@@ -153,7 +155,8 @@ pub fn expr_col(name: &str) -> PyExpr {

 /// Create a literal value expression.
 ///
-/// Supported Python types: `bool`, `int`, `float`, `str`, `bytes`.
+/// Supported Python types: `bool`, `int`, `float`, `str`, `bytes`, `date`,
+/// `datetime`, `Decimal`.
 #[pyfunction]
 pub fn expr_lit(value: Bound<'_, PyAny>) -> PyResult<PyExpr> {
    // bool must be checked before int because bool is a subclass of int in Python
@@ -163,6 +166,19 @@ pub fn expr_lit(value: Bound<'_, PyAny>) -> PyResult<PyExpr> {
    if let Ok(i) = value.extract::<i64>() {
        return Ok(PyExpr(df_lit(i)));
    }
+    // Decimal must be checked before f64: Python's Decimal implements __float__,
+    // so value.extract::<f64>() would succeed and silently truncate the value to
+    // f64, losing precision. Build a Decimal128 scalar to preserve it instead.
+    if value.get_type().name()? == "Decimal" {
+        let s = value.call_method0("__str__")?.extract::<String>()?;
+        // Parse the decimal string into an i128 value, precision, and scale.
+        let (val, precision, scale) = parse_decimal(&s)?;
+        return Ok(PyExpr(df_lit(ScalarValue::Decimal128(
+            Some(val),
+            precision,
+            scale,
+        ))));
+    }
    if let Ok(f) = value.extract::<f64>() {
        return Ok(PyExpr(df_lit(f)));
    }
@@ -173,12 +189,48 @@ pub fn expr_lit(value: Bound<'_, PyAny>) -> PyResult<PyExpr> {
        let bytes = value.extract::<Vec<u8>>()?;
        return Ok(PyExpr(df_lit(ScalarValue::Binary(Some(bytes)))));
    }
+
+    // datetime.datetime is a subclass of datetime.date, so it must be checked first.
+    if let Ok(dt) = value.cast::<PyDateTime>() {
+        let ts: f64 = dt.call_method0("timestamp")?.extract()?;
+        let micros = (ts * 1_000_000.0).round() as i64;
+        return Ok(PyExpr(df_lit(ScalarValue::TimestampMicrosecond(
+            Some(micros),
+            None,
+        ))));
+    }
+    if let Ok(d) = value.cast::<PyDate>() {
+        let ordinal: i32 = d.call_method0("toordinal")?.extract()?;
+        let days = ordinal - 719163; // Unix epoch is 1970-01-01
+        return Ok(PyExpr(df_lit(ScalarValue::Date32(Some(days)))));
+    }
+
    Err(PyValueError::new_err(format!(
-        "unsupported literal type: {}. Supported: bool, int, float, str, bytes",
+        "unsupported literal type: {}. Supported: bool, int, float, str, bytes, date, datetime, Decimal",
        value.get_type().name()?
    )))
 }

+fn parse_decimal(s: &str) -> PyResult<(i128, u8, i8)> {
+    let s = s.trim();
+    let dot_pos = s.find('.');
+    let scale = if let Some(pos) = dot_pos {
+        (s.len() - pos - 1) as i8
+    } else {
+        0
+    };
+
+    let digits = s.replace('.', "");
+    let val = digits
+        .parse::<i128>()
+        .map_err(|e| PyValueError::new_err(format!("failed to parse decimal digits: {}", e)))?;
+
+    // Precision is total number of digits
+    let precision = digits.trim_start_matches('-').len() as u8;
+
+    Ok((val, precision, scale))
+}
+
 /// Call an arbitrary registered SQL function by name.
 ///
 /// See `lancedb::expr::func` for the list of supported function names.
@@ -2,7 +2,7 @@
 // SPDX-FileCopyrightText: Copyright The LanceDB Authors

 use arrow::RecordBatchStream;
-use connection::{Connection, connect, connect_namespace_client};
+use connection::{Connection, connect, connect_namespace, connect_namespace_client};
 use env_logger::Env;
 use expr::{PyExpr, expr_col, expr_func, expr_lit};
 use index::IndexConfig;
@@ -27,6 +27,7 @@ pub mod header;
 pub mod index;
 pub mod namespace;
 pub mod oauth;
+pub mod otel;
 pub mod permutation;
 pub mod query;
 pub mod runtime;
@@ -61,7 +62,17 @@ pub fn _lancedb(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
    m.add_class::<PyAsyncPermutationBuilder>()?;
    m.add_class::<PyPermutationReader>()?;
    m.add_class::<PyExpr>()?;
+    // OpenTelemetry metrics bridge
+    m.add_class::<otel::PyMetricPoint>()?;
+    m.add_class::<otel::PyMetricDescription>()?;
+    m.add_function(wrap_pyfunction!(
+        otel::register_lancedb_metrics_recorder,
+        m
+    )?)?;
+    m.add_function(wrap_pyfunction!(otel::lancedb_metrics_catalog, m)?)?;
+    m.add_function(wrap_pyfunction!(otel::snapshot_lancedb_metrics, m)?)?;
    m.add_function(wrap_pyfunction!(connect, m)?)?;
+    m.add_function(wrap_pyfunction!(connect_namespace, m)?)?;
    m.add_function(wrap_pyfunction!(connect_namespace_client, m)?)?;
    m.add_function(wrap_pyfunction!(permutation::async_permutation_builder, m)?)?;
    m.add_function(wrap_pyfunction!(util::validate_table_name, m)?)?;
@@ -0,0 +1,96 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+//! Python-facing wrappers over [`lancedb::metrics_otel`].
+//!
+//! The aggregation, catalog, and histogram bucketing all live in the LanceDB
+//! core crate; this module only converts the core snapshot types into PyO3
+//! classes and exposes the three entry points to Python, where
+//! `lancedb/otel.py` bridges them into the user's OpenTelemetry `MeterProvider`.
+
+use std::collections::HashMap;
+
+use lancedb::metrics_otel::{MetricPoint, MetricValue};
+use pyo3::prelude::*;
+
+/// One metric data point exposed to Python. For counters and gauges only
+/// `value` is set; for histograms `buckets` (cumulative `le` counts), `count`,
+/// and `sum` are set.
+#[pyclass(name = "MetricPoint", get_all)]
+pub struct PyMetricPoint {
+    name: String,
+    kind: String,
+    attributes: HashMap<String, String>,
+    value: Option<f64>,
+    buckets: Option<Vec<(String, u64)>>,
+    count: Option<u64>,
+    sum: Option<f64>,
+}
+
+impl From<MetricPoint> for PyMetricPoint {
+    fn from(point: MetricPoint) -> Self {
+        let kind = point.kind.as_str().to_string();
+        let (value, buckets, count, sum) = match point.value {
+            MetricValue::Scalar(v) => (Some(v), None, None, None),
+            MetricValue::Histogram {
+                buckets,
+                count,
+                sum,
+            } => (None, Some(buckets), Some(count), Some(sum)),
+        };
+        Self {
+            name: point.name,
+            kind,
+            attributes: point.attributes,
+            value,
+            buckets,
+            count,
+            sum,
+        }
+    }
+}
+
+/// A described metric, used by the Python layer to create instruments up front.
+#[pyclass(name = "MetricDescription", get_all)]
+pub struct PyMetricDescription {
+    name: String,
+    kind: String,
+    unit: Option<String>,
+    description: String,
+}
+
+/// Install the LanceDB metrics recorder as the process-global `metrics` recorder.
+///
+/// Returns `True` if the recorder is installed (now or previously). Returns
+/// `False` if a *different* recorder is already installed — `metrics` allows
+/// only one global recorder per process, so LanceDB cannot coexist with another.
+#[pyfunction]
+pub fn register_lancedb_metrics_recorder() -> bool {
+    lancedb::metrics_otel::register_metrics_recorder()
+}
+
+/// The catalog of described LanceDB metrics. Empty until the recorder is installed.
+#[pyfunction]
+pub fn lancedb_metrics_catalog() -> Vec<PyMetricDescription> {
+    lancedb::metrics_otel::metrics_catalog()
+        .into_iter()
+        .map(|desc| PyMetricDescription {
+            name: desc.name,
+            kind: desc.kind.as_str().to_string(),
+            unit: desc.unit,
+            description: desc.description,
+        })
+        .collect()
+}
+
+/// A point-in-time snapshot of every recorded metric. Empty until the recorder
+/// is installed.
+///
+/// The read is lock-free but not O(1): it walks every registered series and
+/// allocates owned copies of their names and labels. The GIL is released across
+/// that work so a periodic collection doesn't stall other Python threads.
+#[pyfunction]
+pub fn snapshot_lancedb_metrics(py: Python<'_>) -> Vec<PyMetricPoint> {
+    let points = py.detach(lancedb::metrics_otel::snapshot_metrics);
+    points.into_iter().map(PyMetricPoint::from).collect()
+}
@@ -79,13 +79,14 @@ impl PyAsyncPermutationBuilder {

 #[pymethods]
 impl PyAsyncPermutationBuilder {
-    #[pyo3(signature = (*, ratios=None, counts=None, fixed=None, seed=None, split_names=None))]
+    #[pyo3(signature = (*, ratios=None, counts=None, fixed=None, seed=None, clump_size=None, split_names=None))]
    pub fn split_random(
        slf: PyRefMut<'_, Self>,
        ratios: Option<Vec<f64>>,
        counts: Option<Vec<u64>>,
        fixed: Option<u64>,
        seed: Option<u64>,
+        clump_size: Option<u64>,
        split_names: Option<Vec<String>>,
    ) -> PyResult<Self> {
        // Check that exactly one split type is provided
@@ -111,7 +112,14 @@ impl PyAsyncPermutationBuilder {
        };

        slf.modify(|builder| {
-            builder.with_split_strategy(SplitStrategy::Random { seed, sizes }, split_names)
+            builder.with_split_strategy(
+                SplitStrategy::Random {
+                    seed,
+                    sizes,
+                    clump_size,
+                },
+                split_names,
+            )
        })
    }

@@ -322,6 +322,12 @@ impl From<LsmWriteSpec> for lancedb::table::LsmWriteSpec {
    }
 }

+impl From<lancedb::table::LsmWriteSpec> for LsmWriteSpec {
+    fn from(inner: lancedb::table::LsmWriteSpec) -> Self {
+        Self { inner }
+    }
+}
+
 #[pyclass(get_all, from_py_object)]
 #[derive(Clone, Debug)]
 pub struct AddColumnsResult {
@@ -1029,6 +1035,14 @@ impl Table {
        })
    }

+    pub fn get_lsm_write_spec(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.inner_ref()?.clone();
+        future_into_py(self_.py(), async move {
+            let spec = inner.get_lsm_write_spec().await.infer_error()?;
+            Ok(spec.map(LsmWriteSpec::from))
+        })
+    }
+
    pub fn close_lsm_writers(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
        let inner = self_.inner_ref()?.clone();
        future_into_py(self_.py(), async move {
@@ -0,0 +1,101 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+import lancedb
+import pyarrow as pa
+import pytest
+
+# The metrics recorder is process-global and installed once, so the whole
+# bridge is exercised in a single test to avoid cross-test global-state coupling.
+
+
+def _metrics_by_name(reader):
+    data = reader.get_metrics_data()
+    result = {}
+    for resource_metrics in data.resource_metrics:
+        for scope_metrics in resource_metrics.scope_metrics:
+            for metric in scope_metrics.metrics:
+                result[metric.name] = metric
+    return result
+
+
+def test_instrument_lancedb_metrics_exports_object_store_metrics(tmp_path):
+    pytest.importorskip("opentelemetry.sdk.metrics")
+    from lancedb.otel import instrument_lancedb_metrics
+    from opentelemetry.sdk.metrics import MeterProvider
+    from opentelemetry.sdk.metrics.export import InMemoryMetricReader
+
+    reader = InMemoryMetricReader()
+    provider = MeterProvider(metric_readers=[reader])
+    assert instrument_lancedb_metrics(provider)
+
+    # The catalog is populated once the recorder is installed.
+    from lancedb._lancedb import lancedb_metrics_catalog
+
+    catalog = {desc.name: desc for desc in lancedb_metrics_catalog()}
+    # Every metric kind emitted by the object store must be described so it is
+    # surfaced by the bridge (counter, histogram, and gauge).
+    assert catalog["lance_object_store_requests_total"].kind == "counter"
+    assert catalog["lance_object_store_request_duration_seconds"].kind == "histogram"
+    assert catalog["lance_object_store_in_flight_requests"].kind == "gauge"
+    assert catalog["lance_object_store_retryable_responses_total"].kind == "counter"
+
+    # Generate object store activity on the local filesystem (scheme "file").
+    db = lancedb.connect(str(tmp_path))
+    table = db.create_table("t", pa.table({"id": pa.array(range(256))}))
+    assert table.count_rows() == 256
+    assert table.to_arrow().num_rows == 256
+
+    metrics = _metrics_by_name(reader)
+
+    requests = metrics["lance_object_store_requests_total"]
+    points = list(requests.data.data_points)
+    assert points, "expected at least one request data point"
+    # Object store metrics are labelled by `operation` and `base` (the store
+    # scheme, e.g. "file", by default).
+    assert all("base" in p.attributes and "operation" in p.attributes for p in points)
+    assert sum(p.value for p in points) > 0
+
+    # Histograms are decomposed into bucket / count / sum observable counters.
+    bucket = metrics["lance_object_store_request_duration_seconds_bucket"]
+    bucket_points = list(bucket.data.data_points)
+    assert bucket_points
+    assert all("le" in p.attributes for p in bucket_points)
+    # The implicit +Inf bucket must be present and is the cumulative maximum.
+    assert any(p.attributes["le"] == "+Inf" for p in bucket_points)
+
+    count = metrics["lance_object_store_request_duration_seconds_count"]
+    assert sum(p.value for p in count.data.data_points) > 0
+
+    # The `_sum` instrument must also be wired and report positive latency.
+    duration_sum = metrics["lance_object_store_request_duration_seconds_sum"]
+    assert sum(p.value for p in duration_sum.data.data_points) > 0
+
+    # Unit handling: only `_sum` keeps the histogram's unit (seconds); `_bucket`
+    # and `_count` observe cumulative counts and are unitless.
+    assert duration_sum.unit == "s"
+    assert bucket.unit == ""
+    assert count.unit == ""
+
+
+def test_snapshot_empty_before_install_is_safe():
+    # snapshot is callable regardless of installation state and never raises.
+    from lancedb._lancedb import snapshot_lancedb_metrics
+
+    assert isinstance(snapshot_lancedb_metrics(), list)
+
+
+def test_instrument_warns_when_recorder_unavailable(monkeypatch):
+    # A foreign `metrics` recorder already installed -> register returns False;
+    # instrument_lancedb_metrics must warn and return False without instrumenting.
+    pytest.importorskip("opentelemetry.sdk.metrics")
+    import lancedb.otel as otel
+    from opentelemetry.sdk.metrics import MeterProvider
+    from opentelemetry.sdk.metrics.export import InMemoryMetricReader
+
+    monkeypatch.setattr(otel, "register_lancedb_metrics_recorder", lambda: False)
+
+    reader = InMemoryMetricReader()
+    provider = MeterProvider(metric_readers=[reader])
+    with pytest.warns(UserWarning, match="recorder"):
+        assert otel.instrument_lancedb_metrics(provider) is False
@@ -780,7 +780,7 @@ name = "cuda-bindings"
 version = "13.3.1"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
-    { name = "cuda-pathfinder" },
+    { name = "cuda-pathfinder", marker = "(python_full_version < '3.14' and sys_platform == 'emscripten') or (python_full_version < '3.14' and sys_platform == 'win32') or (sys_platform != 'emscripten' and sys_platform != 'win32')" },
 ]
 wheels = [
    { url = "https://files.pythonhosted.org/packages/a9/21/8464d133752951c154feafb3b65c297e7d80f301183d220bec4c830f1441/cuda_bindings-13.3.1-cp310-cp310-manylinux_2_24_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:120fcc53d57903df529c3486962c56528cba5b7d6c57c99537320ed9922c8b86", size = 6073403, upload-time = "2026-05-29T23:11:36.22Z" },
@@ -815,37 +815,37 @@ wheels = [

 [package.optional-dependencies]
 cublas = [
-    { name = "nvidia-cublas", marker = "sys_platform == 'linux' or sys_platform == 'win32'" },
+    { name = "nvidia-cublas", marker = "(python_full_version < '3.14' and sys_platform == 'win32') or sys_platform == 'linux'" },
 ]
 cudart = [
-    { name = "nvidia-cuda-runtime", marker = "sys_platform == 'linux' or sys_platform == 'win32'" },
+    { name = "nvidia-cuda-runtime", marker = "(python_full_version < '3.14' and sys_platform == 'win32') or sys_platform == 'linux'" },
 ]
 cufft = [
-    { name = "nvidia-cufft", marker = "sys_platform == 'linux' or sys_platform == 'win32'" },
+    { name = "nvidia-cufft", marker = "(python_full_version < '3.14' and sys_platform == 'win32') or sys_platform == 'linux'" },
 ]
 cufile = [
    { name = "nvidia-cufile", marker = "sys_platform == 'linux'" },
 ]
 cupti = [
-    { name = "nvidia-cuda-cupti", marker = "sys_platform == 'linux' or sys_platform == 'win32'" },
+    { name = "nvidia-cuda-cupti", marker = "(python_full_version < '3.14' and sys_platform == 'win32') or sys_platform == 'linux'" },
 ]
 curand = [
-    { name = "nvidia-curand", marker = "sys_platform == 'linux' or sys_platform == 'win32'" },
+    { name = "nvidia-curand", marker = "(python_full_version < '3.14' and sys_platform == 'win32') or sys_platform == 'linux'" },
 ]
 cusolver = [
-    { name = "nvidia-cusolver", marker = "sys_platform == 'linux' or sys_platform == 'win32'" },
+    { name = "nvidia-cusolver", marker = "(python_full_version < '3.14' and sys_platform == 'win32') or sys_platform == 'linux'" },
 ]
 cusparse = [
-    { name = "nvidia-cusparse", marker = "sys_platform == 'linux' or sys_platform == 'win32'" },
+    { name = "nvidia-cusparse", marker = "(python_full_version < '3.14' and sys_platform == 'win32') or sys_platform == 'linux'" },
 ]
 nvjitlink = [
-    { name = "nvidia-nvjitlink", marker = "sys_platform == 'linux' or sys_platform == 'win32'" },
+    { name = "nvidia-nvjitlink", marker = "(python_full_version < '3.14' and sys_platform == 'win32') or sys_platform == 'linux'" },
 ]
 nvrtc = [
-    { name = "nvidia-cuda-nvrtc", marker = "sys_platform == 'linux' or sys_platform == 'win32'" },
+    { name = "nvidia-cuda-nvrtc", marker = "(python_full_version < '3.14' and sys_platform == 'win32') or sys_platform == 'linux'" },
 ]
 nvtx = [
-    { name = "nvidia-nvtx", marker = "sys_platform == 'linux' or sys_platform == 'win32'" },
+    { name = "nvidia-nvtx", marker = "(python_full_version < '3.14' and sys_platform == 'win32') or sys_platform == 'linux'" },
 ]

 [[package]]
@@ -1909,6 +1909,9 @@ embeddings = [
    { name = "sentencepiece" },
    { name = "torch" },
 ]
+otel = [
+    { name = "opentelemetry-api" },
+]
 pylance = [
    { name = "pylance" },
 ]
@@ -1923,6 +1926,7 @@ tests = [
    { name = "boto3" },
    { name = "datafusion" },
    { name = "duckdb" },
+    { name = "opentelemetry-sdk" },
    { name = "pandas", version = "2.2.3", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" },
    { name = "pandas", version = "2.3.3", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11' and python_full_version < '3.14'" },
    { name = "pandas", version = "3.0.3", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.14'" },
@@ -1963,6 +1967,8 @@ requires-dist = [
    { name = "open-clip-torch", marker = "extra == 'clip'" },
    { name = "open-clip-torch", marker = "extra == 'embeddings'", specifier = ">=2.20.0" },
    { name = "openai", marker = "extra == 'embeddings'", specifier = ">=1.6.1" },
+    { name = "opentelemetry-api", marker = "extra == 'otel'" },
+    { name = "opentelemetry-sdk", marker = "extra == 'tests'", specifier = ">=1.30.0" },
    { name = "overrides", marker = "python_full_version < '3.12'", specifier = ">=0.7" },
    { name = "packaging", specifier = ">=23.0" },
    { name = "pandas", marker = "extra == 'tests'", specifier = ">=1.4" },
@@ -1994,7 +2000,7 @@ requires-dist = [
    { name = "transformers", marker = "extra == 'siglip'", specifier = ">=4.41.0" },
    { name = "typing-extensions", marker = "python_full_version < '3.11' and extra == 'dev'", specifier = ">=4.0.0" },
 ]
-provides-extras = ["azure", "clip", "dev", "docs", "embeddings", "pylance", "siglip", "tests"]
+provides-extras = ["azure", "clip", "dev", "docs", "embeddings", "otel", "pylance", "siglip", "tests"]

 [[package]]
 name = "lomond"
@@ -2775,7 +2781,7 @@ name = "nvidia-cudnn-cu13"
 version = "9.19.0.56"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
-    { name = "nvidia-cublas" },
+    { name = "nvidia-cublas", marker = "(python_full_version < '3.14' and sys_platform == 'emscripten') or (python_full_version < '3.14' and sys_platform == 'win32') or (sys_platform != 'emscripten' and sys_platform != 'win32')" },
 ]
 wheels = [
    { url = "https://files.pythonhosted.org/packages/f1/84/26025437c1e6b61a707442184fa0c03d083b661adf3a3eecfd6d21677740/nvidia_cudnn_cu13-9.19.0.56-py3-none-manylinux_2_27_aarch64.whl", hash = "sha256:6ed29ffaee1176c612daf442e4dd6cfeb6a0caa43ddcbeb59da94953030b1be4", size = 433781201, upload-time = "2026-02-03T20:40:53.805Z" },
@@ -2787,7 +2793,7 @@ name = "nvidia-cufft"
 version = "12.0.0.61"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
-    { name = "nvidia-nvjitlink" },
+    { name = "nvidia-nvjitlink", marker = "(python_full_version < '3.14' and sys_platform == 'emscripten') or (python_full_version < '3.14' and sys_platform == 'win32') or (sys_platform != 'emscripten' and sys_platform != 'win32')" },
 ]
 wheels = [
    { url = "https://files.pythonhosted.org/packages/8b/ae/f417a75c0259e85c1d2f83ca4e960289a5f814ed0cea74d18c353d3e989d/nvidia_cufft-12.0.0.61-py3-none-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:2708c852ef8cd89d1d2068bdbece0aa188813a0c934db3779b9b1faa8442e5f5", size = 214053554, upload-time = "2025-09-04T08:31:38.196Z" },
@@ -2817,9 +2823,9 @@ name = "nvidia-cusolver"
 version = "12.0.4.66"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
-    { name = "nvidia-cublas" },
-    { name = "nvidia-cusparse" },
-    { name = "nvidia-nvjitlink" },
+    { name = "nvidia-cublas", marker = "(python_full_version < '3.14' and sys_platform == 'emscripten') or (python_full_version < '3.14' and sys_platform == 'win32') or (sys_platform != 'emscripten' and sys_platform != 'win32')" },
+    { name = "nvidia-cusparse", marker = "(python_full_version < '3.14' and sys_platform == 'emscripten') or (python_full_version < '3.14' and sys_platform == 'win32') or (sys_platform != 'emscripten' and sys_platform != 'win32')" },
+    { name = "nvidia-nvjitlink", marker = "(python_full_version < '3.14' and sys_platform == 'emscripten') or (python_full_version < '3.14' and sys_platform == 'win32') or (sys_platform != 'emscripten' and sys_platform != 'win32')" },
 ]
 wheels = [
    { url = "https://files.pythonhosted.org/packages/c8/c3/b30c9e935fc01e3da443ec0116ed1b2a009bb867f5324d3f2d7e533e776b/nvidia_cusolver-12.0.4.66-py3-none-manylinux_2_27_aarch64.whl", hash = "sha256:02c2457eaa9e39de20f880f4bd8820e6a1cfb9f9a34f820eb12a155aa5bc92d2", size = 223467760, upload-time = "2025-09-04T08:33:04.222Z" },
@@ -2831,7 +2837,7 @@ name = "nvidia-cusparse"
 version = "12.6.3.3"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
-    { name = "nvidia-nvjitlink" },
+    { name = "nvidia-nvjitlink", marker = "(python_full_version < '3.14' and sys_platform == 'emscripten') or (python_full_version < '3.14' and sys_platform == 'win32') or (sys_platform != 'emscripten' and sys_platform != 'win32')" },
 ]
 wheels = [
    { url = "https://files.pythonhosted.org/packages/f8/94/5c26f33738ae35276672f12615a64bd008ed5be6d1ebcb23579285d960a9/nvidia_cusparse-12.6.3.3-py3-none-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:80bcc4662f23f1054ee334a15c72b8940402975e0eab63178fc7e670aa59472c", size = 162155568, upload-time = "2025-09-04T08:33:42.864Z" },
@@ -2934,6 +2940,45 @@ wheels = [
    { url = "https://files.pythonhosted.org/packages/f6/46/180e14be801a75bc13f234cb1b594b232adeb9c84e60a9ab1832e8333591/openai-2.40.0-py3-none-any.whl", hash = "sha256:2b205637ff214477f9ce9ab035e9f494db0e3fa8f1e599008953735fbf6ff1ff", size = 1350935, upload-time = "2026-06-01T21:48:21.462Z" },
 ]

+[[package]]
+name = "opentelemetry-api"
+version = "1.43.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "typing-extensions" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/ae/cc/e4c9584181f86494df0f6bdec1a4f3280c50db44704dc2a407e994fc87bb/opentelemetry_api-1.43.0.tar.gz", hash = "sha256:107d0d03857ea8fc7c5fcbbbd83f800c281f0d560553d61c1d675fccfd1761c1", size = 73476, upload-time = "2026-06-24T15:19:55.323Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/17/83/6dba32b85f31868400440dc7ad2ca1eab94cbbf3a7b0459ed39f8311a9e2/opentelemetry_api-1.43.0-py3-none-any.whl", hash = "sha256:20acf45e9b21851926835292e4045d290acade1edd2ff3de86d2f069687ba1fd", size = 61912, upload-time = "2026-06-24T15:19:35.434Z" },
+]
+
+[[package]]
+name = "opentelemetry-sdk"
+version = "1.43.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "opentelemetry-api" },
+    { name = "opentelemetry-semantic-conventions" },
+    { name = "typing-extensions" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/3e/eb/5041074274ac0956b03637cc039d434569112468e875eddfcc9a0674ce06/opentelemetry_sdk-1.43.0.tar.gz", hash = "sha256:d8187c81c162df9913e4003dd6485f7390d9a24fc17026ec7387b8b8218b08e9", size = 254744, upload-time = "2026-06-24T15:20:08.467Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/49/e3/b17be23af124201c9f52eececd4cc8ddfed1597d37b4ee771895d325805c/opentelemetry_sdk-1.43.0-py3-none-any.whl", hash = "sha256:d1323a547c1ce69d6a069a17a44b7da82bb8b332051ecb074041f87642c86823", size = 178852, upload-time = "2026-06-24T15:19:52.169Z" },
+]
+
+[[package]]
+name = "opentelemetry-semantic-conventions"
+version = "0.64b0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "opentelemetry-api" },
+    { name = "typing-extensions" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/5a/30/5f26df29509eccd86b99b481ac9ffa39da49ba9577cc69071c552ae30447/opentelemetry_semantic_conventions-0.64b0.tar.gz", hash = "sha256:72f76fb2d1582d9d033dd1fcd84532e961e6ff3d90d24ba6fabc72975a83864c", size = 148340, upload-time = "2026-06-24T15:20:09.267Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/f2/ca/23ba87a221b574a7c5a99d48849d80bfe8b047624681357e2b002e566187/opentelemetry_semantic_conventions-0.64b0-py3-none-any.whl", hash = "sha256:ea77e85e354b8f604ddbe5f3d9135216f982fa4d77e5859ac30f6d8a50505aa6", size = 203713, upload-time = "2026-06-24T15:19:53.339Z" },
+]
+
 [[package]]
 name = "overrides"
 version = "7.7.0"
@@ -1,6 +1,6 @@
 [package]
 name = "lancedb"
-version = "0.31.0-beta.4"
+version = "0.32.0-beta.0"
 edition.workspace = true
 description = "LanceDB: A serverless, low-latency vector database for AI applications"
 license.workspace = true
@@ -14,6 +14,7 @@ rust-version.workspace = true
 ahash = { workspace = true }
 arrow = { workspace = true }
 arrow-array = { workspace = true }
+arrow-buffer = { workspace = true }
 arrow-data = { workspace = true }
 arrow-schema = { workspace = true }
 arrow-select = { workspace = true }
@@ -48,6 +49,8 @@ lance-encoding = { workspace = true }
 lance-arrow = { workspace = true }
 lance-namespace = { workspace = true }
 lance-namespace-impls = { workspace = true }
+metrics = { workspace = true, optional = true }
+metrics-util = { workspace = true, optional = true }
 moka = { workspace = true }
 pin-project = { workspace = true }
 tokio = { version = "1.23", features = ["rt-multi-thread", "sync"] }
@@ -107,6 +110,8 @@ http-body = "1"                                        # Matching reqwest
 rstest = "0.23.0"
 test-log = "0.2"
 serial_test = "3"
+[target.'cfg(unix)'.dev-dependencies]
+pprof = { version = "0.14", features = ["flamegraph"] }


 [features]
@@ -125,6 +130,12 @@ azure = [
    "lance-namespace-impls/dir-azure",
    "lance-namespace-impls/credential-vendor-azure",
 ]
+cos = ["lance/tencent", "lance-io/tencent"]
+goosefs = [
+    "lance/goosefs",
+    "lance-io/goosefs",
+    "lance-namespace-impls/dir-goosefs",
+]
 huggingface = [
    "lance/huggingface",
    "lance-io/huggingface",
@@ -138,6 +149,15 @@ remote = [
    "lance-namespace-impls/rest",
    "lance-namespace-impls/rest-adapter",
 ]
+# Publish LanceDB's internal metrics (currently object store request counts,
+# bytes, latency, errors, and throttles) through the `metrics` crate facade,
+# and re-export the `metrics` crate as `lancedb::metrics`. Install any
+# `metrics`-compatible recorder to collect them.
+metrics = ["dep:metrics", "lance/metrics", "lance-io/metrics"]
+# Additional adapter on top of `metrics` that installs a process-global recorder
+# and exposes a pull-based snapshot/catalog API (see `lancedb::metrics_otel`)
+# for bridging metrics into OpenTelemetry or other pull-based exporters.
+metrics-otel = ["metrics", "dep:metrics-util"]
 fp16kernels = ["lance-linalg/fp16kernels"]
 s3-test = []
 bedrock = ["dep:aws-sdk-bedrockruntime"]
@@ -163,9 +183,16 @@ required-features = ["sentence-transformers"]
 name = "bedrock"
 required-features = ["bedrock"]

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

+[[example]]
+name = "polars"
+required-features = ["polars"]
+
 [[example]]
 name = "full_text_search"

@@ -0,0 +1,272 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+//! Benchmark + CPU profiler for the PermutationReader used by the elastic
+//! streaming dataloader.
+//!
+//! Normal sweep:
+//!   cargo run --release --example bench_streaming_dataloader
+//!
+//! Flamegraph (self-contained, no perf/dtrace needed):
+//!   BENCH_PROFILE=1 BENCH_CHUNK=64 cargo run --release \
+//!       --example bench_streaming_dataloader
+//!   # writes flamegraph.svg in the current directory
+//!
+//! Environment variables:
+//!   BENCH_NUM_ROWS   – total rows (default 49152 = 24 × 2048)
+//!   BENCH_NUM_SPLITS – number of splits (default 24)
+//!   BENCH_STEPS      – round-robin cycles per chunk-size trial (default 200)
+//!   BENCH_ROW_BYTES  – bytes of payload per row (default 4096)
+//!   BENCH_CHUNK      – restrict sweep to this single chunk size
+//!   BENCH_PROFILE    – if set to "1", capture a pprof flamegraph SVG
+
+use std::{sync::Arc, time::Instant};
+
+use arrow_array::{Int32Array, LargeBinaryArray, RecordBatch};
+use arrow_schema::{DataType, Field, Schema};
+use lancedb::{
+    Result, Table,
+    arrow::{SendableRecordBatchStream, SimpleRecordBatchStream},
+    connect,
+    dataloader::permutation::{
+        builder::{PermutationBuilder, ShuffleStrategy},
+        reader::PermutationReader,
+        split::{SplitSizes, SplitStrategy},
+    },
+    query::Select,
+};
+
+fn env_usize(key: &str, default: usize) -> usize {
+    std::env::var(key)
+        .ok()
+        .and_then(|v| v.parse().ok())
+        .unwrap_or(default)
+}
+
+// ---------------------------------------------------------------------------
+// Table creation
+// ---------------------------------------------------------------------------
+
+async fn make_base_table(num_rows: usize, row_bytes: usize) -> Result<Table> {
+    let schema = Arc::new(Schema::new(vec![
+        Field::new("id", DataType::Int32, false),
+        Field::new("payload", DataType::LargeBinary, false),
+    ]));
+    let payload = vec![0u8; row_bytes];
+    let ids: Int32Array = (0..num_rows as i32).collect();
+    let payloads: LargeBinaryArray = (0..num_rows).map(|_| Some(payload.as_slice())).collect();
+    let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(ids), Arc::new(payloads)])?;
+    let stream: SendableRecordBatchStream = Box::pin(SimpleRecordBatchStream::new(
+        futures::stream::once(std::future::ready(Ok(batch))),
+        schema,
+    ));
+    let db = connect("memory:///").execute().await?;
+    db.create_table("base", stream).execute().await
+}
+
+async fn make_permutation_table(base: &Table, num_splits: usize) -> Result<Table> {
+    PermutationBuilder::new(base.clone())
+        .with_split_strategy(
+            SplitStrategy::Random {
+                seed: Some(42),
+                sizes: SplitSizes::Fixed(num_splits as u64),
+                clump_size: None,
+            },
+            None,
+        )
+        .with_shuffle_strategy(ShuffleStrategy::Random {
+            seed: Some(42),
+            clump_size: None,
+        })
+        .build()
+        .await
+}
+
+// ---------------------------------------------------------------------------
+// Round-robin hot loop (mirrors StreamingDataset.__iter__)
+// ---------------------------------------------------------------------------
+
+async fn run_hot_loop(
+    readers: &[PermutationReader],
+    chunk_size: usize,
+    steps: usize,
+) -> Result<(usize, f64)> {
+    let n = readers.len();
+    let split_sizes: Vec<usize> = readers.iter().map(|r| r.count_rows() as usize).collect();
+
+    struct SplitBuf {
+        batch: Option<RecordBatch>,
+        row_in_batch: usize,
+        consumed: usize,
+    }
+    let mut bufs: Vec<SplitBuf> = (0..n)
+        .map(|_| SplitBuf {
+            batch: None,
+            row_in_batch: 0,
+            consumed: 0,
+        })
+        .collect();
+
+    // Pre-fill
+    for i in 0..n {
+        let fetch = chunk_size.min(split_sizes[i]);
+        if fetch > 0 {
+            let offsets: Vec<u64> = (0..fetch as u64).collect();
+            bufs[i].batch = Some(readers[i].take_offsets(&offsets, Select::All).await?);
+        }
+    }
+
+    let mut total_rows = 0usize;
+    let t0 = Instant::now();
+
+    'outer: for _step in 0..steps {
+        for i in 0..n {
+            if bufs[i].consumed >= split_sizes[i] {
+                break 'outer;
+            }
+            let need_refill = bufs[i]
+                .batch
+                .as_ref()
+                .map(|b| bufs[i].row_in_batch >= b.num_rows())
+                .unwrap_or(true);
+            if need_refill {
+                let start = bufs[i].consumed as u64;
+                let remaining = (split_sizes[i] - bufs[i].consumed) as u64;
+                let fetch = chunk_size.min(remaining as usize);
+                let offsets: Vec<u64> = (start..start + fetch as u64).collect();
+                bufs[i].batch = Some(readers[i].take_offsets(&offsets, Select::All).await?);
+                bufs[i].row_in_batch = 0;
+            }
+            bufs[i].row_in_batch += 1;
+            bufs[i].consumed += 1;
+            total_rows += 1;
+        }
+    }
+
+    Ok((total_rows, t0.elapsed().as_secs_f64()))
+}
+
+// ---------------------------------------------------------------------------
+// Main
+// ---------------------------------------------------------------------------
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    let num_splits = env_usize("BENCH_NUM_SPLITS", 24);
+    let num_rows = env_usize("BENCH_NUM_ROWS", num_splits * 2048);
+    let steps = env_usize("BENCH_STEPS", 200);
+    let row_bytes = env_usize("BENCH_ROW_BYTES", 4096);
+    let single_chunk: Option<usize> = std::env::var("BENCH_CHUNK")
+        .ok()
+        .and_then(|v| v.parse().ok());
+    let do_profile = std::env::var("BENCH_PROFILE")
+        .map(|v| v == "1")
+        .unwrap_or(false);
+
+    assert_eq!(
+        num_rows % num_splits,
+        0,
+        "NUM_ROWS must be divisible by NUM_SPLITS"
+    );
+
+    println!("Benchmark config:");
+    println!(
+        "  num_rows={}  num_splits={}  rows/split={}  steps={}  row_bytes={}",
+        num_rows,
+        num_splits,
+        num_rows / num_splits,
+        steps,
+        row_bytes,
+    );
+    println!(
+        "  ~{:.1} MB total",
+        (num_rows * row_bytes) as f64 / (1024.0 * 1024.0)
+    );
+    println!();
+
+    print!("Building base table... ");
+    let _ = std::io::Write::flush(&mut std::io::stdout());
+    let base = make_base_table(num_rows, row_bytes).await?;
+    println!("done");
+
+    print!("Building permutation table... ");
+    let _ = std::io::Write::flush(&mut std::io::stdout());
+    let perm = make_permutation_table(&base, num_splits).await?;
+    println!("done");
+
+    print!("Building {} PermutationReaders... ", num_splits);
+    let _ = std::io::Write::flush(&mut std::io::stdout());
+    let base_inner = base.base_table().clone();
+    let perm_inner = perm.base_table().clone();
+    let mut readers = Vec::with_capacity(num_splits);
+    for split in 0..num_splits {
+        readers.push(
+            PermutationReader::try_from_tables(
+                base_inner.clone(),
+                perm_inner.clone(),
+                split as u64,
+            )
+            .await?,
+        );
+    }
+    println!("done ({} rows/split)", readers[0].count_rows());
+    println!();
+
+    let chunk_sizes: Vec<usize> = if let Some(c) = single_chunk {
+        vec![c]
+    } else {
+        vec![1, 4, 16, 64, 256, 1024, 4096, 16384]
+    };
+
+    if do_profile {
+        #[cfg(unix)]
+        {
+            let chunk = chunk_sizes[0];
+            println!("Profiling chunk={chunk} for {steps} steps...");
+            // Warm-up outside the profiler window
+            let _ = run_hot_loop(&readers, chunk, 1).await?;
+
+            let guard = pprof::ProfilerGuardBuilder::default()
+                .frequency(1000)
+                .build()
+                .unwrap();
+
+            let (rows, elapsed) = run_hot_loop(&readers, chunk, steps).await?;
+
+            if let Ok(report) = guard.report().build() {
+                let svg_path = "flamegraph.svg";
+                let file = std::fs::File::create(svg_path).unwrap();
+                report.flamegraph(file).unwrap();
+                println!("Flamegraph written to {svg_path}");
+            }
+
+            let rows_per_sec = rows as f64 / elapsed;
+            println!("chunk={chunk}  {rows} rows  {elapsed:.3}s  {rows_per_sec:.0} rows/s");
+        }
+        #[cfg(not(unix))]
+        {
+            println!("Flamegraph profiling (BENCH_PROFILE=1) is not supported on this platform.");
+            println!("Run without BENCH_PROFILE to get throughput numbers.");
+        }
+    } else {
+        println!(
+            "{:>6}  {:>7}  {:>8}  {:>11}  {:>10}",
+            "chunk", "rows", "elapsed", "rows/s", "ms/step"
+        );
+        println!("{}", "-".repeat(52));
+
+        for &chunk in &chunk_sizes {
+            let _ = run_hot_loop(&readers, chunk, 1).await?;
+            let (rows, elapsed) = run_hot_loop(&readers, chunk, steps).await?;
+            let rows_per_sec = rows as f64 / elapsed;
+            let ms_per_step = elapsed / steps as f64 * 1000.0;
+            println!(
+                "{:>6}  {:>7}  {:>7.3}s  {:>11.0}  {:>9.1}ms",
+                chunk, rows, elapsed, rows_per_sec, ms_per_step,
+            );
+        }
+    }
+
+    println!("\nDone.");
+    Ok(())
+}
@@ -0,0 +1,47 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+//! This example demonstrates ingesting a Polars DataFrame into LanceDB and
+//! reading it back out as a Polars DataFrame.
+
+use lancedb::arrow::IntoPolars;
+use lancedb::query::ExecutableQuery;
+use lancedb::{Result, connect};
+use polars::prelude::{DataFrame, NamedFrom, Series};
+
+fn make_dataframe() -> DataFrame {
+    let ids = Series::new("id", &[1i32, 2, 3, 4, 5]);
+    let names = Series::new("name", &["Alice", "Bob", "Carol", "Dave", "Eve"]);
+    let scores = Series::new("score", &[9.5f64, 8.1, 7.3, 9.0, 6.5]);
+    DataFrame::new(vec![ids, names, scores]).unwrap()
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    let tmp = tempfile::tempdir().unwrap();
+    let db = connect(tmp.path().to_str().unwrap()).execute().await?;
+
+    // Ingest a Polars DataFrame directly — DataFrame now implements Scannable.
+    let df = make_dataframe();
+    println!("Input DataFrame:\n{df}");
+
+    let table = db.create_table("people", df).execute().await?;
+
+    // Append more rows.
+    let more = DataFrame::new(vec![
+        Series::new("id", &[6i32, 7]),
+        Series::new("name", &["Frank", "Grace"]),
+        Series::new("score", &[7.8f64, 8.9]),
+    ])
+    .unwrap();
+    table.add(more).execute().await?;
+
+    // Read back as a Polars DataFrame.
+    let result_df = table.query().execute().await?.into_polars().await?;
+
+    println!(
+        "\nRound-tripped DataFrame ({} rows):\n{result_df}",
+        result_df.height()
+    );
+    Ok(())
+}
@@ -3,7 +3,19 @@

 use std::{pin::Pin, sync::Arc};

+// Re-export the arrow crates we depend on so downstream consumers can build
+// `RecordBatch`/arrays/builders against the exact same arrow line lancedb was
+// compiled against, instead of declaring their own (potentially mismatched)
+// direct arrow dependencies. See https://github.com/lancedb/lancedb/issues/3575.
+pub use arrow;
+pub use arrow_array;
+pub use arrow_buffer;
+pub use arrow_cast;
+pub use arrow_data;
+pub use arrow_ipc;
+pub use arrow_ord;
 pub use arrow_schema;
+pub use arrow_select;
 use datafusion_common::DataFusionError;
 use datafusion_physical_plan::stream::RecordBatchStreamAdapter;
 use futures::{Stream, StreamExt, TryStreamExt};
@@ -112,54 +124,14 @@ impl<S: Stream<Item = Result<arrow_array::RecordBatch>>> RecordBatchStream

 /// A trait for converting incoming data to Arrow
 ///
-/// Integrations should implement this trait to allow data to be
-/// imported directly from the integration.  For example, implementing
-/// this trait for `Vec<Vec<...>>` would allow the `Vec` to be directly
-/// used in methods like [`crate::connection::Connection::create_table`]
-/// or [`crate::table::Table::add`]
-pub trait IntoArrow {
-    /// Convert the data into an iterator of Arrow batches
-    fn into_arrow(self) -> Result<Box<dyn arrow_array::RecordBatchReader + Send>>;
-}
-
 pub type BoxedRecordBatchReader = Box<dyn arrow_array::RecordBatchReader + Send>;

-impl<T: arrow_array::RecordBatchReader + Send + 'static> IntoArrow for T {
-    fn into_arrow(self) -> Result<Box<dyn arrow_array::RecordBatchReader + Send>> {
-        Ok(Box::new(self))
-    }
-}
-
-/// A trait for converting incoming data to Arrow asynchronously
-///
-/// Serves the same purpose as [`IntoArrow`], but for asynchronous data.
-///
-/// Note: Arrow has no async equivalent to RecordBatchReader and so
-pub trait IntoArrowStream {
-    /// Convert the data into a stream of Arrow batches
-    fn into_arrow(self) -> Result<SendableRecordBatchStream>;
-}
-
 impl<S: Stream<Item = Result<arrow_array::RecordBatch>>> SimpleRecordBatchStream<S> {
    pub fn new(stream: S, schema: Arc<arrow_schema::Schema>) -> Self {
        Self { schema, stream }
    }
 }

-impl IntoArrowStream for SendableRecordBatchStream {
-    fn into_arrow(self) -> Result<SendableRecordBatchStream> {
-        Ok(self)
-    }
-}
-
-impl IntoArrowStream for datafusion_physical_plan::SendableRecordBatchStream {
-    fn into_arrow(self) -> Result<SendableRecordBatchStream> {
-        let schema = self.schema();
-        let stream = self.map_err(|df_err| df_err.into());
-        Ok(Box::pin(SimpleRecordBatchStream::new(stream, schema)))
-    }
-}
-
 pub trait LanceDbDatagenExt {
    fn into_ldb_stream(
        self,
@@ -264,9 +236,7 @@ impl IntoPolars for SendableRecordBatchStream {
 #[cfg(all(test, feature = "polars"))]
 mod tests {
    use super::SendableRecordBatchStream;
-    use crate::arrow::{
-        IntoArrow, IntoPolars, PolarsDataFrameRecordBatchReader, SimpleRecordBatchStream,
-    };
+    use crate::arrow::{IntoPolars, PolarsDataFrameRecordBatchReader, SimpleRecordBatchStream};
    use polars::prelude::{DataFrame, NamedFrom, Series};

    fn get_record_batch_reader_from_polars() -> Box<dyn arrow_array::RecordBatchReader + Send> {
@@ -280,10 +250,7 @@ mod tests {
        float_series = Series::new("float", &[2.0]);
        let df2 = DataFrame::new(vec![string_series, int_series, float_series]).unwrap();

-        PolarsDataFrameRecordBatchReader::new(df1.vstack(&df2).unwrap())
-            .unwrap()
-            .into_arrow()
-            .unwrap()
+        Box::new(PolarsDataFrameRecordBatchReader::new(df1.vstack(&df2).unwrap()).unwrap())
    }

    #[test]
@@ -185,6 +185,43 @@ impl Scannable for SendableRecordBatchStream {
    }
 }

+#[cfg(feature = "polars")]
+impl Scannable for polars::frame::DataFrame {
+    fn schema(&self) -> SchemaRef {
+        crate::polars_arrow_convertors::convert_polars_df_schema_to_arrow_rb_schema(
+            self.schema().clone(),
+        )
+        .expect("failed to convert Polars DataFrame schema to Arrow schema")
+    }
+
+    fn scan_as_stream(&mut self) -> SendableRecordBatchStream {
+        let schema = Scannable::schema(self);
+        let batches: crate::Result<Vec<RecordBatch>> =
+            match crate::arrow::PolarsDataFrameRecordBatchReader::new(self.clone()) {
+                Err(e) => Err(e),
+                Ok(reader) => reader.map(|b| b.map_err(Into::into)).collect(),
+            };
+        match batches {
+            Err(e) => Box::pin(SimpleRecordBatchStream {
+                schema,
+                stream: once(async move { Err(e) }),
+            }),
+            Ok(batches) => {
+                let stream = futures::stream::iter(batches.into_iter().map(Ok));
+                Box::pin(SimpleRecordBatchStream { schema, stream })
+            }
+        }
+    }
+
+    fn num_rows(&self) -> Option<usize> {
+        Some(self.height())
+    }
+
+    fn rescannable(&self) -> bool {
+        true
+    }
+}
+
 #[async_trait]
 impl StreamingWriteSource for Box<dyn Scannable> {
    fn arrow_schema(&self) -> SchemaRef {
@@ -1089,4 +1126,60 @@ mod tests {
            );
        }
    }
+
+    #[cfg(feature = "polars")]
+    mod polars_tests {
+        use super::*;
+        use crate::arrow::IntoPolars;
+        use crate::query::ExecutableQuery;
+        use polars::prelude::{DataFrame, NamedFrom, Series};
+
+        fn make_df() -> DataFrame {
+            DataFrame::new(vec![
+                Series::new("id", &[1i32, 2, 3]),
+                Series::new("val", &[1.1f64, 2.2, 3.3]),
+            ])
+            .unwrap()
+        }
+
+        #[tokio::test]
+        async fn test_dataframe_scannable_round_trip() {
+            let tmp = tempfile::tempdir().unwrap();
+            let db = crate::connect(tmp.path().to_str().unwrap())
+                .execute()
+                .await
+                .unwrap();
+
+            let df = make_df();
+            let table = db.create_table("t", df.clone()).execute().await.unwrap();
+
+            // Append the same rows again.
+            table.add(df.clone()).execute().await.unwrap();
+
+            let result = table
+                .query()
+                .execute()
+                .await
+                .unwrap()
+                .into_polars()
+                .await
+                .unwrap();
+
+            assert_eq!(result.height(), df.height() * 2);
+            assert_eq!(result.schema(), df.schema());
+        }
+
+        #[tokio::test]
+        async fn test_dataframe_scannable_rescannable() {
+            let mut df = make_df();
+            assert!(df.rescannable());
+
+            let batches1: Vec<RecordBatch> = df.scan_as_stream().try_collect().await.unwrap();
+            assert_eq!(batches1.iter().map(|b| b.num_rows()).sum::<usize>(), 3);
+
+            // Can be scanned again.
+            let batches2: Vec<RecordBatch> = df.scan_as_stream().try_collect().await.unwrap();
+            assert_eq!(batches2.iter().map(|b| b.num_rows()).sum::<usize>(), 3);
+        }
+    }
 }
@@ -342,6 +342,19 @@ impl ListingDatabase {
        ))
    }

+    fn storage_base_uri(uri: &str) -> String {
+        let Ok(mut url) = url::Url::parse(uri) else {
+            return uri.to_string();
+        };
+        url.set_query(None);
+        let Some((storage_scheme, _commit_scheme)) = url.scheme().split_once('+') else {
+            return url.to_string();
+        };
+        let storage_scheme = storage_scheme.to_string();
+        let _ = url.set_scheme(&storage_scheme);
+        url.to_string()
+    }
+
    async fn prepare_namespace_root(
        uri: &str,
        storage_options: &HashMap<String, String>,
@@ -520,6 +533,8 @@ impl ListingDatabase {
                // will add a trailing '?' to the url
                url.set_query(None);

+                let storage_base_uri = Self::storage_base_uri(url.as_str());
+
                let table_base_uri = if let Some(store) = engine {
                    static WARN_ONCE: std::sync::Once = std::sync::Once::new();
                    WARN_ONCE.call_once(|| {
@@ -532,8 +547,6 @@ impl ListingDatabase {
                    url.to_string()
                };

-                let plain_uri = url.to_string();
-
                let session = request
                    .session
                    .clone()
@@ -550,13 +563,13 @@ impl ListingDatabase {
                };
                let (object_store, base_path) = ObjectStore::from_uri_and_params(
                    session.store_registry(),
-                    &plain_uri,
+                    &storage_base_uri,
                    &os_params,
                )
                .await?;
                if object_store.is_local() {
-                    Self::try_create_dir(&plain_uri).context(CreateDirSnafu {
-                        path: plain_uri.clone(),
+                    Self::try_create_dir(&storage_base_uri).context(CreateDirSnafu {
+                        path: storage_base_uri.clone(),
                    })?;
                }

@@ -570,7 +583,7 @@ impl ListingDatabase {
                };

                let namespace_database = Self::connect_namespace_database(
-                    &table_base_uri,
+                    &storage_base_uri,
                    options.storage_options.clone(),
                    request.namespace_client_properties.clone(),
                    request.read_consistency_interval,
@@ -1309,6 +1322,60 @@ mod tests {
        (tempdir, db)
    }

+    #[tokio::test]
+    async fn test_listing_database_root_ops_do_not_create_manifest() {
+        let tempdir = tempdir().unwrap();
+        let uri = tempdir.path().to_str().unwrap();
+
+        let request = ConnectRequest {
+            uri: uri.to_string(),
+            #[cfg(feature = "remote")]
+            client_config: Default::default(),
+            options: Default::default(),
+            namespace_client_properties: Default::default(),
+            manifest_enabled: false,
+            read_consistency_interval: None,
+            session: None,
+        };
+
+        let db = ListingDatabase::connect_with_options(&request)
+            .await
+            .unwrap();
+
+        assert!(!tempdir.path().join("__manifest").exists());
+
+        let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
+        db.create_table(CreateTableRequest {
+            name: "root_table".to_string(),
+            namespace_path: vec![],
+            data: Box::new(RecordBatch::new_empty(schema)) as Box<dyn Scannable>,
+            mode: CreateTableMode::Create,
+            write_options: Default::default(),
+            location: None,
+            namespace_client: None,
+        })
+        .await
+        .unwrap();
+
+        db.open_table(OpenTableRequest {
+            name: "root_table".to_string(),
+            namespace_path: vec![],
+            index_cache_size: None,
+            lance_read_params: None,
+            location: None,
+            namespace_client: None,
+            managed_versioning: None,
+        })
+        .await
+        .unwrap();
+
+        #[allow(deprecated)]
+        let table_names = db.table_names(TableNamesRequest::default()).await.unwrap();
+
+        assert_eq!(table_names, vec!["root_table".to_string()]);
+        assert!(!tempdir.path().join("__manifest").exists());
+    }
+
    #[tokio::test]
    async fn test_clone_table_basic() {
        let (_tempdir, db) = setup_database().await;
@@ -2283,6 +2350,24 @@ mod tests {
        assert_eq!(captured.as_deref(), Some("foo=bar"));
    }

+    #[test]
+    fn test_storage_base_uri_strips_commit_engine_scheme() {
+        assert_eq!(
+            ListingDatabase::storage_base_uri("s3+ddb://bucket/prefix?ddbTableName=commit_table"),
+            "s3://bucket/prefix"
+        );
+
+        assert_eq!(
+            ListingDatabase::storage_base_uri("s3://bucket/prefix?foo=bar"),
+            "s3://bucket/prefix"
+        );
+
+        assert_eq!(
+            ListingDatabase::storage_base_uri("/tmp/lancedb"),
+            "/tmp/lancedb"
+        );
+    }
+
    /// Regression: connecting via a URL-style URI (which goes through
    /// `url::Url::parse` and the `query_pairs_mut()` path) must not
    /// append a trailing `?` to per-table URIs when the input URI has
@@ -391,6 +391,7 @@ mod tests {
                SplitStrategy::Random {
                    seed: Some(42),
                    sizes: SplitSizes::Percentages(vec![0.05, 0.30]),
+                    clump_size: None,
                },
                None,
            )
@@ -21,6 +21,7 @@ use arrow::compute::concat_batches;
 use arrow::datatypes::UInt64Type;
 use arrow_array::{RecordBatch, UInt64Array};
 use arrow_schema::SchemaRef;
+use datafusion_expr::{Expr, col, lit};
 use futures::{StreamExt, TryStreamExt};
 use lance::dataset::scanner::DatasetRecordBatchStream;
 use lance::io::RecordBatchStream;
@@ -196,17 +197,10 @@ impl PermutationReader {
            .expect_ok()?
            .values();

-        let filter = format!(
-            "_rowid in ({})",
-            row_ids
-                .iter()
-                .map(|o| o.to_string())
-                .collect::<Vec<_>>()
-                .join(",")
-        );
+        let in_list: Vec<Expr> = row_ids.iter().map(|id| lit(*id)).collect();

        let base_query = QueryRequest {
-            filter: Some(QueryFilter::Sql(filter)),
+            filter: Some(QueryFilter::Datafusion(col(ROW_ID).in_list(in_list, false))),
            select: selection,
            with_row_id: true,
            ..Default::default()
--- a/Show More
+++ b/Show More