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

feat: support date, datetime, bytes, and Decimal literals in expr builder (#3235 )
### **Summary** Closes #3212 Extends the Python `lit()` helper to natively support three additional types (`date`, `datetime`, and `Decimal`) and implements reflexive operators for the `Expr` class. This implementation specifically addresses the blocking feedback regarding precision loss, CI discovery, and query engine limitations: * **Logic Refactoring**: Simplified `lit()` by combining `date` and `datetime` normalization into ISO-8601 strings, ensuring stable SQL parsing across different engine locales. * **Precision Preservation**: `decimal.Decimal` objects are now passed as high-precision strings to the Rust bridge, bypassing intermediate float conversions and preserving full 128-bit decimal precision for DataFusion. * **Averted CI Failures**: Temporarily deferred `bytes` literal support to a future PR to resolve a known DataFusion `expr_to_sql` limitation that was crashing the `Doctest` runner. * **Reflexive Operators**: Added support for "literal-first" arithmetic and logical operations (e.g., `10 + col('a')` or `True & col('active')`). Redundant reflexive comparisons (e.g., `__rlt__`) were pruned as Python's data model handles them automatically. * **Integration Verification**: Added dedicated integration tests in the official test directory to ensure the query engine correctly handles the new types and preserves bit-perfect fidelity. ### **Changes** #### [python/python/lancedb/expr.py](file:///c:/Users/Laksh/Documents/lancedb/python/python/lancedb/expr.py) * Updated `lit()` to handle `date`, `datetime`, and `Decimal` natively. * Implemented reflexive operators (`__radd__`, `__rand__`, `__rmul__`, etc.) to support literals on the left-hand side. * Removed the problematic `bytes` doctest example and `lit()` type support to unblock CI. #### [python/src/expr.rs](file:///c:/Users/Laksh/Documents/lancedb/python/src/expr.rs) * Modified the Rust FFI bridge to extract `Decimal` objects as strings. * Ensured the `expr_lit` handler is ready to receive normalized temporal strings. * Consolidated imports and added missing operator documentation. #### [python/python/lancedb/_lancedb.pyi](file:///c:/Users/Laksh/Documents/lancedb/python/python/lancedb/_lancedb.pyi) * Updated type stubs for `expr_lit` to include `Any` (allowing for `Decimal`). ### **Testing** Added several new advanced test cases in [python/python/tests/test_expr.py](file:///c:/Users/Laksh/Documents/lancedb/python/python/tests/test_expr.py) covering: * **High-precision Decimal preservation**: Verified against 128-bit boundaries with a "one point off" test case (`1.234567890123456789 < 1.234567890123456790`). * **Reflexive operator positioning**: Verified successful query construction with literals on the left. * **Timezone-aware normalization**: Confirmed stable behavior for `datetime` objects. * **Integration Testing**: Confirmed Date32 and Decimal columns return the correct Python types and values from the engine during `.to_arrow()` calls. --------- Co-authored-by: Will Jones <willjones127@gmail.com> Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
2026-08-28 00:48:40 +00:00 · 2026-07-06 21:10:10 +00:00 · 2026-07-06 11:30:16 -07:00 · 2026-07-06 05:50:45 -07:00 · 2026-07-04 11:11:10 -07:00 · 2026-07-02 16:54:34 -07:00
69 changed files with 3738 additions and 310 deletions
@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "0.31.0-beta.4"
+current_version = "0.31.0-beta.6"
 parse = """(?x)
    (?P<major>0|[1-9]\\d*)\\.
    (?P<minor>0|[1-9]\\d*)\\.
@@ -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
@@ -170,13 +170,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 +190,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 +244,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 +256,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 +287,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 +303,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 +339,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 +353,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 +398,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
@@ -3423,8 +3423,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.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
 "arrow-array",
 "rand 0.9.4",
@@ -4726,8 +4726,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.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
 "arc-swap",
 "arrow",
@@ -4801,8 +4801,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.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4823,7 +4823,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.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4837,7 +4837,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.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -4846,8 +4846,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.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
 "arrayref",
 "crunchy",
@@ -4857,8 +4857,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.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4896,8 +4896,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.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -4927,8 +4927,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.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -4945,8 +4945,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.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
 "proc-macro2",
 "quote",
@@ -4955,8 +4955,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.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
 "arrow-arith",
 "arrow-array",
@@ -4991,8 +4991,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.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
 "arrow-arith",
 "arrow-array",
@@ -5022,8 +5022,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.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
 "arc-swap",
 "arrow",
@@ -5088,8 +5088,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.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
 "arrow",
 "arrow-arith",
@@ -5130,8 +5130,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.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -5147,8 +5147,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.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
 "arrow",
 "async-trait",
@@ -5160,8 +5160,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.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
 "arrow",
 "arrow-ipc",
@@ -5215,8 +5215,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.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -5231,8 +5231,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.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5271,22 +5271,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.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 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.16"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
 "icu_segmenter",
 "jieba-rs",
@@ -5299,12 +5299,13 @@ dependencies = [

 [[package]]
 name = "lancedb"
-version = "0.31.0-beta.4"
+version = "0.31.0-beta.6"
 dependencies = [
 "ahash",
 "anyhow",
 "arrow",
 "arrow-array",
+ "arrow-buffer",
 "arrow-cast",
 "arrow-data",
 "arrow-ipc",
@@ -5360,6 +5361,7 @@ dependencies = [
 "pin-project",
 "polars",
 "polars-arrow",
+ "pprof 0.14.1",
 "rand 0.9.4",
 "random_word",
 "regex",
@@ -5383,7 +5385,7 @@ dependencies = [

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

 [[package]]
 name = "lancedb-python"
-version = "0.34.0-beta.4"
+version = "0.34.0-beta.6"
 dependencies = [
 "arrow",
 "async-trait",
@@ -5951,9 +5953,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 +5978,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 +5992,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",
@@ -7325,6 +7327,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"
@@ -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.16", default-features = false, "tag" = "v9.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
+lance-core = { "version" = "=9.0.0-beta.16", "tag" = "v9.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
+lance-datagen = { "version" = "=9.0.0-beta.16", "tag" = "v9.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
+lance-file = { "version" = "=9.0.0-beta.16", "tag" = "v9.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
+lance-io = { "version" = "=9.0.0-beta.16", default-features = false, "tag" = "v9.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
+lance-index = { "version" = "=9.0.0-beta.16", "tag" = "v9.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
+lance-linalg = { "version" = "=9.0.0-beta.16", "tag" = "v9.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace = { "version" = "=9.0.0-beta.16", "tag" = "v9.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace-impls = { "version" = "=9.0.0-beta.16", default-features = false, "tag" = "v9.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
+lance-table = { "version" = "=9.0.0-beta.16", "tag" = "v9.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
+lance-testing = { "version" = "=9.0.0-beta.16", "tag" = "v9.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
+lance-datafusion = { "version" = "=9.0.0-beta.16", "tag" = "v9.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
+lance-encoding = { "version" = "=9.0.0-beta.16", "tag" = "v9.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
+lance-arrow = { "version" = "=9.0.0-beta.16", "tag" = "v9.0.0-beta.16", "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"
@@ -119,6 +119,18 @@ ignore = [
    # 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.31.0-beta.6</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
@@ -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
@@ -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.31.0-beta.6</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.31.0-beta.6</version>
    <packaging>pom</packaging>
    <name>${project.artifactId}</name>
    <description>LanceDB Java SDK Parent POM</description>
@@ -28,7 +28,7 @@
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <arrow.version>15.0.0</arrow.version>
-        <lance-core.version>9.0.0-beta.10</lance-core.version>
+        <lance-core.version>9.0.0-beta.16</lance-core.version>
        <spotless.skip>false</spotless.skip>
        <spotless.version>2.30.0</spotless.version>
        <spotless.java.googlejavaformat.version>1.7</spotless.java.googlejavaformat.version>
@@ -1,7 +1,7 @@
 [package]
 name = "lancedb-nodejs"
 edition.workspace = true
-version = "0.31.0-beta.4"
+version = "0.31.0-beta.6"
 publish = false
 license.workspace = true
 description.workspace = true
@@ -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()
@@ -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));
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-darwin-arm64",
-	"version": "0.31.0-beta.4",
+	"version": "0.31.0-beta.6",
 	"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.31.0-beta.6",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-gnu.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-musl",
-	"version": "0.31.0-beta.4",
+	"version": "0.31.0-beta.6",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-musl.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-gnu",
-	"version": "0.31.0-beta.4",
+	"version": "0.31.0-beta.6",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-gnu.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-musl",
-	"version": "0.31.0-beta.4",
+	"version": "0.31.0-beta.6",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-musl.node",
@@ -1,6 +1,6 @@
 {
  "name": "@lancedb/lancedb-win32-arm64-msvc",
-  "version": "0.31.0-beta.4",
+  "version": "0.31.0-beta.6",
  "os": [
    "win32"
  ],
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-win32-x64-msvc",
-	"version": "0.31.0-beta.4",
+	"version": "0.31.0-beta.6",
 	"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.31.0-beta.6",
  "lockfileVersion": 3,
  "requires": true,
  "packages": {
    "": {
      "name": "@lancedb/lancedb",
-      "version": "0.31.0-beta.4",
+      "version": "0.31.0-beta.6",
      "cpu": [
        "x64",
        "arm64"
@@ -11,7 +11,7 @@
    "ann"
  ],
  "private": false,
-  "version": "0.31.0-beta.4",
+  "version": "0.31.0-beta.6",
  "main": "dist/index.js",
  "exports": {
    ".": "./dist/index.js",
@@ -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(),
            )
        })
@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "0.34.0-beta.5"
+current_version = "0.34.0-beta.6"
 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.5"
+version = "0.34.0-beta.6"
 publish = false
 edition.workspace = true
 description = "Python bindings for LanceDB"
@@ -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()
@@ -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
@@ -53,7 +54,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:
@@ -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
@@ -45,11 +45,6 @@ from lancedb._lancedb import (
 )
 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,
@@ -58,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
@@ -389,7 +377,7 @@ def _builds_namespace_natively(
    return namespace_client_impl == "rest" and bool(namespace_client_properties)


-def _supports_native_sync_namespace(namespace_client_impl: str) -> bool:
+def _supports_native_namespace(namespace_client_impl: str) -> bool:
    return namespace_client_impl in {"dir", "rest"}


@@ -412,7 +400,6 @@ class LanceNamespaceDBConnection(DBConnection):
        namespace_client_impl: Optional[str] = None,
        namespace_client_properties: Optional[Dict[str, str]] = None,
        _inner: Optional[AsyncConnection] = None,
-        _route_pushdown_to_rust: Optional[bool] = None,
    ):
        """
        Initialize a namespace-based LanceDB connection.
@@ -454,16 +441,12 @@ 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 = (
-            _route_pushdown_to_rust
-            if _route_pushdown_to_rust is not None
-            else _builds_namespace_natively(
-                namespace_client_impl, namespace_client_properties
-            )
+        # 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
        )
        if _inner is not None:
            self._inner = _inner
@@ -909,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,
@@ -917,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.
@@ -958,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,
@@ -1004,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,
@@ -1071,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
@@ -1093,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,
@@ -1109,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."""
@@ -1128,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,
@@ -1159,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(
@@ -1192,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,
@@ -1226,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]
@@ -1257,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,
@@ -1290,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:
@@ -1310,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


@@ -1360,7 +1337,7 @@ def connect_namespace(
    LanceNamespaceDBConnection
        A namespace-based connection to LanceDB
    """
-    if _supports_native_sync_namespace(namespace_client_impl):
+    if _supports_native_namespace(namespace_client_impl):
        inner = AsyncConnection(
            _connect_namespace(
                namespace_client_impl,
@@ -1384,7 +1361,6 @@ def connect_namespace(
            namespace_client_impl=namespace_client_impl,
            namespace_client_properties=namespace_client_properties,
            _inner=inner,
-            _route_pushdown_to_rust=True,
        )

    namespace_client = namespace_connect(
@@ -1462,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
    )
@@ -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
@@ -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)
@@ -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),
@@ -4020,7 +4030,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()

@@ -242,6 +242,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):
@@ -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 ───────────────────────────────────


@@ -599,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):
@@ -870,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()
@@ -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,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.
@@ -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,
+            )
        })
    }

@@ -1,6 +1,6 @@
 [package]
 name = "lancedb"
-version = "0.31.0-beta.4"
+version = "0.31.0-beta.6"
 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 }
@@ -107,6 +108,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]
@@ -163,6 +166,9 @@ required-features = ["sentence-transformers"]
 name = "bedrock"
 required-features = ["bedrock"]

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

@@ -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(())
+}
@@ -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};
@@ -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()
@@ -35,9 +35,13 @@ pub enum SplitStrategy {
    /// Rows will be randomly assigned to splits
    ///
    /// A seed can be provided to make the assignment deterministic.
+    ///
+    /// A clump_size can be provided to shuffle contiguous groups of rows together,
+    /// preserving I/O locality while still randomising the split assignment.
    Random {
        seed: Option<u64>,
        sizes: SplitSizes,
+        clump_size: Option<u64>,
    },
    /// Rows will be assigned to splits based on the values in the specified columns.
    ///
@@ -323,13 +327,17 @@ impl Splitter {
                self.apply_sequential(source, num_rows, &SplitSizes::Counts(vec![num_rows]))
                    .await
            }
-            SplitStrategy::Random { seed, sizes } => {
+            SplitStrategy::Random {
+                seed,
+                sizes,
+                clump_size,
+            } => {
                let shuffler = Shuffler::new(ShufflerConfig {
                    seed: *seed,
                    // In this case we are only shuffling row ids so we can use a large max_rows_per_file
                    max_rows_per_file: 10 * 1024 * 1024,
                    temp_dir: self.temp_dir.clone(),
-                    clump_size: None,
+                    clump_size: *clump_size,
                });

                let shuffled = shuffler.shuffle(source, num_rows).await?;
@@ -692,6 +700,7 @@ mod tests {
            SplitStrategy::Random {
                seed: Some(42),
                sizes: SplitSizes::Fixed(3),
+                clump_size: None,
            },
        );

@@ -718,6 +727,7 @@ mod tests {
            SplitStrategy::Random {
                seed: Some(42),
                sizes: SplitSizes::Counts(vec![5, 15, 10]),
+                clump_size: None,
            },
        );

@@ -744,6 +754,7 @@ mod tests {
            SplitStrategy::Random {
                seed: Some(42),
                sizes: SplitSizes::Percentages(vec![0.217, 0.168, 0.17]),
+                clump_size: None,
            },
        );

@@ -342,3 +342,9 @@ pub use connection::connect_namespace;
 /// Re-export Lance Session and ObjectStoreRegistry for custom session creation
 pub use lance::session::Session;
 pub use lance_io::object_store::ObjectStoreRegistry;
+
+/// Re-export DataFusion so consumers can build the `Expr` values that public
+/// query/merge APIs (e.g. [`query::QueryBase::only_if_expr`]) accept without
+/// declaring their own (potentially mismatched) direct `datafusion` dependency.
+/// See <https://github.com/lancedb/lancedb/issues/3575>.
+pub use datafusion;
@@ -7,7 +7,7 @@ use std::{future::Future, time::Duration};
 use arrow::compute::concat_batches;
 use arrow_array::{Array, Float16Array, Float32Array, Float64Array, RecordBatch, make_array};
 use arrow_schema::{DataType, SchemaRef};
-use datafusion_expr::Expr;
+use datafusion_expr::{Expr, col, lit};
 use datafusion_physical_plan::ExecutionPlan;
 use futures::{FutureExt, TryFutureExt, TryStreamExt, stream, try_join};
 use half::f16;
@@ -401,6 +401,9 @@ pub trait QueryBase {
    ///
    /// 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
+    /// (i.e. `(previous) AND (new)`) rather than replacing the previous filter.
    fn only_if(self, filter: impl AsRef<str>) -> Self;

    /// Only return rows which match the filter, using an expression builder.
@@ -423,6 +426,9 @@ pub trait QueryBase {
    ///
    /// Note: Expression filters are not supported for remote/server-side queries.
    /// Use [`QueryBase::only_if`] with SQL strings for remote tables.
+    ///
+    /// Calling this multiple times combines the expressions with a logical AND
+    /// rather than replacing the previous filter.
    fn only_if_expr(self, filter: datafusion_expr::Expr) -> Self;

    /// Perform a full text search on the table.
@@ -535,12 +541,13 @@ impl<T: HasQuery> QueryBase for T {
    }

    fn only_if(mut self, filter: impl AsRef<str>) -> Self {
-        self.mut_query().filter = Some(QueryFilter::Sql(filter.as_ref().to_string()));
+        self.mut_query()
+            .add_filter(QueryFilter::Sql(filter.as_ref().to_string()));
        self
    }

    fn only_if_expr(mut self, filter: datafusion_expr::Expr) -> Self {
-        self.mut_query().filter = Some(QueryFilter::Datafusion(filter));
+        self.mut_query().add_filter(QueryFilter::Datafusion(filter));
        self
    }

@@ -716,6 +723,39 @@ pub enum QueryFilter {
    Datafusion(Expr),
 }

+/// Combine two filters with a logical AND.
+///
+/// This is used when a query receives more than one filter (for example when
+/// `where`/`only_if` is called multiple times) so the filters are composed
+/// with AND rather than the later filter silently replacing the earlier one.
+///
+/// SQL string and expression filters are combined within their own
+/// representation. When the two representations are mixed, the expression is
+/// lowered to SQL (via [`crate::expr::expr_to_sql_string`]) and the filters are
+/// combined as SQL strings. Substrait filters cannot be combined and return an
+/// error.
+fn and_filters(existing: QueryFilter, new: QueryFilter) -> Result<QueryFilter> {
+    match (existing, new) {
+        (QueryFilter::Sql(lhs), QueryFilter::Sql(rhs)) => {
+            Ok(QueryFilter::Sql(format!("({lhs}) AND ({rhs})")))
+        }
+        (QueryFilter::Datafusion(lhs), QueryFilter::Datafusion(rhs)) => {
+            Ok(QueryFilter::Datafusion(lhs.and(rhs)))
+        }
+        (QueryFilter::Sql(lhs), QueryFilter::Datafusion(rhs)) => {
+            let rhs = crate::expr::expr_to_sql_string(&rhs)?;
+            Ok(QueryFilter::Sql(format!("({lhs}) AND ({rhs})")))
+        }
+        (QueryFilter::Datafusion(lhs), QueryFilter::Sql(rhs)) => {
+            let lhs = crate::expr::expr_to_sql_string(&lhs)?;
+            Ok(QueryFilter::Sql(format!("({lhs}) AND ({rhs})")))
+        }
+        _ => Err(Error::InvalidInput {
+            message: "cannot combine a Substrait filter with another filter".to_string(),
+        }),
+    }
+}
+
 /// A basic query into a table without any kind of search
 ///
 /// This will result in a (potentially filtered) scan if executed
@@ -730,6 +770,13 @@ pub struct QueryRequest {
    /// Apply filter to the returned rows.
    pub filter: Option<QueryFilter>,

+    /// An error recorded while combining repeated filters that could not be
+    /// composed (see [`QueryRequest::add_filter`]). It is surfaced when the
+    /// query is executed via [`QueryRequest::check_filter`]. We defer the error
+    /// because the builder methods that set filters return `Self` rather than a
+    /// `Result`.
+    pub(crate) filter_error: Option<String>,
+
    /// Perform a full text search on the table.
    pub full_text_search: Option<FullTextSearchQuery>,

@@ -775,6 +822,7 @@ impl Default for QueryRequest {
            limit: None,
            offset: None,
            filter: None,
+            filter_error: None,
            full_text_search: None,
            select: Select::All,
            fast_search: false,
@@ -788,6 +836,41 @@ impl Default for QueryRequest {
    }
 }

+impl QueryRequest {
+    /// Add a filter, combining it with any existing filter using a logical AND.
+    ///
+    /// If the new filter cannot be combined with the existing one (because they
+    /// use different representations) the error is recorded and surfaced later
+    /// by [`Self::check_filter`].
+    pub(crate) fn add_filter(&mut self, new: QueryFilter) {
+        self.filter = Some(match self.filter.take() {
+            None => new,
+            Some(existing) => match and_filters(existing, new) {
+                Ok(combined) => combined,
+                Err(err) => {
+                    // The filters were consumed while attempting to combine
+                    // them; the recorded error is surfaced by `check_filter`
+                    // before the query executes.
+                    self.filter_error = Some(err.to_string());
+                    return;
+                }
+            },
+        });
+    }
+
+    /// Return an error if combining filters failed (see [`Self::add_filter`]).
+    ///
+    /// This must be called by every backend before executing a query.
+    pub(crate) fn check_filter(&self) -> Result<()> {
+        if let Some(message) = &self.filter_error {
+            return Err(Error::InvalidInput {
+                message: message.clone(),
+            });
+        }
+        Ok(())
+    }
+}
+
 /// A builder for LanceDB queries.
 ///
 /// See [`crate::Table::query`] for more details on queries
@@ -1385,18 +1468,13 @@ impl TakeQuery {
    ///
    /// See [`crate::Table::take_offsets`] for more details.
    pub fn from_offsets(parent: Arc<dyn BaseTable>, offsets: Vec<u64>) -> Self {
-        let filter = format!(
-            "_rowoffset in ({})",
-            offsets
-                .iter()
-                .map(|o| o.to_string())
-                .collect::<Vec<_>>()
-                .join(",")
-        );
+        let in_list: Vec<Expr> = offsets.iter().map(|o| lit(*o)).collect();
        Self {
            parent,
            request: QueryRequest {
-                filter: Some(QueryFilter::Sql(filter)),
+                filter: Some(QueryFilter::Datafusion(
+                    col("_rowoffset").in_list(in_list, false),
+                )),
                ..Default::default()
            },
        }
@@ -1406,18 +1484,11 @@ impl TakeQuery {
    ///
    /// See [`crate::Table::take_row_ids`] for more details.
    pub fn from_row_ids(parent: Arc<dyn BaseTable>, row_ids: Vec<u64>) -> Self {
-        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();
        Self {
            parent,
            request: QueryRequest {
-                filter: Some(QueryFilter::Sql(filter)),
+                filter: Some(QueryFilter::Datafusion(col(ROW_ID).in_list(in_list, false))),
                ..Default::default()
            },
        }
@@ -1682,6 +1753,70 @@ mod tests {
        }
    }

+    #[tokio::test]
+    async fn test_repeated_only_if_combines_with_and() {
+        use crate::expr::{col, lit};
+
+        let tmp_dir = tempdir().unwrap();
+        let dataset_path = tmp_dir.path().join("test.lance");
+        let uri = dataset_path.to_str().unwrap();
+
+        let conn = connect(uri).execute().await.unwrap();
+        let table = conn
+            .create_table("my_table", make_non_empty_batches())
+            .execute()
+            .await
+            .unwrap();
+
+        let query = table.query().only_if("id > 0").only_if("id < 100");
+        match &query.request.filter {
+            Some(QueryFilter::Sql(sql)) => assert_eq!(sql, "(id > 0) AND (id < 100)"),
+            other => panic!("expected combined SQL filter, got {other:?}"),
+        }
+
+        // A single filter is left untouched.
+        let query = table.query().only_if("id > 0");
+        match &query.request.filter {
+            Some(QueryFilter::Sql(sql)) => assert_eq!(sql, "id > 0"),
+            other => panic!("expected single SQL filter, got {other:?}"),
+        }
+
+        // Expression filters are combined with a logical AND as well.
+        let query = table
+            .query()
+            .only_if_expr(col("id").gt(lit(0i32)))
+            .only_if_expr(col("id").lt(lit(100i32)));
+        match &query.request.filter {
+            Some(QueryFilter::Datafusion(expr)) => {
+                assert_eq!(
+                    expr,
+                    &col("id").gt(lit(0i32)).and(col("id").lt(lit(100i32)))
+                );
+            }
+            other => panic!("expected combined Datafusion filter, got {other:?}"),
+        }
+
+        // Mixing an SQL string filter with an expression filter lowers the
+        // expression to SQL and combines them as SQL strings.
+        let query = table
+            .query()
+            .only_if("id > 0")
+            .only_if_expr(col("id").lt(lit(100i32)));
+        match &query.request.filter {
+            Some(QueryFilter::Sql(sql)) => {
+                let expected = format!(
+                    "(id > 0) AND ({})",
+                    crate::expr::expr_to_sql_string(&col("id").lt(lit(100i32))).unwrap()
+                );
+                assert_eq!(sql, &expected);
+            }
+            other => panic!("expected combined SQL filter, got {other:?}"),
+        }
+        assert!(query.request.check_filter().is_ok());
+        // The combined filter executes without error.
+        query.execute().await.unwrap();
+    }
+
    #[tokio::test]
    async fn test_select_with_transform() {
        // TODO: Switch back to memory://foo after https://github.com/lancedb/lancedb/issues/1051
@@ -612,6 +612,7 @@ impl<S: HttpSend> RemoteTable<S> {
        body: &mut serde_json::Value,
        params: &QueryRequest,
    ) -> Result<()> {
+        params.check_filter()?;
        body["prefilter"] = params.prefilter.into();
        if let Some(offset) = params.offset {
            body["offset"] = serde_json::Value::Number(serde_json::Number::from(offset));
@@ -35,6 +35,15 @@ pub enum AnyQuery {
    VectorQuery(VectorQueryRequest),
 }

+impl AnyQuery {
+    pub(crate) fn base(&self) -> &QueryRequest {
+        match self {
+            Self::Query(query) => query,
+            Self::VectorQuery(query) => &query.base,
+        }
+    }
+}
+
 //Decide between namespace or local
 pub async fn execute_query(
    table: &NativeTable,
@@ -108,6 +117,7 @@ pub async fn create_plan(
        AnyQuery::VectorQuery(query) => query.clone(),
        AnyQuery::Query(query) => VectorQueryRequest::from_plain_query(query.clone()),
    };
+    query.base.check_filter()?;

    let ds_ref = table.dataset.get().await?;
    let schema = ds_ref.schema();
@@ -357,6 +367,7 @@ async fn execute_namespace_query(

 /// Convert an AnyQuery to the namespace QueryTableRequest format.
 fn convert_to_namespace_query(query: &AnyQuery) -> Result<NsQueryTableRequest> {
+    query.base().check_filter()?;
    match query {
        AnyQuery::VectorQuery(vq) => {
            // Extract the query vector(s)