Bump version: 0.34.0-beta.6 → 0.35.0-beta.0

feat: expose Lance metrics via OpenTelemetry in Python and Node (#3609 )
Bridges Lance's internal `metrics`-crate instrumentation (object store request counts, bytes, latency, errors, and throttles) into OpenTelemetry, in both the Python and Node bindings, with a shared adapter in the Rust core. This is the LanceDB counterpart to lance-format/lance#7537. ## Rust core (`rust/lancedb`) Two new, **off-by-default** features: - `metrics` — re-exports the [`metrics`](https://docs.rs/metrics) crate as `lancedb::metrics` and turns on Lance's object-store instrumentation. Install any `metrics`-compatible recorder to collect them. - `metrics-otel` — adds `lancedb::metrics_otel`, a pull-based adapter that installs a process-global recorder aggregating into lock-free cumulative storage and exposes a snapshot/catalog API (`register_metrics_recorder`, `metrics_catalog`, `snapshot_metrics`, `MetricPoint`/`MetricValue`/`MetricKind`/`MetricDescription`). Both bindings build on this. ## Python `lancedb.otel.instrument_lancedb_metrics()` registers each metric as an OpenTelemetry observable instrument on the given (or global) `MeterProvider`. Available via the `otel` extra (`pip install lancedb[otel]`), which pulls in only `opentelemetry-api` — the application supplies and configures the SDK. ## Node `instrumentLanceDbMetrics()` provides the equivalent wiring against `@opentelemetry/api`. This is the only public entry point; the underlying recorder/catalog/snapshot functions stay internal. Because OpenTelemetry has no asynchronous histogram instrument, histograms are exported Prometheus-style as `<name>_bucket` (with an `le` attribute), `<name>_count`, and `<name>_sum`. Only `_sum` carries the histogram's unit; `_bucket` and `_count` observe cumulative counts and are unitless. The adapter is enabled by default in the Python and Node builds, and off by default in the Rust crate. ## Notes - Requires Lance ≥ `v9.0.0-beta.19`, which ships the object-store metrics APIs (upstream lance-format/lance#7537, now merged). `main` is already on beta.19, so this is a single feature commit with no dependency bump. - Tests: 8 Rust unit tests, 3 Python tests, 2 Node tests, all covering the end-to-end object-store-metrics → OpenTelemetry path. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-09-01 02:58:36 +00:00 · 2026-07-10 05:25:24 +00:00 · 2026-07-09 15:36:03 -07:00 · 2026-07-09 12:39:34 -07:00 · 2026-07-09 11:07:26 -07:00 · 2026-07-09 10:58:41 -07:00
155 changed files with 17367 additions and 2487 deletions
@@ -0,0 +1,137 @@
+---
+name: lancedb-branch-ops
+description: Branch management for LanceDB tables via the REST API. Use this skill whenever someone wants to create, delete, list, or switch branches on a LanceDB table — or needs to make sure a write (metadata update, index build, etc.) lands on a specific branch instead of main. Invoke it even without the word "branch" if context makes clear they want an experimental copy of a table, want to isolate changes, or want to confirm a mutation didn't touch main. Covers: branches/list, branches/create, branches/delete, and passing "branch" in describe/update_field_metadata/create_index to target a non-main version.
+---
+
+## Goal
+
+Manage branches on a LanceDB table: list what exists, create new ones, delete stale ones, and direct read/write operations at a specific branch without touching main.
+
+## Step 0: Establish the connection
+
+Use the `lancedb-connect` skill to resolve the base URL and auth headers (`x-api-key`, `x-lancedb-database`). Skip this only if the connection is already known from the current conversation.
+
+All examples below use `{base_url}` — substitute the resolved endpoint and include the auth headers on every request.
+
+## The branch model (important)
+
+LanceDB branches are named snapshots that diverge from the table's current state at creation time. There is **no checkout command** — you never switch the whole table to a branch. Instead, you **pass `"branch": "<name>"` in the request body** of any operation to target that branch. Omitting the key (or sending an empty body) always targets main.
+
+`branches/list` returns only non-main branches. Main always exists and is not listed.
+
+## List branches
+
+```http
+POST {base_url}/v1/table/{table_id}/branches/list
+Content-Type: application/json
+
+{}
+```
+
+Response:
+```json
+{
+  "branches": {
+    "experiment-reindex": {"parentVersion": 1, "createAt": 1782506085, "manifestSize": 1029}
+  }
+}
+```
+
+If `branches` is `{}`, the table has no branches besides main.
+
+## Create a branch
+
+```http
+POST {base_url}/v1/table/{table_id}/branches/create
+Content-Type: application/json
+
+{"name": "experiment-reindex"}
+```
+
+HTTP 200 with `{}` body = success. The branch is created off the table's current state on main.
+
+Verify by calling `branches/list` and confirming the new name appears.
+
+## Delete a branch
+
+```http
+POST {base_url}/v1/table/{table_id}/branches/delete
+Content-Type: application/json
+
+{"name": "stale-2024"}
+```
+
+HTTP 200 with `{}` body = success. Only the branch pointer is removed — main and all row data remain intact.
+
+Verify by calling `branches/list` (name gone) and `describe` with no branch param (main still responds).
+
+## Operate on a specific branch
+
+Pass `"branch": "<name>"` in the body of any operation to scope it to that branch:
+
+**Read schema on a branch:**
+```http
+POST {base_url}/v1/table/{table_id}/describe
+Content-Type: application/json
+
+{"branch": "wip-branch"}
+```
+
+**Write metadata to a branch (not main):**
+```http
+POST {base_url}/v1/table/{table_id}/update_field_metadata
+Content-Type: application/json
+
+{
+  "branch": "wip-branch",
+  "updates": [
+    {
+      "path": "category",
+      "metadata": {"lancedb:description": "Product category label."},
+      "replace": false
+    }
+  ]
+}
+```
+
+**Build an index on a branch:**
+```http
+POST {base_url}/v1/table/{table_id}/create_index
+Content-Type: application/json
+
+{
+  "branch": "wip-branch",
+  "column": "category",
+  "index_type": "BTREE"
+}
+```
+
+## Verifying isolation
+
+After writing to a branch, always confirm the change did NOT land on main:
+
+```bash
+# Should show the new metadata
+curl -s -X POST {base_url}/v1/table/{table_id}/describe \
+  -H "x-api-key: <key>" -H "x-lancedb-database: <db>" \
+  -H "content-type: application/json" \
+  -d '{"branch": "wip-branch"}'
+
+# Should NOT show the new metadata
+curl -s -X POST {base_url}/v1/table/{table_id}/describe \
+  -H "x-api-key: <key>" -H "x-lancedb-database: <db>" \
+  -H "content-type: application/json" \
+  -d '{}'
+```
+
+## Quick reference
+
+| Goal | Endpoint | Body |
+|------|----------|------|
+| List all branches | `branches/list` | `{}` |
+| Create a branch | `branches/create` | `{"name": "..."}` |
+| Delete a branch | `branches/delete` | `{"name": "..."}` |
+| Read schema on branch | `describe` | `{"branch": "..."}` |
+| Write metadata on branch | `update_field_metadata` | `{"branch": "...", "updates": [...]}` |
+| Build index on branch | `create_index` | `{"branch": "...", "column": ..., "index_type": ...}` |
+| Target main (default) | any endpoint | omit `"branch"` key |
@@ -0,0 +1,178 @@
+---
+name: lancedb-column-metadata
+description: Column metadata authoring for LanceDB tables via the REST API. This skill is required for tasks like writing field descriptions, setting tags on columns (field_type, model, project_id, version), classifying columns as embeddings vs labels vs eval metrics, or grouping versioned columns into logical families — because it has the API integration needed to read the schema and persist metadata back. Invoke whenever someone wants to document, annotate, tag, or classify what their table columns ARE. Trigger even without an explicit "LanceDB" mention, as long as the context is column-level documentation or tagging for an ML or vector database table.
+metadata:
+  short-description: Write column descriptions, tags, and logical groupings to a LanceDB table
+---
+
+## Overview
+
+This skill authors column-level metadata for a LanceDB table. It connects to a LanceDB deployment over its REST API, inspects the table schema, generates appropriate metadata, and writes it back.
+
+## Step 0: Establish the connection
+
+Use the `lancedb-connect` skill (invoke it via the Skill tool) to resolve the base URL and auth headers (`x-api-key`, `x-lancedb-database`) for whichever deployment the user is working against — enterprise/self-hosted or a local dev server. Skip it only if the connection details are already established in the conversation.
+
+All examples below use `{base_url}` — substitute the resolved endpoint and include the resolved headers on every request.
+
+## Metadata keys
+
+All metadata uses namespaced keys:
+
+| Key | Purpose | Example value |
+|-----|---------|---------------|
+| `lancedb:description` | Human-readable explanation of what the column contains | `"CLIP ViT-L/14 image embedding, L2-normalized (768-dim)"` |
+| `lancedb:tag:<name>` | Flexible key-value tag; the suffix names the tag category | `lancedb:tag:field_type: "embedding"`, `lancedb:tag:model: "clip"`, `lancedb:tag:project_id: "foo"` |
+| `lancedb:logical-column` | Logical group/family this column belongs to | `"clip_features"` |
+
+Tags are open-ended — use whatever key suffix and value make sense given the user's intent. The tag suffix should describe *what is being classified* (e.g., `field_type`, `model`, `project_id`) and the value describes *how*.
+
+## Step 1: Resolve the table identifier
+
+You need:
+- **Table name** (required) — e.g., `my_table` or `my_namespace.my_table`
+- **Database name** — ask if not provided and not inferable from context; it goes in the `x-lancedb-database` header, never in the URL path
+
+The table identifier in the URL path is typically `table_name` for a top-level table, or `namespace$table_name` if the table lives in a namespace. The API accepts a `delimiter` query parameter to parse compound identifiers (default `$`).
+
+## Step 2: Describe the table
+
+```http
+POST {base_url}/v1/table/{table_id}/describe
+Content-Type: application/json
+
+{}
+```
+
+The response contains `schema.fields` — an array of field objects:
+
+```json
+{
+  "schema": {
+    "fields": [
+      {
+        "name": "clip_embedding_v3",
+        "type": { "type": "FixedSizeList", "fields": [...], "listSize": 768 },
+        "nullable": true,
+        "metadata": { "lancedb:description": "..." }
+      }
+    ]
+  }
+}
+```
+
+Each field has:
+- `name` — field name
+- `type` — Arrow data type (check `type.type` for the type string)
+- `nullable` — boolean
+- `metadata` — existing key-value metadata (read this before writing to avoid redundant updates)
+
+For struct/nested fields, recurse into `type.fields` and represent them as dot-notation paths (e.g., `parent.child`).
+
+If the user hasn't specified which columns to update, work with all columns.
+
+## Step 3: Generate metadata
+
+Decide what to generate based on the user's request.
+
+### Writing descriptions (`lancedb:description`)
+
+Base descriptions on:
+- The column name and Arrow type (e.g., `FixedSizeList` of floats → likely an embedding)
+- User-supplied context (upstream pipeline, sample values, domain knowledge)
+- Name patterns: `_embedding`/`_vec`/`_embed` → vector; `_label`/`_class` → label; `_score`/`_eval`/`_metric` → evaluation metric
+
+Be specific and concise. Good: `"Sentence-BERT embedding of the query text (768-dim)."` Not: `"An embedding column."`
+
+### Tagging columns (`lancedb:tag:<name>`)
+
+Choose tag key names that match what the user asked to annotate. Common patterns:
+
+- Semantic field type → `lancedb:tag:field_type: "embedding"` / `"text"` / `"image"` / `"label"` / `"eval"` / `"id"` / `"metadata"`
+- Model or source → `lancedb:tag:model: "clip"` / `"bert"` / `"vit"`
+- Project affiliation → `lancedb:tag:project_id: "<name>"`
+- Version → `lancedb:tag:version: "v3"` (and `lancedb:tag:latest: "true"` for the newest)
+
+Use Arrow type as a hint: `FixedSizeList` + float → embedding; `Utf8`/`LargeUtf8` → text; `Binary` → image or blob.
+
+Multiple tags on the same column are fine — each is a separate key.
+
+### Grouping into logical columns (`lancedb:logical-column`)
+
+Look for naming patterns across columns:
+- `clip_v1`, `clip_v2`, `clip_v3` → logical column `"clip"`, latest is `v3`
+- `text_embed_20240101`, `text_embed_20240601` → logical column `"text_embed"`, latest is the most recent date suffix
+
+Write `lancedb:logical-column` on all members of a group. Mark the newest with `lancedb:tag:latest: "true"` (in addition to its version tag).
+
+## Step 4: Write the metadata
+
+```http
+POST {base_url}/v1/table/{table_id}/update_field_metadata
+Content-Type: application/json
+
+{
+  "updates": [
+    {
+      "path": "clip_v3",
+      "metadata": {
+        "lancedb:description": "CLIP ViT-L/14 image embedding, L2-normalized (1024-dim).",
+        "lancedb:tag:field_type": "embedding",
+        "lancedb:tag:model": "clip",
+        "lancedb:tag:version": "v3",
+        "lancedb:tag:latest": "true",
+        "lancedb:logical-column": "clip"
+      },
+      "replace": false
+    },
+    {
+      "path": "clip_v2",
+      "metadata": {
+        "lancedb:description": "CLIP ViT-B/32 image embedding (768-dim), superseded by v3.",
+        "lancedb:tag:field_type": "embedding",
+        "lancedb:tag:model": "clip",
+        "lancedb:tag:version": "v2",
+        "lancedb:logical-column": "clip"
+      },
+      "replace": false
+    }
+  ]
+}
+```
+
+Rules:
+- **Use `"replace": false`** (merge) by default — this preserves existing metadata the user didn't ask to change
+- Use `"replace": true` only if the user explicitly asks to overwrite all existing metadata on a column
+- Set a value to `null` to delete a specific key
+- Batch all updates in a single request when possible
+
+The response includes `version` (new table version) and `fields` (the updated metadata per field).
+
+## Step 5: Confirm
+
+Report back:
+- Which columns were updated and what was written
+- The new table version number
+- Any columns skipped (e.g., already had up-to-date metadata)
+
+---
+
+## Quick examples
+
+**"Write descriptions for all columns in the `product_embeddings` table"**
+1. POST `/v1/table/product_embeddings/describe` → get all fields
+2. Generate a `lancedb:description` for each column based on name + type
+3. POST `update_field_metadata` with descriptions
+4. Report
+
+**"Tag the columns in `model_outputs` with their field type and model"**
+1. Describe `model_outputs`
+2. For each field, classify by name + Arrow type → set `lancedb:tag:field_type` and `lancedb:tag:model` where applicable
+3. POST `update_field_metadata`
+4. Report
+
+**"Group the feature columns in `training_features` into logical families and mark the latest version"**
+1. Describe the table
+2. Find version patterns → assign `lancedb:logical-column` and `lancedb:tag:version`; mark newest with `lancedb:tag:latest: "true"`
+3. POST `update_field_metadata`
+4. Show the grouping
@@ -0,0 +1,42 @@
+---
+name: lancedb-connect
+description: Resolve how to connect to a LanceDB deployment over the REST API — figure out the base URL, API key, and database header. Use this before making any REST requests to a LanceDB table, whenever the endpoint or auth setup is not already known. Also useful on its own when someone asks how to connect, authenticate, or curl their LanceDB instance.
+metadata:
+  short-description: Resolve the base URL and auth headers for a LanceDB deployment
+---
+
+## Goal
+
+Produce two things every REST request needs:
+
+1. **Base URL** — the endpoint
+2. **Headers** — `x-api-key`, and usually `x-lancedb-database`
+
+## Resolution steps
+
+1. If the user already gave a URL and API key (or said which environment they're working against), use that.
+2. Otherwise, look for credentials already available in the environment:
+   - Env vars like `LANCEDB_URI` / `LANCEDB_HOST` / `LANCEDB_API_KEY`
+   - A LanceDB endpoint already running or port-forwarded locally (the REST default port is 2333, i.e. `http://localhost:2333`)
+3. If you didn't find both pieces, ask the user directly: **"What's your LanceDB endpoint's URL, and what's your API key?"** Also ask which database to use if it isn't obvious. Don't guess or probe further — the user knows their deployment.
+
+## Validating the connection
+
+Make a cheap authenticated request and check the status:
+
+```bash
+curl -s -w "\n%{http_code}" "{base_url}/v1/table/?limit=1" \
+  -H "x-api-key: <key>" \
+  -H "x-lancedb-database: <database>"
+```
+
+- `200` — connection, key, and database header all good
+- `401` — API key missing or wrong
+- `400` mentioning a database header — this deployment expects `x-lancedb-database`
+
+## Non-REST equivalents
+
+If the caller would rather use the SDK or CLI than raw REST, the same credentials work:
+
+- Python SDK: `lancedb.connect("db://<database>", api_key="<key>", host_override="<base_url>")`
+- `lancedb` CLI: a `[profiles.<name>]` entry in `~/.lancedb/config.toml` with `http_server_url`, `api_key`, `database`
@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "0.30.1-beta.2"
+current_version = "0.31.0-beta.6"
 parse = """(?x)
    (?P<major>0|[1-9]\\d*)\\.
    (?P<minor>0|[1-9]\\d*)\\.
@@ -23,6 +23,8 @@ allow_dirty = true
 commit = true
 message = "Bump version: {current_version} → {new_version}"
 commit_args = ""
+# bump-my-version >=1.4.0 rejects pre_commit_hooks containing shell syntax unless opted in.
+allow_shell_hooks = true

 # Java maven files
 pre_commit_hooks = [
@@ -0,0 +1,21 @@
+# CODEOWNERS
+#
+# These owners will be the default owners for everything in the repo.
+# They will be requested for review when someone opens a pull request.
+#
+# See https://docs.github.com/en/repositories/managing-your-repositorys-settings-and-features/customizing-your-repository/about-code-owners
+
+# Default owners for everything
+* @jackye1995 @wjones127
+
+# Release and publish workflows — changes here can affect supply chain security
+/.github/workflows/ @jackye1995 @wjones127 @Xuanwo
+
+# Remote client and auth — sensitive networking and auth code
+/rust/lancedb/src/remote/ @jackye1995 @wjones127
+
+# Python FFI boundary
+/python/src/ @jackye1995 @wjones127 @AyushExel
+
+# NodeJS FFI boundary
+/nodejs/src/ @jackye1995 @wjones127
@@ -34,15 +34,16 @@ runs:
        maturin-version: "1.12.4"
        command: build
        working-directory: python
-        docker-options: "-e PIP_EXTRA_INDEX_URL='https://pypi.fury.io/lance-format/ https://pypi.fury.io/lancedb/'"
+        docker-options: "-e PIP_EXTRA_INDEX_URL='https://pypi.fury.io/lance-format/ https://pypi.fury.io/lancedb/' -e PROTOC=/usr/local/bin/protoc"
        target: x86_64-unknown-linux-gnu
        manylinux: ${{ inputs.manylinux }}
        args: ${{ inputs.args }}
        before-script-linux: |
          set -e
-          curl -L https://github.com/protocolbuffers/protobuf/releases/download/v24.4/protoc-24.4-linux-$(uname -m).zip > /tmp/protoc.zip \
-            && unzip /tmp/protoc.zip -d /usr/local \
-            && rm /tmp/protoc.zip
+          curl -fsSL https://github.com/protocolbuffers/protobuf/releases/download/v24.4/protoc-24.4-linux-x86_64.zip -o /tmp/protoc.zip
+          unzip /tmp/protoc.zip -d /usr/local
+          rm /tmp/protoc.zip
+          /usr/local/bin/protoc --version
    - name: Build Arm Manylinux Wheel
      if: ${{ inputs.arm-build == 'true' }}
      uses: PyO3/maturin-action@v1
@@ -50,13 +51,14 @@ runs:
        maturin-version: "1.12.4"
        command: build
        working-directory: python
-        docker-options: "-e PIP_EXTRA_INDEX_URL='https://pypi.fury.io/lance-format/ https://pypi.fury.io/lancedb/'"
+        docker-options: "-e PIP_EXTRA_INDEX_URL='https://pypi.fury.io/lance-format/ https://pypi.fury.io/lancedb/' -e PROTOC=/usr/local/bin/protoc"
        target: aarch64-unknown-linux-gnu
        manylinux: ${{ inputs.manylinux }}
        args: ${{ inputs.args }}
        before-script-linux: |
          set -e
-          yum install -y clang \
-            && curl -L https://github.com/protocolbuffers/protobuf/releases/download/v24.4/protoc-24.4-linux-aarch_64.zip > /tmp/protoc.zip \
-            && unzip /tmp/protoc.zip -d /usr/local \
-            && rm /tmp/protoc.zip
+          yum install -y clang
+          curl -fsSL https://github.com/protocolbuffers/protobuf/releases/download/v24.4/protoc-24.4-linux-aarch_64.zip -o /tmp/protoc.zip
+          unzip /tmp/protoc.zip -d /usr/local
+          rm /tmp/protoc.zip
+          /usr/local/bin/protoc --version
@@ -25,7 +25,7 @@ jobs:
    # Only runs on tags that matches the make-release action
    if: startsWith(github.ref, 'refs/tags/v')
    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
      - uses: Swatinem/rust-cache@v2
        with:
          workspaces: rust
@@ -47,7 +47,7 @@ jobs:
      contents: read
      issues: write
    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
      - uses: ./.github/actions/create-failure-issue
        with:
          job-results: ${{ toJSON(needs) }}
@@ -36,14 +36,14 @@ jobs:
          echo "guidelines = ${{ inputs.guidelines }}"

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

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

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

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

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

@@ -27,7 +27,7 @@ jobs:
    name: Verify PR title / description conforms to semantic-release
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/setup-node@v4
+      - uses: actions/setup-node@v6
        with:
          node-version: "18"
      # These rules are disabled because Github will always ensure there
@@ -35,7 +35,7 @@ jobs:
    runs-on: ubuntu-24.04
    steps:
      - name: Checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@v6
      - name: Install dependencies needed for ubuntu
        run: |
          sudo apt install -y protobuf-compiler libssl-dev
@@ -53,7 +53,7 @@ jobs:
          python -m pip install --extra-index-url https://pypi.fury.io/lance-format/ --extra-index-url https://pypi.fury.io/lancedb/ -e .
          python -m pip install --extra-index-url https://pypi.fury.io/lance-format/ --extra-index-url https://pypi.fury.io/lancedb/ -r ../docs/requirements.txt
      - name: Set up node
-        uses: actions/setup-node@v4
+        uses: actions/setup-node@v6
        with:
          node-version: 20
          cache: 'npm'
@@ -32,7 +32,7 @@ jobs:
        working-directory: ./java
    steps:
      - name: Checkout repository
-        uses: actions/checkout@v4
+        uses: actions/checkout@v6
      - name: Set up Java 8
        uses: actions/setup-java@v4
        with:
@@ -73,7 +73,7 @@ jobs:
      contents: read
      issues: write
    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
      - uses: ./.github/actions/create-failure-issue
        with:
          job-results: ${{ toJSON(needs) }}
@@ -36,7 +36,7 @@ jobs:
        working-directory: ./java
    steps:
      - name: Checkout repository
-        uses: actions/checkout@v4
+        uses: actions/checkout@v6
      - name: Set up Java 17
        uses: actions/setup-java@v4
        with:
@@ -19,7 +19,7 @@ jobs:
    runs-on: ubuntu-latest
    steps:
      - name: Check out code
-        uses: actions/checkout@v4
+        uses: actions/checkout@v6
      - name: Install license-header-checker
        working-directory: /tmp
        run: |
@@ -49,7 +49,7 @@ jobs:
    steps:
      - name: Output Inputs
        run: echo "${{ toJSON(github.event.inputs) }}"
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
        with:
          fetch-depth: 0
          lfs: true
@@ -38,14 +38,14 @@ jobs:
      CC: gcc-12
      CXX: g++-12
    steps:
-    - uses: actions/checkout@v4
+    - uses: actions/checkout@v6
      with:
        fetch-depth: 0
        lfs: true
-    - uses: pnpm/action-setup@v4
+    - uses: pnpm/action-setup@v6
      with:
        version: 11.1.1
-    - uses: actions/setup-node@v4
+    - uses: actions/setup-node@v6
      with:
        # pnpm 11 requires Node >= 22.13; use 24 since 22 hits EOL
        # in October. The library itself still supports Node >= 18
@@ -86,14 +86,14 @@ jobs:
        shell: bash
        working-directory: nodejs
    steps:
-    - uses: actions/checkout@v4
+    - uses: actions/checkout@v6
      with:
        fetch-depth: 0
        lfs: true
-    - uses: pnpm/action-setup@v4
+    - uses: pnpm/action-setup@v6
      with:
        version: 11.1.1
-    - uses: actions/setup-node@v4
+    - uses: actions/setup-node@v6
      name: Setup Node.js 24 for build
      with:
        # pnpm 11 requires Node >= 22.13; use 24 since 22 hits EOL
@@ -130,7 +130,7 @@ jobs:
          echo "Run 'pnpm run docs', fix any warnings, and commit the changes."
          exit 1
        fi
-    - uses: actions/setup-node@v4
+    - uses: actions/setup-node@v6
      name: Setup Node.js ${{ matrix.node-version }} for test
      with:
        node-version: ${{ matrix.node-version }}
@@ -166,14 +166,14 @@ jobs:
        shell: bash
        working-directory: nodejs
    steps:
-    - uses: actions/checkout@v4
+    - uses: actions/checkout@v6
      with:
        fetch-depth: 0
        lfs: true
-    - uses: pnpm/action-setup@v4
+    - uses: pnpm/action-setup@v6
      with:
        version: 11.1.1
-    - uses: actions/setup-node@v4
+    - uses: actions/setup-node@v6
      with:
        # pnpm 11 requires Node >= 22.13; use 24 since 22 hits EOL
        # in October.
@@ -32,7 +32,7 @@ jobs:
    permissions:
      contents: write
    steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
        with:
          fetch-depth: 0
          lfs: true
@@ -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
@@ -13,24 +13,25 @@ categories = ["database-implementations"]
 rust-version = "1.91.0"

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

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

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

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

 # ---------------------------------------------------------------------------
@@ -14,7 +14,7 @@ Add the following dependency to your `pom.xml`:
 <dependency>
    <groupId>com.lancedb</groupId>
    <artifactId>lancedb-core</artifactId>
-    <version>0.30.1-beta.2</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
@@ -295,6 +295,23 @@ await table.createIndex("my_float_col");

 ***

+### currentBranch()
+
+```ts
+abstract currentBranch(): null | string
+```
+
+The branch this table handle is scoped to, or `null` for the main branch.
+
+A handle returned by [Branches.create](Branches.md#create) or [Branches.checkout](Branches.md#checkout)
+reports the branch it targets; a handle opened normally reports `null`.
+
+#### Returns
+
+`null` \| `string`
+
+***
+
 ### delete()

 ```ts
@@ -381,6 +398,26 @@ Drop an index from the table.

 ***

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

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

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

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

 - [FullTextQueryType](enumerations/FullTextQueryType.md)
+- [OAuthFlowType](enumerations/OAuthFlowType.md)
 - [Occur](enumerations/Occur.md)
 - [Operator](enumerations/Operator.md)

@@ -85,6 +86,8 @@
 - [ListNamespacesResponse](interfaces/ListNamespacesResponse.md)
 - [LsmWriteSpec](interfaces/LsmWriteSpec.md)
 - [MergeResult](interfaces/MergeResult.md)
+- [NativeOAuthConfig](interfaces/NativeOAuthConfig.md)
+- [OAuthConfig](interfaces/OAuthConfig.md)
 - [OpenTableOptions](interfaces/OpenTableOptions.md)
 - [OptimizeOptions](interfaces/OptimizeOptions.md)
 - [OptimizeStats](interfaces/OptimizeStats.md)
@@ -128,6 +131,7 @@
 - [RecordBatchIterator](functions/RecordBatchIterator.md)
 - [connect](functions/connect.md)
 - [connectNamespace](functions/connectNamespace.md)
+- [instrumentLanceDbMetrics](functions/instrumentLanceDbMetrics.md)
 - [makeArrowTable](functions/makeArrowTable.md)
 - [packBits](functions/packBits.md)
 - [permutationBuilder](functions/permutationBuilder.md)
@@ -64,6 +64,19 @@ client used by manifest-enabled native connections.

 ***

+### oauthConfig?
+
+```ts
+optional oauthConfig: NativeOAuthConfig;
+```
+
+(For LanceDB cloud only): OAuth configuration for IdP-based
+authentication (e.g., Azure Entra ID). When set, token acquisition
+and refresh are handled entirely in Rust. TypeScript users should pass
+the public `OAuthConfig` type exported from `@lancedb/lancedb`.
+
+***
+
 ### readConsistencyInterval?

 ```ts
@@ -23,6 +23,31 @@ be more columns to represent composite indices.

 ***

+### createdAt?
+
+```ts
+optional createdAt: Date;
+```
+
+When the index was created.
+
+`undefined` for remote tables or indices created before timestamps were tracked.
+
+***
+
+### indexDetails?
+
+```ts
+optional indexDetails: any;
+```
+
+Index-type-specific details parsed as a JavaScript object.
+
+Falls back to a raw string if JSON parsing fails. `undefined` for
+remote tables or when details are unavailable.
+
+***
+
 ### indexType

 ```ts
@@ -33,6 +58,30 @@ The type of the index

 ***

+### indexUuid?
+
+```ts
+optional indexUuid: string;
+```
+
+The UUID of the first segment of the index.
+
+`undefined` for remote tables, which do not yet surface this.
+
+***
+
+### indexVersion?
+
+```ts
+optional indexVersion: number;
+```
+
+The on-disk index format version.
+
+`undefined` for remote tables.
+
+***
+
 ### name

 ```ts
@@ -40,3 +89,63 @@ name: string;
 ```

 The name of the index
+
+***
+
+### numIndexedRows?
+
+```ts
+optional numIndexedRows: number;
+```
+
+The number of rows indexed, across all segments.
+
+`undefined` for remote tables.
+
+***
+
+### numSegments?
+
+```ts
+optional numSegments: number;
+```
+
+The number of segments that make up the index.
+
+`undefined` for remote tables.
+
+***
+
+### numUnindexedRows?
+
+```ts
+optional numUnindexedRows: number;
+```
+
+The number of rows not yet covered by this index.
+
+`undefined` for remote tables.
+
+***
+
+### sizeBytes?
+
+```ts
+optional sizeBytes: number;
+```
+
+The total size in bytes of all index files across all segments.
+
+`undefined` for remote tables or indices without size tracking.
+
+***
+
+### typeUrl?
+
+```ts
+optional typeUrl: string;
+```
+
+The protobuf type URL, a precise type identifier for the index.
+
+`undefined` for remote tables.
@@ -0,0 +1,88 @@
+[**@lancedb/lancedb**](../README.md) • **Docs**
+
+***
+
+[@lancedb/lancedb](../globals.md) / NativeOAuthConfig
+
+# Interface: NativeOAuthConfig
+
+OAuth configuration for LanceDB authentication.
+
+This is the generated napi-rs binding shape. TypeScript users should prefer
+the public `OAuthConfig` type exported from `@lancedb/lancedb`.
+
+All token acquisition and refresh is handled in the Rust layer.
+
+## Properties
+
+### clientId
+
+```ts
+clientId: string;
+```
+
+Application / Client ID.
+
+***
+
+### clientSecret?
+
+```ts
+optional clientSecret: string;
+```
+
+Client secret (required for client_credentials).
+
+***
+
+### flow?
+
+```ts
+optional flow: string;
+```
+
+Authentication flow: "client_credentials" or "azure_managed_identity"
+
+***
+
+### issuerUrl
+
+```ts
+issuerUrl: string;
+```
+
+OIDC issuer URL or OAuth authority URL.
+For Azure: `https://login.microsoftonline.com/{tenant_id}/v2.0`
+
+***
+
+### managedIdentityClientId?
+
+```ts
+optional managedIdentityClientId: string;
+```
+
+Client ID for user-assigned managed identity (azure_managed_identity).
+
+***
+
+### refreshBufferSecs?
+
+```ts
+optional refreshBufferSecs: number;
+```
+
+Seconds before expiry to trigger proactive refresh (default: 300).
+Keep this well below the token TTL; if it is greater than or equal to
+the TTL, each request refreshes the token.
+
+***
+
+### scopes
+
+```ts
+scopes: string[];
+```
+
+OAuth scopes to request. For Azure managed identity, exactly one scope
+or resource is required. For example: `["api://{app_id}/.default"]`
@@ -0,0 +1,111 @@
+[**@lancedb/lancedb**](../README.md) • **Docs**
+
+***
+
+[@lancedb/lancedb](../globals.md) / OAuthConfig
+
+# Interface: OAuthConfig
+
+OAuth configuration for LanceDB authentication.
+
+This is the public TypeScript OAuth configuration type. The generated
+`NativeOAuthConfig` type has the same runtime shape but is an implementation
+detail of the napi-rs binding.
+
+All token acquisition and refresh is handled in the Rust layer.
+This config is passed through to Rust via napi-rs.
+
+## Examples
+
+```typescript
+const config: OAuthConfig = {
+  issuerUrl: "https://login.microsoftonline.com/{tenant}/v2.0",
+  clientId: "app-id",
+  clientSecret: "secret",
+  scopes: ["api://lancedb-api/.default"],
+};
+```
+
+```typescript
+const config: OAuthConfig = {
+  issuerUrl: "https://login.microsoftonline.com/{tenant}/v2.0",
+  clientId: "app-id",
+  scopes: ["api://lancedb-api/.default"],
+  flow: OAuthFlowType.AzureManagedIdentity,
+};
+```
+
+## Properties
+
+### clientId
+
+```ts
+clientId: string;
+```
+
+Application / Client ID.
+
+***
+
+### clientSecret?
+
+```ts
+optional clientSecret: string;
+```
+
+Client secret (required for ClientCredentials).
+
+***
+
+### flow?
+
+```ts
+optional flow: OAuthFlowType;
+```
+
+Authentication flow (default: ClientCredentials).
+
+***
+
+### issuerUrl
+
+```ts
+issuerUrl: string;
+```
+
+OIDC issuer URL or OAuth authority URL.
+For Azure: `https://login.microsoftonline.com/{tenant_id}/v2.0`
+
+***
+
+### managedIdentityClientId?
+
+```ts
+optional managedIdentityClientId: string;
+```
+
+Client ID for user-assigned managed identity (AzureManagedIdentity).
+
+***
+
+### refreshBufferSecs?
+
+```ts
+optional refreshBufferSecs: number;
+```
+
+Seconds before expiry to trigger proactive refresh (default: 300).
+Keep this well below the token TTL; if it is greater than or equal to
+the TTL, each request refreshes the token.
+
+***
+
+### scopes
+
+```ts
+scopes: string[];
+```
+
+OAuth scopes to request.
+For Azure managed identity, exactly one scope or resource is required.
+For example: `["api://{app_id}/.default"]`
@@ -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.30.1-beta.2</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.30.1-beta.2</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>8.0.0-beta.9</lance-core.version>
+        <lance-core.version>9.0.0-beta.19</lance-core.version>
        <spotless.skip>false</spotless.skip>
        <spotless.version>2.30.0</spotless.version>
        <spotless.java.googlejavaformat.version>1.7</spotless.java.googlejavaformat.version>
@@ -1,7 +1,7 @@
 [package]
 name = "lancedb-nodejs"
 edition.workspace = true
-version = "0.30.1-beta.2"
+version = "0.31.0-beta.6"
 publish = false
 license.workspace = true
 description.workspace = true
@@ -25,8 +25,12 @@ lancedb = { path = "../rust/lancedb", default-features = false }
 lance-namespace.workspace = true
 napi = { version = "3.8.3", default-features = false, features = [
    "napi9",
-    "async"
+    "async",
+    "chrono_date",
+    "serde-json",
 ] }
+chrono = { version = "0.4", default-features = false, features = ["clock"] }
+serde_json = "1"
 napi-derive = "3.5.2"
 # Prevent dynamic linking of lzma, which comes from datafusion
 lzma-sys = { version = "0.1", features = ["static"] }
@@ -40,6 +44,6 @@ aws-lc-rs = "=1.16.3"
 napi-build = "2.3.1"

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

+  it("should combine repeated where clauses with AND", async () => {
+    const results = await table
+      .query()
+      .where("score > 1.0")
+      .where("score < 3.0")
+      .orderBy({ columnName: "score" })
+      .toArray();
+    // Only rows matching both predicates should be returned, rather than the
+    // second where() silently replacing the first.
+    expect(results.length).toBe(2);
+    expect(results[0].score).toBeCloseTo(1.2, 0.001);
+    expect(results[1].score).toBeCloseTo(2.8, 0.001);
+  });
+
  it("should support method chaining with limit", async () => {
    const results = await table
      .query()
@@ -191,30 +191,36 @@ describe("remote connection", () => {
    );
  });

-  it("allows version on remote but rejects a non-main branch", async () => {
+  it("supports version time-travel and branches on remote", async () => {
    await withMockDatabase(
-      (_req, res) => {
-        // describe (table open + version validation) always succeeds
-        const body = JSON.stringify({
-          name: "t",
-          version: 2,
-          schema: { fields: [] },
-        });
+      (req, res) => {
+        const body = req.url?.includes("/branches/list")
+          ? JSON.stringify({
+              branches: {
+                exp: { parentVersion: 1, createAt: 1, manifestSize: 1 },
+              },
+            })
+          : JSON.stringify({ name: "t", version: 2, schema: { fields: [] } });
        res.writeHead(200, { "Content-Type": "application/json" }).end(body);
      },
      async (db) => {
-        // version-only (and "main" + version) is allowed: remote supports
-        // version time-travel even though it has no branches
-        await db.openTable("t", undefined, { version: 2 });
-        await db.openTable("t", undefined, { branch: "main", version: 2 });
+        // version-only (and "main" + version) time-travel the main chain
+        const v2 = await db.openTable("t", undefined, { version: 2 });
+        expect(v2.currentBranch()).toBeNull();
+        const mainV2 = await db.openTable("t", undefined, {
+          branch: "main",
+          version: 2,
+        });
+        expect(mainV2.currentBranch()).toBeNull();

-        // a non-main branch is rejected, with or without a version
-        await expect(
-          db.openTable("t", undefined, { branch: "exp" }),
-        ).rejects.toThrow(/branching/);
-        await expect(
-          db.openTable("t", undefined, { branch: "exp", version: 2 }),
-        ).rejects.toThrow(/branching/);
+        // a non-main branch opens a handle scoped to that branch
+        const exp = await db.openTable("t", undefined, { branch: "exp" });
+        expect(exp.currentBranch()).toBe("exp");
+        const expV2 = await db.openTable("t", undefined, {
+          branch: "exp",
+          version: 2,
+        });
+        expect(expV2.currentBranch()).toBe("exp");
      },
    );
  });
@@ -89,8 +89,11 @@ describe.each([arrow15, arrow16, arrow17, arrow18])(
      await table.add([{ id: 1 }]);
      expect(await table.countRows()).toBe(1);

+      expect(table.currentBranch()).toBeNull();
+
      // fork an isolated, writable branch from main
      const branch = await (await table.branches()).create("exp");
+      expect(branch.currentBranch()).toBe("exp");
      expect(await branch.countRows()).toBe(1);
      await branch.add([{ id: 2 }]);
      expect(await branch.countRows()).toBe(2);
@@ -109,6 +112,7 @@ describe.each([arrow15, arrow16, arrow17, arrow18])(

      // checkout returns a handle scoped to the branch's latest
      const checkedOut = await (await table.branches()).checkout("exp");
+      expect(checkedOut.currentBranch()).toBe("exp");
      expect(await checkedOut.countRows()).toBe(2);

      // delete removes it
@@ -845,11 +849,13 @@ describe("When creating an index", () => {
    expect(fs.readdirSync(indexDir)).toHaveLength(1);
    const indices = await tbl.listIndices();
    expect(indices.length).toBe(1);
-    expect(indices[0]).toEqual({
-      name: "vec_idx",
-      indexType: "IvfPq",
-      columns: ["vec"],
-    });
+    expect(indices[0]).toEqual(
+      expect.objectContaining({
+        name: "vec_idx",
+        indexType: "IvfPq",
+        columns: ["vec"],
+      }),
+    );
    const stats = await tbl.indexStats("vec_idx");
    expect(stats).toBeDefined();

@@ -1011,51 +1017,51 @@ describe("When creating an index", () => {
    const indices = await nestedTable.listIndices();
    expect(indices).toEqual(
      expect.arrayContaining([
-        {
+        expect.objectContaining({
          name: "row_id_idx",
          indexType: "BTree",
          columns: ["rowId"],
-        },
-        {
+        }),
+        expect.objectContaining({
          name: "row_dash_id_idx",
          indexType: "BTree",
          columns: ["`row-id`"],
-        },
-        {
+        }),
+        expect.objectContaining({
          name: "top_user_id_idx",
          indexType: "BTree",
          columns: ["userId"],
-        },
-        {
+        }),
+        expect.objectContaining({
          name: "nested_user_id_idx",
          indexType: "BTree",
          columns: ["metadata.user_id"],
-        },
-        {
+        }),
+        expect.objectContaining({
          name: "mixed_case_metadata_user_id_idx",
          indexType: "BTree",
          columns: ["MetaData.userId"],
-        },
-        {
+        }),
+        expect.objectContaining({
          name: "escaped_names_idx",
          indexType: "BTree",
          columns: ["`meta-data`.`user-id`"],
-        },
-        {
+        }),
+        expect.objectContaining({
          name: "literal_dot_idx",
          indexType: "BTree",
          columns: ["literal.`a.b`"],
-        },
-        {
+        }),
+        expect.objectContaining({
          name: "image_embedding_idx",
          indexType: "IvfPq",
          columns: ["image.embedding"],
-        },
-        {
+        }),
+        expect.objectContaining({
          name: "payload_text_idx",
          indexType: "FTS",
          columns: ["payload.text"],
-        },
+        }),
      ]),
    );

@@ -1109,16 +1115,16 @@ describe("When creating an index", () => {
    const indicesAfterOptimize = await nestedTable.listIndices();
    expect(indicesAfterOptimize).toEqual(
      expect.arrayContaining([
-        {
+        expect.objectContaining({
          name: "mixed_case_metadata_user_id_idx",
          indexType: "BTree",
          columns: ["MetaData.userId"],
-        },
-        {
+        }),
+        expect.objectContaining({
          name: "image_embedding_idx",
          indexType: "IvfPq",
          columns: ["image.embedding"],
-        },
+        }),
      ]),
    );
  });
@@ -1254,11 +1260,13 @@ describe("When creating an index", () => {
    expect(fs.readdirSync(indexDir)).toHaveLength(1);
    const indices = await tbl.listIndices();
    expect(indices.length).toBe(1);
-    expect(indices[0]).toEqual({
-      name: "vec_idx",
-      indexType: "IvfHnswSq",
-      columns: ["vec"],
-    });
+    expect(indices[0]).toEqual(
+      expect.objectContaining({
+        name: "vec_idx",
+        indexType: "IvfHnswSq",
+        columns: ["vec"],
+      }),
+    );

    // Search without specifying the column
    let rst = await tbl
@@ -1604,6 +1612,35 @@ describe("When creating an index", () => {
    expect(rst64Query.toString()).toEqual(rst64Search.toString());
    expect(rst64Query.numRows).toBe(2);
  });
+
+  it("should expose rich metadata fields on IndexConfig", async () => {
+    await tbl.createIndex("id", { config: Index.btree() });
+    await tbl.createIndex("vec");
+
+    const indicesByName = Object.fromEntries(
+      (await tbl.listIndices()).map((idx) => [idx.name, idx]),
+    );
+
+    const scalarIdx = indicesByName["id_idx"];
+    expect(scalarIdx).toBeDefined();
+    expect(typeof scalarIdx.indexUuid).toBe("string");
+    expect(scalarIdx.numIndexedRows).toBe(300);
+    expect(scalarIdx.numUnindexedRows).toBe(0);
+    expect(scalarIdx.numSegments).toBeGreaterThanOrEqual(1);
+    expect(scalarIdx.sizeBytes).toBeGreaterThan(0);
+    // Use toString check to avoid cross-realm instanceof failures with native Date objects
+    expect(Object.prototype.toString.call(scalarIdx.createdAt)).toBe(
+      "[object Date]",
+    );
+    expect((scalarIdx.createdAt as Date).getTime()).toBeGreaterThan(0);
+    expect(typeof scalarIdx.indexDetails).toBe("object");
+
+    const vectorIdx = indicesByName["vec_idx"];
+    expect(vectorIdx).toBeDefined();
+    expect(typeof vectorIdx.indexUuid).toBe("string");
+    expect(vectorIdx.numIndexedRows).toBe(300);
+    expect(typeof vectorIdx.indexDetails).toBe("object");
+  });
 });

 describe("When querying a table", () => {
@@ -2955,6 +2992,56 @@ describe("setLsmWriteSpec / unsetLsmWriteSpec", () => {
      }),
    ).rejects.toThrow();
  });
+
+  it("reads back the installed spec via getLsmWriteSpec", async () => {
+    const conn = await connect(tmpDir.name);
+    const table = await makeTable(conn);
+    await table.setUnenforcedPrimaryKey("id");
+
+    // Nothing installed yet.
+    expect(await table.getLsmWriteSpec()).toBeUndefined();
+
+    // A real scalar index is needed to name it as a maintained index.
+    await table.add([{ id: 1 }, { id: 2 }, { id: 3 }]);
+    await table.createIndex("id");
+    const indexName = (await table.listIndices())[0].name;
+
+    // Bucket spec round-trips, including maintained indexes and writer config
+    // defaults. Lance writer-config keys are canonically snake_case.
+    // biome-ignore lint/style/useNamingConvention: Lance writer-config keys are snake_case
+    const writerConfigDefaults = { durable_write: "false" };
+    await table.setLsmWriteSpec({
+      specType: "bucket",
+      column: "id",
+      numBuckets: 4,
+      maintainedIndexes: [indexName],
+      writerConfigDefaults,
+    });
+    const spec = await table.getLsmWriteSpec();
+    expect(spec).toBeDefined();
+    expect(spec?.specType).toBe("bucket");
+    expect(spec?.column).toBe("id");
+    expect(spec?.numBuckets).toBe(4);
+    expect(spec?.maintainedIndexes).toEqual([indexName]);
+    expect(spec?.writerConfigDefaults).toEqual(writerConfigDefaults);
+
+    // After unset, undefined again.
+    await table.unsetLsmWriteSpec();
+    expect(await table.getLsmWriteSpec()).toBeUndefined();
+
+    // Identity round-trips (column recovered from the schema).
+    await table.setLsmWriteSpec({ specType: "identity", column: "id" });
+    const identity = await table.getLsmWriteSpec();
+    expect(identity?.specType).toBe("identity");
+    expect(identity?.column).toBe("id");
+    await table.unsetLsmWriteSpec();
+
+    // Unsharded round-trips (no routing column).
+    await table.setLsmWriteSpec({ specType: "unsharded" });
+    const unsharded = await table.getLsmWriteSpec();
+    expect(unsharded?.specType).toBe("unsharded");
+    expect(unsharded?.column).toBeFalsy();
+  });
 });

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

+// OpenTelemetry metrics bridge. Only the high-level entry point is public; the
+// underlying recorder/catalog/snapshot functions remain internal plumbing that
+// `otel.ts` consumes from the native module.
+export { instrumentLanceDbMetrics } from "./otel";
+
 export {
  AddColumnsSql,
  ConnectionOptions,
@@ -52,6 +57,7 @@ export {
  SplitHashOptions,
  SplitSequentialOptions,
  ShuffleOptions,
+  OAuthConfig as NativeOAuthConfig,
 } from "./native.js";

 export {
@@ -130,6 +136,8 @@ export {
  TokenResponse,
 } from "./header";

+export { OAuthConfig, OAuthFlowType } from "./oauth";
+
 export { MergeInsertBuilder, WriteExecutionOptions } from "./merge";

 export * as embedding from "./embedding";
@@ -0,0 +1,76 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+/**
+ * OAuth authentication flow types.
+ */
+export enum OAuthFlowType {
+  /** Client Credentials grant (service-to-service / M2M). */
+  ClientCredentials = "client_credentials",
+  /** Azure Managed Identity via IMDS. */
+  AzureManagedIdentity = "azure_managed_identity",
+}
+
+/**
+ * OAuth configuration for LanceDB authentication.
+ *
+ * This is the public TypeScript OAuth configuration type. The generated
+ * `NativeOAuthConfig` type has the same runtime shape but is an implementation
+ * detail of the napi-rs binding.
+ *
+ * All token acquisition and refresh is handled in the Rust layer.
+ * This config is passed through to Rust via napi-rs.
+ *
+ * @example Client Credentials (service-to-service):
+ * ```typescript
+ * const config: OAuthConfig = {
+ *   issuerUrl: "https://login.microsoftonline.com/{tenant}/v2.0",
+ *   clientId: "app-id",
+ *   clientSecret: "secret",
+ *   scopes: ["api://lancedb-api/.default"],
+ * };
+ * ```
+ *
+ * @example Azure Managed Identity:
+ * ```typescript
+ * const config: OAuthConfig = {
+ *   issuerUrl: "https://login.microsoftonline.com/{tenant}/v2.0",
+ *   clientId: "app-id",
+ *   scopes: ["api://lancedb-api/.default"],
+ *   flow: OAuthFlowType.AzureManagedIdentity,
+ * };
+ * ```
+ */
+export interface OAuthConfig {
+  /**
+   * OIDC issuer URL or OAuth authority URL.
+   * For Azure: `https://login.microsoftonline.com/{tenant_id}/v2.0`
+   */
+  issuerUrl: string;
+
+  /** Application / Client ID. */
+  clientId: string;
+
+  /**
+   * OAuth scopes to request.
+   * For Azure managed identity, exactly one scope or resource is required.
+   * For example: `["api://{app_id}/.default"]`
+   */
+  scopes: string[];
+
+  /** Authentication flow (default: ClientCredentials). */
+  flow?: OAuthFlowType;
+
+  /** Client secret (required for ClientCredentials). */
+  clientSecret?: string;
+
+  /** Client ID for user-assigned managed identity (AzureManagedIdentity). */
+  managedIdentityClientId?: string;
+
+  /**
+   * Seconds before expiry to trigger proactive refresh (default: 300).
+   * Keep this well below the token TTL; if it is greater than or equal to
+   * the TTL, each request refreshes the token.
+   */
+  refreshBufferSecs?: number;
+}
@@ -0,0 +1,137 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+import {
+  type Attributes,
+  type MeterProvider,
+  type ObservableResult,
+  metrics,
+} from "@opentelemetry/api";
+
+import {
+  lancedbMetricsCatalog,
+  registerLancedbMetricsRecorder,
+  snapshotLancedbMetrics,
+} from "./native";
+
+let instrumented = false;
+
+/**
+ * Register LanceDB metrics as OpenTelemetry observable instruments.
+ *
+ * Installs a process-global metrics recorder and creates one observable
+ * instrument per LanceDB metric (currently object store request counts, bytes,
+ * latency, errors, and throttles) on the given (or global) `MeterProvider`. The
+ * configured `MetricReader` then collects them on its own schedule.
+ *
+ * Counters and gauges map directly to observable counters/gauges. Because
+ * OpenTelemetry has no asynchronous histogram instrument, each histogram is
+ * exported Prometheus-style as cumulative `le` bucket counts (`<name>_bucket`,
+ * with an `le` attribute) plus `<name>_count` and `<name>_sum`.
+ *
+ * Requires `@opentelemetry/api` (a dependency) and, to actually export, an
+ * OpenTelemetry SDK such as `@opentelemetry/sdk-metrics`.
+ *
+ * @param meterProvider The provider to register instruments on. Defaults to the
+ *   global provider from `@opentelemetry/api`.
+ * @returns `true` if the recorder is installed and instruments are registered.
+ *   `false` if a different `metrics` recorder is already installed in this
+ *   process (only one global recorder is permitted), in which case a warning is
+ *   emitted and no instruments are created. Calling this more than once is safe;
+ *   instruments are created only on the first successful call.
+ */
+export function instrumentLanceDbMetrics(
+  meterProvider?: MeterProvider,
+): boolean {
+  if (!registerLancedbMetricsRecorder()) {
+    console.warn(
+      "Could not install the LanceDB metrics recorder: another `metrics` " +
+        "recorder is already installed in this process. LanceDB metrics will " +
+        "not be exported via OpenTelemetry.",
+    );
+    return false;
+  }
+
+  if (instrumented) {
+    return true;
+  }
+
+  const provider = meterProvider ?? metrics.getMeterProvider();
+  const meter = provider.getMeter("lancedb");
+
+  const scalarCallback = (metricName: string) => (result: ObservableResult) => {
+    for (const point of snapshotLancedbMetrics()) {
+      if (point.name === metricName && point.value != null) {
+        result.observe(point.value, point.attributes);
+      }
+    }
+  };
+
+  const bucketCallback = (metricName: string) => (result: ObservableResult) => {
+    for (const point of snapshotLancedbMetrics()) {
+      if (point.name !== metricName || point.buckets == null) {
+        continue;
+      }
+      for (const bucket of point.buckets) {
+        const attributes: Attributes = {
+          ...point.attributes,
+          le: bucket.le,
+        };
+        result.observe(bucket.cumulativeCount, attributes);
+      }
+    }
+  };
+
+  const fieldCallback =
+    (metricName: string, field: "count" | "sum") =>
+    (result: ObservableResult) => {
+      for (const point of snapshotLancedbMetrics()) {
+        if (point.name !== metricName) {
+          continue;
+        }
+        const value = point[field];
+        if (value != null) {
+          result.observe(value, point.attributes);
+        }
+      }
+    };
+
+  for (const desc of lancedbMetricsCatalog()) {
+    const unit = desc.unit ?? "";
+    if (desc.kind === "counter") {
+      const counter = meter.createObservableCounter(desc.name, {
+        unit,
+        description: desc.description,
+      });
+      counter.addCallback(scalarCallback(desc.name));
+    } else if (desc.kind === "gauge") {
+      const gauge = meter.createObservableGauge(desc.name, {
+        unit,
+        description: desc.description,
+      });
+      gauge.addCallback(scalarCallback(desc.name));
+    } else if (desc.kind === "histogram") {
+      // `_bucket` and `_count` observe cumulative sample counts, not the
+      // histogram's measured quantity, so they are unitless; only `_sum`
+      // carries the histogram's unit.
+      const bucket = meter.createObservableCounter(`${desc.name}_bucket`, {
+        description: `${desc.description} (cumulative buckets)`,
+      });
+      bucket.addCallback(bucketCallback(desc.name));
+
+      const count = meter.createObservableCounter(`${desc.name}_count`, {
+        description: `${desc.description} (count)`,
+      });
+      count.addCallback(fieldCallback(desc.name, "count"));
+
+      const sum = meter.createObservableCounter(`${desc.name}_sum`, {
+        unit,
+        description: `${desc.description} (sum)`,
+      });
+      sum.addCallback(fieldCallback(desc.name, "sum"));
+    }
+  }
+
+  instrumented = true;
+  return true;
+}
@@ -362,6 +362,9 @@ export class StandardQueryBase<
   *
   * Filtering performance can often be improved by creating a scalar index
   * on the filter column(s).
+   *
+   * Calling this multiple times combines the filters with a logical AND rather
+   * than replacing the previous filter.
   */
  where(predicate: string): this {
    this.doCall((inner: NativeQueryType) => inner.onlyIf(predicate));
@@ -585,6 +585,17 @@ export abstract class Table {
   * @returns {Promise<void>}
   */
  abstract unsetLsmWriteSpec(): Promise<void>;
+  /**
+   * Read the {@link LsmWriteSpec} currently installed on this table.
+   *
+   * Resolves to `undefined` when the MemWAL LSM write path is not enabled (no
+   * spec has been set, or it was removed with {@link Table#unsetLsmWriteSpec}).
+   * The returned spec — including its `maintainedIndexes` and
+   * `writerConfigDefaults` — mirrors what was passed to
+   * {@link Table#setLsmWriteSpec}.
+   * @returns {Promise<LsmWriteSpec | undefined>}
+   */
+  abstract getLsmWriteSpec(): Promise<LsmWriteSpec | undefined>;
  /**
   * Drain and close any cached MemWAL shard writers held for this table.
   *
@@ -663,6 +674,14 @@ export abstract class Table {
   */
  abstract branches(): Promise<Branches>;

+  /**
+   * The branch this table handle is scoped to, or `null` for the main branch.
+   *
+   * A handle returned by {@link Branches.create} or {@link Branches.checkout}
+   * reports the branch it targets; a handle opened normally reports `null`.
+   */
+  abstract currentBranch(): string | null;
+
  /**
   * Restore the table to the currently checked out version
   *
@@ -1083,6 +1102,15 @@ export class LocalTable extends Table {
    return await this.inner.unsetLsmWriteSpec();
  }

+  async getLsmWriteSpec(): Promise<LsmWriteSpec | undefined> {
+    // The native binding types `specType` as a plain `string`; narrow it back
+    // to the public union. The Rust `From` impl only ever emits one of the
+    // three valid values, so the cast is safe.
+    return ((await this.inner.getLsmWriteSpec()) ?? undefined) as
+      | LsmWriteSpec
+      | undefined;
+  }
+
  async closeLsmWriters(): Promise<void> {
    return await this.inner.closeLsmWriters();
  }
@@ -1122,6 +1150,10 @@ export class LocalTable extends Table {
    return new Branches(await this.inner.branches());
  }

+  currentBranch(): string | null {
+    return this.inner.currentBranch() ?? null;
+  }
+
  async optimize(options?: Partial<OptimizeOptions>): Promise<OptimizeStats> {
    let cleanupOlderThanMs;
    if (
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-darwin-arm64",
-	"version": "0.30.1-beta.2",
+	"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.30.1-beta.2",
+	"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.30.1-beta.2",
+	"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.30.1-beta.2",
+	"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.30.1-beta.2",
+	"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.30.1-beta.2",
+  "version": "0.31.0-beta.6",
  "os": [
    "win32"
  ],
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-win32-x64-msvc",
-	"version": "0.30.1-beta.2",
+	"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.30.1-beta.2",
+  "version": "0.31.0-beta.6",
  "lockfileVersion": 3,
  "requires": true,
  "packages": {
    "": {
      "name": "@lancedb/lancedb",
-      "version": "0.30.1-beta.2",
+      "version": "0.31.0-beta.6",
      "cpu": [
        "x64",
        "arm64"
@@ -11,7 +11,7 @@
    "ann"
  ],
  "private": false,
-  "version": "0.30.1-beta.2",
+  "version": "0.31.0-beta.6",
  "main": "dist/index.js",
  "exports": {
    ".": "./dist/index.js",
@@ -44,6 +44,7 @@
    "@biomejs/biome": "^1.7.3",
    "@jest/globals": "^29.7.0",
    "@napi-rs/cli": "3.7.0",
+    "@opentelemetry/sdk-metrics": "^1.30.0",
    "@types/axios": "^0.14.0",
    "@types/jest": "^29.1.2",
    "@types/node": "22.7.4",
@@ -92,6 +93,7 @@
    "version": "napi version"
  },
  "dependencies": {
+    "@opentelemetry/api": "^1.9.0",
    "reflect-metadata": "^0.2.2"
  },
  "optionalDependencies": {
@@ -8,6 +8,9 @@ importers:

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

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

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

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

@@ -112,6 +112,12 @@ impl Connection {

        builder = builder.client_config(rust_config);

+        if let Some(oauth_config) = options.oauth_config {
+            let config: lancedb::remote::oauth::OAuthConfig =
+                oauth_config.try_into().default_error()?;
+            builder = builder.oauth_config(config);
+        }
+
        if let Some(api_key) = options.api_key {
            builder = builder.api_key(&api_key);
        }
@@ -12,6 +12,7 @@ mod header;
 mod index;
 mod iterator;
 pub mod merge;
+pub mod otel;
 pub mod permutation;
 mod query;
 pub mod remote;
@@ -65,6 +66,11 @@ pub struct ConnectionOptions {
    /// (For LanceDB cloud only): the host to use for LanceDB cloud. Used
    /// for testing purposes.
    pub host_override: Option<String>,
+    /// (For LanceDB cloud only): OAuth configuration for IdP-based
+    /// authentication (e.g., Azure Entra ID). When set, token acquisition
+    /// and refresh are handled entirely in Rust. TypeScript users should pass
+    /// the public `OAuthConfig` type exported from `@lancedb/lancedb`.
+    pub oauth_config: Option<remote::OAuthConfig>,
 }

 #[napi(object)]
@@ -3,7 +3,7 @@

 use std::time::Duration;

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

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

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

        let this = self.clone();

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

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

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

        self.modify(|builder| {
            builder.with_split_strategy(
-                SplitStrategy::Random { seed, sizes },
+                SplitStrategy::Random {
+                    seed,
+                    sizes,
+                    clump_size,
+                },
                options.split_names.clone(),
            )
        })
@@ -3,6 +3,7 @@

 use std::collections::HashMap;

+use lancedb::error::Error;
 use napi_derive::*;

 /// Timeout configuration for remote HTTP client.
@@ -140,6 +141,84 @@ impl From<TlsConfig> for lancedb::remote::TlsConfig {
    }
 }

+/// OAuth configuration for LanceDB authentication.
+///
+/// This is the generated napi-rs binding shape. TypeScript users should prefer
+/// the public `OAuthConfig` type exported from `@lancedb/lancedb`.
+///
+/// All token acquisition and refresh is handled in the Rust layer.
+#[napi(object)]
+#[derive(Clone)]
+pub struct OAuthConfig {
+    /// OIDC issuer URL or OAuth authority URL.
+    /// For Azure: `https://login.microsoftonline.com/{tenant_id}/v2.0`
+    pub issuer_url: String,
+    /// Application / Client ID.
+    pub client_id: String,
+    /// OAuth scopes to request. For Azure managed identity, exactly one scope
+    /// or resource is required. For example: `["api://{app_id}/.default"]`
+    pub scopes: Vec<String>,
+    /// Authentication flow: "client_credentials" or "azure_managed_identity"
+    pub flow: Option<String>,
+    /// Client secret (required for client_credentials).
+    pub client_secret: Option<String>,
+    /// Client ID for user-assigned managed identity (azure_managed_identity).
+    pub managed_identity_client_id: Option<String>,
+    /// Seconds before expiry to trigger proactive refresh (default: 300).
+    /// Keep this well below the token TTL; if it is greater than or equal to
+    /// the TTL, each request refreshes the token.
+    pub refresh_buffer_secs: Option<u32>,
+}
+
+impl std::fmt::Debug for OAuthConfig {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("OAuthConfig")
+            .field("issuer_url", &self.issuer_url)
+            .field("client_id", &self.client_id)
+            .field("scopes", &self.scopes)
+            .field("flow", &self.flow)
+            .field(
+                "client_secret",
+                &self.client_secret.as_deref().map(|_| "<redacted>"),
+            )
+            .field(
+                "managed_identity_client_id",
+                &self.managed_identity_client_id,
+            )
+            .field("refresh_buffer_secs", &self.refresh_buffer_secs)
+            .finish()
+    }
+}
+
+impl TryFrom<OAuthConfig> for lancedb::remote::oauth::OAuthConfig {
+    type Error = Error;
+
+    fn try_from(config: OAuthConfig) -> Result<Self, Self::Error> {
+        use lancedb::remote::oauth::OAuthFlow;
+
+        let flow = match config.flow.as_deref().unwrap_or("client_credentials") {
+            "client_credentials" => OAuthFlow::ClientCredentials,
+            "azure_managed_identity" => OAuthFlow::AzureManagedIdentity {
+                client_id: config.managed_identity_client_id,
+            },
+            other => {
+                return Err(Error::InvalidInput {
+                    message: format!("Unknown OAuth flow type: {other}"),
+                });
+            }
+        };
+
+        Ok(Self {
+            issuer_url: config.issuer_url,
+            client_id: config.client_id,
+            client_secret: config.client_secret,
+            scopes: config.scopes,
+            flow,
+            refresh_buffer_secs: config.refresh_buffer_secs.map(|v| v as u64),
+        })
+    }
+}
+
 impl From<ClientConfig> for lancedb::remote::ClientConfig {
    fn from(config: ClientConfig) -> Self {
        Self {
@@ -156,3 +235,45 @@ impl From<ClientConfig> for lancedb::remote::ClientConfig {
        }
    }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_unknown_oauth_flow_returns_invalid_input() {
+        let config = OAuthConfig {
+            issuer_url: "https://issuer.example.com".to_string(),
+            client_id: "client-id".to_string(),
+            scopes: vec!["scope".to_string()],
+            flow: Some("typo".to_string()),
+            client_secret: None,
+            managed_identity_client_id: None,
+            refresh_buffer_secs: None,
+        };
+
+        let err = lancedb::remote::oauth::OAuthConfig::try_from(config).unwrap_err();
+        assert!(matches!(
+            err,
+            Error::InvalidInput { message }
+                if message == "Unknown OAuth flow type: typo"
+        ));
+    }
+
+    #[test]
+    fn test_oauth_config_debug_redacts_client_secret() {
+        let config = OAuthConfig {
+            issuer_url: "https://issuer.example.com".to_string(),
+            client_id: "client-id".to_string(),
+            scopes: vec!["scope".to_string()],
+            flow: Some("client_credentials".to_string()),
+            client_secret: Some("super-secret".to_string()),
+            managed_identity_client_id: None,
+            refresh_buffer_secs: None,
+        };
+
+        let debug = format!("{config:?}");
+        assert!(!debug.contains("super-secret"));
+        assert!(debug.contains("client_secret: Some(\"<redacted>\")"));
+    }
+}
@@ -3,6 +3,8 @@

 use std::collections::HashMap;

+use chrono::{DateTime, Utc};
+
 use lancedb::ipc::{ipc_file_to_batches, ipc_file_to_schema};
 use lancedb::table::{
    AddDataMode, ColumnAlteration as LanceColumnAlteration, Duration,
@@ -409,6 +411,16 @@ impl Table {
            .default_error()
    }

+    #[napi(catch_unwind)]
+    pub async fn get_lsm_write_spec(&self) -> napi::Result<Option<LsmWriteSpec>> {
+        let spec = self
+            .inner_ref()?
+            .get_lsm_write_spec()
+            .await
+            .default_error()?;
+        Ok(spec.map(LsmWriteSpec::from))
+    }
+
    #[napi(catch_unwind)]
    pub async fn close_lsm_writers(&self) -> napi::Result<()> {
        self.inner_ref()?.close_lsm_writers().await.default_error()
@@ -485,6 +497,12 @@ impl Table {
        })
    }

+    /// The branch this handle is scoped to, or `null` for the main branch.
+    #[napi]
+    pub fn current_branch(&self) -> napi::Result<Option<String>> {
+        Ok(self.inner_ref()?.current_branch())
+    }
+
    #[napi(catch_unwind)]
    pub async fn optimize(
        &self,
@@ -602,6 +620,43 @@ pub struct IndexConfig {
    /// Currently this is always an array of size 1. In the future there may
    /// be more columns to represent composite indices.
    pub columns: Vec<String>,
+    /// The UUID of the first segment of the index.
+    ///
+    /// `undefined` for remote tables, which do not yet surface this.
+    pub index_uuid: Option<String>,
+    /// The protobuf type URL, a precise type identifier for the index.
+    ///
+    /// `undefined` for remote tables.
+    pub type_url: Option<String>,
+    /// When the index was created.
+    ///
+    /// `undefined` for remote tables or indices created before timestamps were tracked.
+    pub created_at: Option<DateTime<Utc>>,
+    /// The number of rows indexed, across all segments.
+    ///
+    /// `undefined` for remote tables.
+    pub num_indexed_rows: Option<i64>,
+    /// The number of rows not yet covered by this index.
+    ///
+    /// `undefined` for remote tables.
+    pub num_unindexed_rows: Option<i64>,
+    /// The total size in bytes of all index files across all segments.
+    ///
+    /// `undefined` for remote tables or indices without size tracking.
+    pub size_bytes: Option<i64>,
+    /// The number of segments that make up the index.
+    ///
+    /// `undefined` for remote tables.
+    pub num_segments: Option<i32>,
+    /// The on-disk index format version.
+    ///
+    /// `undefined` for remote tables.
+    pub index_version: Option<i32>,
+    /// Index-type-specific details parsed as a JavaScript object.
+    ///
+    /// Falls back to a raw string if JSON parsing fails. `undefined` for
+    /// remote tables or when details are unavailable.
+    pub index_details: Option<serde_json::Value>,
 }

 impl From<lancedb::index::IndexConfig> for IndexConfig {
@@ -611,6 +666,17 @@ impl From<lancedb::index::IndexConfig> for IndexConfig {
            index_type,
            columns: value.columns,
            name: value.name,
+            index_uuid: value.index_uuid,
+            type_url: value.type_url,
+            created_at: value.created_at,
+            num_indexed_rows: value.num_indexed_rows.map(|n| n as i64),
+            num_unindexed_rows: value.num_unindexed_rows.map(|n| n as i64),
+            size_bytes: value.size_bytes.map(|n| n as i64),
+            num_segments: value.num_segments.map(|n| n as i32),
+            index_version: value.index_version,
+            index_details: value
+                .index_details
+                .and_then(|s| serde_json::from_str(&s).ok()),
        }
    }
 }
@@ -672,6 +738,47 @@ impl TryFrom<LsmWriteSpec> for lancedb::table::LsmWriteSpec {
    }
 }

+impl From<lancedb::table::LsmWriteSpec> for LsmWriteSpec {
+    fn from(spec: lancedb::table::LsmWriteSpec) -> Self {
+        use lancedb::table::LsmWriteSpec as Native;
+        match spec {
+            Native::Bucket {
+                column,
+                num_buckets,
+                maintained_indexes,
+                writer_config_defaults,
+            } => Self {
+                spec_type: "bucket".to_string(),
+                column: Some(column),
+                num_buckets: Some(num_buckets),
+                maintained_indexes: Some(maintained_indexes),
+                writer_config_defaults: Some(writer_config_defaults),
+            },
+            Native::Identity {
+                column,
+                maintained_indexes,
+                writer_config_defaults,
+            } => Self {
+                spec_type: "identity".to_string(),
+                column: Some(column),
+                num_buckets: None,
+                maintained_indexes: Some(maintained_indexes),
+                writer_config_defaults: Some(writer_config_defaults),
+            },
+            Native::Unsharded {
+                maintained_indexes,
+                writer_config_defaults,
+            } => Self {
+                spec_type: "unsharded".to_string(),
+                column: None,
+                num_buckets: None,
+                maintained_indexes: Some(maintained_indexes),
+                writer_config_defaults: Some(writer_config_defaults),
+            },
+        }
+    }
+}
+
 /// Statistics about a compaction operation.
 #[napi(object)]
 #[derive(Clone, Debug)]
@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "0.33.1-beta.2"
+current_version = "0.35.0-beta.0"
 parse = """(?x)
    (?P<major>0|[1-9]\\d*)\\.
    (?P<minor>0|[1-9]\\d*)\\.
@@ -23,6 +23,8 @@ allow_dirty = true
 commit = true
 message = "Bump version: {current_version} → {new_version}"
 commit_args = ""
+# bump-my-version >=1.4.0 rejects pre_commit_hooks containing shell syntax unless opted in.
+allow_shell_hooks = true

 # Update Cargo.lock after version bump
 pre_commit_hooks = [
@@ -1,6 +1,6 @@
 [package]
 name = "lancedb-python"
-version = "0.33.1-beta.2"
+version = "0.35.0-beta.0"
 publish = false
 edition.workspace = true
 description = "Python bindings for LanceDB"
@@ -26,7 +26,8 @@ lance-namespace-impls.workspace = true
 lance-io.workspace = true
 env_logger.workspace = true
 log.workspace = true
-pyo3 = { version = "0.28", features = ["extension-module", "abi3-py39"] }
+pyo3 = { version = "0.28", features = ["extension-module", "abi3-py39", "chrono"] }
+chrono = { version = "0.4", default-features = false, features = ["clock"] }
 pyo3-async-runtimes = { version = "0.28", features = [
    "attributes",
    "tokio-runtime",
@@ -46,6 +47,6 @@ pyo3-build-config = { version = "0.28", features = [
 ] }

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

    For object storage, use a URI prefix:

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

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

@@ -340,6 +348,7 @@ async def connect_async(
    session: Optional[Session] = None,
    manifest_enabled: bool = False,
    namespace_client_properties: Optional[Dict[str, str]] = None,
+    oauth_config=None,
 ) -> AsyncConnection:
    """Connect to a LanceDB database.

@@ -389,6 +398,10 @@ async def connect_async(
    namespace_client_properties : dict, optional
        Additional directory namespace client properties to use with
        ``manifest_enabled=True``.
+    oauth_config : OAuthConfig, optional
+        OAuth configuration for LanceDB Cloud/Enterprise. This is supported by
+        ``connect_async`` only; synchronous ``connect`` uses API key
+        authentication for ``db://`` URIs.

    Examples
    --------
@@ -435,6 +448,7 @@ async def connect_async(
            session,
            manifest_enabled,
            namespace_client_properties,
+            oauth_config,
        )
    )

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

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

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

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

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

 class Session:
@@ -205,7 +227,7 @@ class Table:
    async def prewarm_index(self, index_name: str) -> None: ...
    async def prewarm_data(self, columns: Optional[List[str]] = None) -> None: ...
    async def list_indices(self) -> list[IndexConfig]: ...
-    async def delete(self, filter: str) -> DeleteResult: ...
+    async def delete(self, filter: Union[str, PyExpr]) -> DeleteResult: ...
    async def add_columns(self, columns: list[tuple[str, str]]) -> AddColumnsResult: ...
    async def add_columns_with_schema(self, schema: pa.Schema) -> AddColumnsResult: ...
    async def alter_columns(
@@ -226,6 +248,7 @@ class Table:
    async def set_unenforced_primary_key(self, columns: List[str]) -> None: ...
    async def set_lsm_write_spec(self, spec: LsmWriteSpec) -> None: ...
    async def unset_lsm_write_spec(self) -> None: ...
+    async def get_lsm_write_spec(self) -> Optional[LsmWriteSpec]: ...
    async def close_lsm_writers(self) -> None: ...
    @property
    def tags(self) -> Tags: ...
@@ -259,6 +282,15 @@ class IndexConfig:
    name: str
    index_type: str
    columns: List[str]
+    index_uuid: Optional[str]
+    type_url: Optional[str]
+    created_at: Optional[datetime]
+    num_indexed_rows: Optional[int]
+    num_unindexed_rows: Optional[int]
+    size_bytes: Optional[int]
+    num_segments: Optional[int]
+    index_version: Optional[int]
+    index_details: Optional[Any]

 async def connect(
    uri: str,
@@ -271,6 +303,24 @@ async def connect(
    session: Optional[Session],
    manifest_enabled: bool = False,
    namespace_client_properties: Optional[Dict[str, str]] = None,
+    oauth_config: Optional[Any] = None,
+) -> Connection: ...
+def connect_namespace(
+    namespace_client_impl: str,
+    namespace_client_properties: Dict[str, str],
+    read_consistency_interval: Optional[float] = None,
+    storage_options: Optional[Dict[str, str]] = None,
+    session: Optional[Session] = None,
+    namespace_client_pushdown_operations: Optional[List[str]] = None,
+) -> Connection: ...
+def connect_namespace_client(
+    namespace_client: Any,
+    read_consistency_interval: Optional[float] = None,
+    storage_options: Optional[Dict[str, str]] = None,
+    session: Optional[Session] = None,
+    namespace_client_pushdown_operations: Optional[List[str]] = None,
+    namespace_client_impl: Optional[str] = None,
+    namespace_client_properties: Optional[Dict[str, str]] = None,
 ) -> Connection: ...

 class RecordBatchStream:
@@ -81,6 +81,7 @@ class ColPaliEmbeddings(EmbeddingFunction):
            warnings.warn(
                "use_token_pooling is deprecated, use pooling_strategy=None instead",
                DeprecationWarning,
+                stacklevel=2,
            )
            self.pooling_strategy = None

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

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

 import numpy as np

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

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

 from __future__ import annotations

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

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


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


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

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

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


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

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

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


@@ -5,7 +5,9 @@
 from __future__ import annotations

 from datetime import timedelta
-from typing import TYPE_CHECKING, List, Optional
+from typing import TYPE_CHECKING, List, Optional, Union
+
+from .expr import Expr

 if TYPE_CHECKING:
    from .common import DATA
@@ -32,6 +34,7 @@ class LanceMergeInsertBuilder(object):
        self._when_not_matched_insert_all = False
        self._when_not_matched_by_source_delete = False
        self._when_not_matched_by_source_condition = None
+        self._when_not_matched_by_source_condition_expr = None
        self._timeout = None
        self._use_index = True
        self._use_lsm_write = None
@@ -48,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
@@ -62,7 +74,7 @@ class LanceMergeInsertBuilder(object):
        return self

    def when_not_matched_by_source_delete(
-        self, condition: Optional[str] = None
+        self, condition: Union[str, Expr, None] = None
    ) -> LanceMergeInsertBuilder:
        """
        Rows that exist only in the target table (old data) will be
@@ -71,13 +83,16 @@ class LanceMergeInsertBuilder(object):

        Parameters
        ----------
-        condition: Optional[str], default None
+        condition: str or :class:`~lancedb.expr.Expr` or None, default None
            If None then all such rows will be deleted.  Otherwise the
-            condition will be used as an SQL filter to limit what rows
-            are deleted.
+            condition will be used as a filter to limit what rows are deleted.
+            Can be a SQL string or a type-safe :class:`~lancedb.expr.Expr`
+            built with :func:`~lancedb.expr.col` and :func:`~lancedb.expr.lit`.
        """
        self._when_not_matched_by_source_delete = True
-        if condition is not None:
+        if isinstance(condition, Expr):
+            self._when_not_matched_by_source_condition_expr = condition._inner
+        elif condition is not None:
            self._when_not_matched_by_source_condition = condition
        return self

@@ -38,15 +38,13 @@ from lance_namespace_urllib3_client.models.query_table_request_vector import (
    QueryTableRequestVector,
 )
 from lance_namespace_urllib3_client.models.string_fts_query import StringFtsQuery
-from lance_namespace.errors import TableNotFoundError
-from lancedb._lancedb import connect_namespace_client as _connect_namespace_client
+from lance_namespace.errors import NamespaceNotEmptyError, TableNotFoundError
+from lancedb._lancedb import (
+    connect_namespace as _connect_namespace,
+    connect_namespace_client as _connect_namespace_client,
+)
 from lancedb.background_loop import LOOP
 from lancedb.db import AsyncConnection, DBConnection
-from lancedb.namespace_utils import (
-    _normalize_create_namespace_mode,
-    _normalize_drop_namespace_mode,
-    _normalize_drop_namespace_behavior,
-)
 from lance_namespace import (
    LanceNamespace,
    connect as namespace_connect,
@@ -55,12 +53,6 @@ from lance_namespace import (
    DropNamespaceResponse,
    ListNamespacesResponse,
    ListTablesResponse,
-    ListTablesRequest,
-    DescribeNamespaceRequest,
-    DropTableRequest,
-    ListNamespacesRequest,
-    CreateNamespaceRequest,
-    DropNamespaceRequest,
 )
 from lancedb.table import AsyncTable, LanceTable, Table
 from lancedb.util import validate_table_name
@@ -70,6 +62,9 @@ from lancedb.embeddings import EmbeddingFunctionConfig
 from ._lancedb import Session


+_MAX_QUERY_K = 2**31 - 1
+
+
 def _query_to_namespace_request(
    table_id: List[str],
    query: "Query",
@@ -147,7 +142,8 @@ def _query_to_namespace_request(
    if query.limit is not None:
        k = query.limit
    elif query.vector is None and query.full_text_query is None:
-        k = sys.maxsize
+        # limit k to max i32 value to avoid client overflows
+        k = _MAX_QUERY_K
    else:
        k = 10

@@ -368,6 +364,23 @@ def _convert_pyarrow_schema_to_json(schema: pa.Schema) -> JsonArrowSchema:
    return JsonArrowSchema(fields=fields, metadata=meta)


+def _builds_namespace_natively(
+    namespace_client_impl: Optional[str],
+    namespace_client_properties: Optional[Dict[str, str]],
+) -> bool:
+    """Whether ``connect_namespace_client`` builds the namespace client natively
+    in Rust (installing the read-freshness context provider) rather than wrapping
+    the pre-built Python client.
+
+    Must mirror Rust ``build_namespace_natively`` in ``python/src/connection.rs``.
+    """
+    return namespace_client_impl == "rest" and bool(namespace_client_properties)
+
+
+def _supports_native_namespace(namespace_client_impl: str) -> bool:
+    return namespace_client_impl in {"dir", "rest"}
+
+
 class LanceNamespaceDBConnection(DBConnection):
    """
    A LanceDB connection that uses a namespace for table management.
@@ -378,7 +391,7 @@ class LanceNamespaceDBConnection(DBConnection):

    def __init__(
        self,
-        namespace_client: LanceNamespace,
+        namespace_client: Optional[LanceNamespace] = None,
        *,
        read_consistency_interval: Optional[timedelta] = None,
        storage_options: Optional[Dict[str, str]] = None,
@@ -386,6 +399,7 @@ class LanceNamespaceDBConnection(DBConnection):
        namespace_client_pushdown_operations: Optional[List[str]] = None,
        namespace_client_impl: Optional[str] = None,
        namespace_client_properties: Optional[Dict[str, str]] = None,
+        _inner: Optional[AsyncConnection] = None,
    ):
        """
        Initialize a namespace-based LanceDB connection.
@@ -427,23 +441,36 @@ class LanceNamespaceDBConnection(DBConnection):
        )
        self._namespace_client_impl = namespace_client_impl
        self._namespace_client_properties = 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,
-            )
+        # 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
+        else:
+            if namespace_client is None:
+                raise ValueError("namespace_client is required without a native _inner")
+            self._inner = AsyncConnection(
+                _connect_namespace_client(
+                    namespace_client,
+                    read_consistency_interval=(
+                        read_consistency_interval.total_seconds()
+                        if read_consistency_interval is not None
+                        else None
+                    ),
+                    storage_options=self.storage_options or None,
+                    session=session,
+                    namespace_client_pushdown_operations=(
+                        list(self._namespace_client_pushdown_operations)
+                    ),
+                    namespace_client_impl=namespace_client_impl,
+                    namespace_client_properties=namespace_client_properties,
+                )
+            )
+        self._uri = self._inner.uri

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

    @override
    def create_table(
@@ -538,6 +565,7 @@ class LanceNamespaceDBConnection(DBConnection):
            namespace_path=namespace_path,
            namespace_client=self._namespace_client,
            pushdown_operations=self._namespace_client_pushdown_operations,
+            route_pushdown_to_rust=self._route_pushdown_to_rust,
            _async=async_table,
        )

@@ -563,8 +591,8 @@ class LanceNamespaceDBConnection(DBConnection):
                    index_cache_size=index_cache_size,
                )
            )
-        except RuntimeError as e:
-            if "Table not found" in str(e):
+        except (RuntimeError, ValueError) as e:
+            if "Table not found" in str(e) or "was not found" in str(e):
                table_id = namespace_path + [name]
                raise TableNotFoundError(f"Table not found: {'$'.join(table_id)}")
            raise
@@ -575,6 +603,7 @@ class LanceNamespaceDBConnection(DBConnection):
            namespace_path=namespace_path,
            namespace_client=self._namespace_client,
            pushdown_operations=self._namespace_client_pushdown_operations,
+            route_pushdown_to_rust=self._route_pushdown_to_rust,
            _async=async_table,
        )
        if branch is not None:
@@ -585,12 +614,9 @@ class LanceNamespaceDBConnection(DBConnection):

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

    @override
    def rename_table(
@@ -604,9 +630,19 @@ class LanceNamespaceDBConnection(DBConnection):
            cur_namespace_path = []
        if new_namespace_path is None:
            new_namespace_path = []
-        raise NotImplementedError(
-            "rename_table is not supported for namespace connections"
-        )
+        try:
+            LOOP.run(
+                self._inner.rename_table(
+                    cur_name,
+                    new_name,
+                    cur_namespace_path=cur_namespace_path,
+                    new_namespace_path=new_namespace_path,
+                )
+            )
+        except RuntimeError as e:
+            if "rename_table not implemented" in str(e):
+                raise NotImplementedError("rename_table not implemented") from e
+            raise

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

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

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

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

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

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

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


@@ -859,12 +892,15 @@ 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,
        session: Optional[Session] = None,
        namespace_client_pushdown_operations: Optional[List[str]] = None,
+        namespace_client_impl: Optional[str] = None,
+        namespace_client_properties: Optional[Dict[str, str]] = None,
+        _inner: Optional[AsyncConnection] = None,
    ):
        """
        Initialize an async namespace-based LanceDB connection.
@@ -890,6 +926,12 @@ class AsyncLanceNamespaceDBConnection:
              namespace.create_table() instead of using declare_table + local write.

            Default is None (no pushdown, all operations run locally).
+        namespace_client_impl : Optional[str]
+            The namespace implementation name used to create this connection.
+            Required (with ``namespace_client_properties``) for the Rust client to
+            be built natively and install the read-freshness provider.
+        namespace_client_properties : Optional[Dict[str, str]]
+            The namespace properties used to create this connection.
        """
        self._namespace_client = namespace_client
        self.read_consistency_interval = read_consistency_interval
@@ -898,23 +940,37 @@ class AsyncLanceNamespaceDBConnection:
        self._namespace_client_pushdown_operations = set(
            namespace_client_pushdown_operations or []
        )
-        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=None,
-                namespace_client_properties=None,
-            )
+        self._namespace_client_impl = namespace_client_impl
+        self._namespace_client_properties = namespace_client_properties
+        # 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
        )
+        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,
@@ -938,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,
@@ -982,6 +1036,7 @@ class AsyncLanceNamespaceDBConnection:
            namespace_path=namespace_path,
            namespace_client=self._namespace_client,
            pushdown_operations=self._namespace_client_pushdown_operations,
+            route_pushdown_to_rust=self._route_pushdown_to_rust,
        )

    async def open_table(
@@ -1004,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
@@ -1019,15 +1074,14 @@ class AsyncLanceNamespaceDBConnection:
            namespace_path=namespace_path,
            namespace_client=self._namespace_client,
            pushdown_operations=self._namespace_client_pushdown_operations,
+            route_pushdown_to_rust=self._route_pushdown_to_rust,
        )

    async def drop_table(self, name: str, namespace_path: Optional[List[str]] = None):
        """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,
@@ -1036,14 +1090,22 @@ class AsyncLanceNamespaceDBConnection:
        cur_namespace_path: Optional[List[str]] = None,
        new_namespace_path: Optional[List[str]] = None,
    ):
-        """Rename is not supported for namespace connections."""
+        """Rename a table in the namespace."""
        if cur_namespace_path is None:
            cur_namespace_path = []
        if new_namespace_path is None:
            new_namespace_path = []
-        raise NotImplementedError(
-            "rename_table is not supported for namespace connections"
-        )
+        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."""
@@ -1055,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,
@@ -1086,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(
@@ -1119,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,
@@ -1153,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]
@@ -1184,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,
@@ -1217,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:
@@ -1237,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


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

 if TYPE_CHECKING:
@@ -48,6 +48,14 @@ class PermutationBuilder:
        By default, the permutation builder will create a single split that contains all
        rows in the same order as the base table.
        """
+        if not hasattr(table, "_inner"):
+            raise TypeError(
+                f"PermutationBuilder requires a local LanceTable, "
+                f"got {type(table).__name__}. "
+                "The permutation API is not supported on remote tables. "
+                "Remote tables connect to LanceDB Cloud or Enterprise and do not have "
+                "direct access to the underlying Lance dataset needed for permutations."
+            )
        self._async = async_permutation_builder(table)

    def split_random(
@@ -57,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":
        """
@@ -79,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.
        """
@@ -87,6 +99,7 @@ class PermutationBuilder:
            counts=counts,
            fixed=fixed,
            seed=seed,
+            clump_size=clump_size,
            split_names=split_names,
        )
        return self
@@ -933,6 +946,7 @@ class Permutation:
            "pandas",
            "arrow",
            "torch",
+            "torch_row",
            "torch_col",
            "polars",
        ],
@@ -948,15 +962,19 @@ class Permutation:
        - "python_col" - the batch will be a dict of lists (one entry per column)
        - "pandas" - the batch will be a pandas DataFrame
        - "arrow" - the batch will be a pyarrow RecordBatch
-        - "torch" - the batch will be a list of tensors, one per row
+        - "torch" - the batch will be a list of per-row dicts mapping column
+          name to a 0-D torch tensor. Works with the default
+          ``torch.utils.data.DataLoader`` collate, which stacks the per-row
+          dicts back into a dict of batched tensors.
+        - "torch_row" - the batch will be a list of tensors, one per row
        - "torch_col" - the batch will be a 2D torch tensor (first dim indexes columns)
        - "polars" - the batch will be a polars DataFrame

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

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

        Conversion to numpy and/or pandas will typically be zero-copy for numeric
        types.  Conversion of strings, lists, and structs will require creating python
@@ -977,6 +995,8 @@ class Permutation:
        elif format == "arrow":
            return self.with_transform(Transforms.arrow2arrow)
        elif format == "torch":
+            return self.with_transform(batch_to_tensor_dict)
+        elif format == "torch_row":
            return self.with_transform(batch_to_tensor_rows)
        elif format == "torch_col":
            return self.with_transform(batch_to_tensor)
@@ -275,7 +275,18 @@ def _py_type_to_arrow_type(py_type: Type[Any], field: FieldInfo) -> pa.DataType:
        tz = get_extras(field, "tz")
        return pa.timestamp("us", tz=tz)
    elif getattr(py_type, "__origin__", None) in (list, tuple):
-        child = py_type.__args__[0]
+        # A bare, unparameterised ``typing.List`` / ``typing.Tuple`` matches this
+        # branch (its ``__origin__`` is ``list`` / ``tuple``) but has no
+        # ``__args__``, so we cannot infer the element type. Raise a clear
+        # ``TypeError`` instead of crashing with an opaque ``AttributeError``.
+        args = getattr(py_type, "__args__", None)
+        if not args:
+            raise TypeError(
+                "Converting Pydantic type to Arrow Type: unsupported type "
+                f"{py_type}. Specify the element type, e.g. List[int] instead "
+                "of a bare List."
+            )
+        child = args[0]
        return _pydantic_list_child_to_arrow(child, field)
    raise TypeError(
        f"Converting Pydantic type to Arrow Type: unsupported type {py_type}."
@@ -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)
@@ -9,6 +9,7 @@ from typing import List, Optional
 from lancedb import __version__

 from .header import HeaderProvider
+from .oauth import OAuthConfig, OAuthFlowType

 __all__ = [
    "TimeoutConfig",
@@ -16,6 +17,8 @@ __all__ = [
    "TlsConfig",
    "ClientConfig",
    "HeaderProvider",
+    "OAuthConfig",
+    "OAuthFlowType",
 ]


@@ -124,6 +124,7 @@ class RemoteDBConnection(DBConnection):
                "request_thread_pool is no longer used and will be removed in "
                "a future release.",
                DeprecationWarning,
+                stacklevel=2,
            )

        if connection_timeout is not None:
@@ -132,6 +133,7 @@ class RemoteDBConnection(DBConnection):
                "release. Please use client_config.timeout_config.connect_timeout "
                "instead.",
                DeprecationWarning,
+                stacklevel=2,
            )
            client_config.timeout_config.connect_timeout = timedelta(
                seconds=connection_timeout
@@ -142,6 +144,7 @@ class RemoteDBConnection(DBConnection):
                "read_timeout is deprecated and will be removed in a future release. "
                "Please use client_config.timeout_config.read_timeout instead.",
                DeprecationWarning,
+                stacklevel=2,
            )
            client_config.timeout_config.read_timeout = timedelta(seconds=read_timeout)

@@ -396,13 +399,13 @@ class RemoteDBConnection(DBConnection):
            The namespace to open the table from.
            None or empty list represents root namespace.
        branch: str, optional
-            Branching is not yet supported on remote tables, so only the
-            default branch is accepted (``None`` or ``"main"``); any other
-            value raises ``NotImplementedError``.
+            If provided, open a handle scoped to this branch instead of the
+            default branch. Reads and writes operate in the branch's context.
        version: int, optional
            If provided, open the table pinned to this version, producing a
-            read-only handle. Call ``checkout_latest`` to return to a writable
-            state.
+            read-only handle. Composes with ``branch``: when both are given,
+            opens that branch at the version; otherwise opens ``main`` at the
+            version. Call ``checkout_latest`` to return to a writable state.

        Returns
        -------
@@ -410,11 +413,6 @@ class RemoteDBConnection(DBConnection):
        """
        from .table import RemoteTable

-        # Remote supports version time-travel but not branches: reject a non-main
-        # branch, but allow a version-only open (or "main").
-        if branch is not None and branch != "main":
-            raise NotImplementedError("branching is not yet supported on remote tables")
-
        if namespace_path is None:
            namespace_path = []
        if index_cache_size is not None:
@@ -430,7 +428,9 @@ class RemoteDBConnection(DBConnection):
            connection_state=self.serialize,
            namespace_path=namespace_path,
        )
-        if version is not None:
+        if branch is not None:
+            tbl = tbl.branches.checkout(branch, version)
+        elif version is not None:
            tbl.checkout(version)
        return tbl

@@ -0,0 +1,75 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import List, Optional
+
+
+class OAuthFlowType(str, Enum):
+    """OAuth authentication flow types."""
+
+    CLIENT_CREDENTIALS = "client_credentials"
+    """Client Credentials grant (service-to-service / M2M)."""
+
+    AZURE_MANAGED_IDENTITY = "azure_managed_identity"
+    """Azure Managed Identity via IMDS."""
+
+
+@dataclass
+class OAuthConfig:
+    """OAuth configuration for LanceDB authentication.
+
+    All token acquisition and refresh is handled in the Rust layer.
+    This config is passed through to Rust via PyO3.
+
+    Parameters
+    ----------
+    issuer_url : str
+        OIDC issuer URL or OAuth authority URL.
+        For Azure: ``https://login.microsoftonline.com/{tenant_id}/v2.0``
+    client_id : str
+        Application / Client ID.
+    scopes : List[str]
+        OAuth scopes to request.
+        For Azure managed identity, exactly one scope or resource is required.
+        For example: ``["api://{app_id}/.default"]``
+    flow : OAuthFlowType
+        Authentication flow to use. Default: CLIENT_CREDENTIALS.
+    client_secret : Optional[str]
+        Client secret (required for CLIENT_CREDENTIALS).
+    managed_identity_client_id : Optional[str]
+        Client ID for user-assigned managed identity (AZURE_MANAGED_IDENTITY).
+    refresh_buffer_secs : Optional[int]
+        Seconds before expiry to trigger proactive refresh (default: 300).
+        Keep this well below the token TTL; if it is greater than or equal to
+        the TTL, each request refreshes the token.
+
+    Examples
+    --------
+    Client Credentials (service-to-service):
+
+    >>> config = OAuthConfig(
+    ...     issuer_url="https://login.microsoftonline.com/{tenant}/v2.0",
+    ...     client_id="app-id",
+    ...     client_secret="secret",
+    ...     scopes=["api://lancedb-api/.default"],
+    ... )
+
+    Azure Managed Identity:
+
+    >>> config = OAuthConfig(
+    ...     issuer_url="https://login.microsoftonline.com/{tenant}/v2.0",
+    ...     client_id="app-id",
+    ...     scopes=["api://lancedb-api/.default"],
+    ...     flow=OAuthFlowType.AZURE_MANAGED_IDENTITY,
+    ... )
+    """
+
+    issuer_url: str
+    client_id: str
+    scopes: List[str]
+    flow: OAuthFlowType = OAuthFlowType.CLIENT_CREDENTIALS
+    client_secret: Optional[str] = field(default=None, repr=False)
+    managed_identity_client_id: Optional[str] = None
+    refresh_buffer_secs: Optional[int] = None
@@ -56,7 +56,7 @@ from lancedb.embeddings import EmbeddingFunctionRegistry
 from lancedb.table import _normalize_progress

 from ..query import LanceVectorQueryBuilder, LanceQueryBuilder, LanceTakeQueryBuilder
-from ..table import AsyncTable, BlobMode, IndexStatistics, Query, Table, Tags
+from ..table import AsyncTable, BlobMode, Branches, IndexStatistics, Query, Table, Tags
 from ..types import BaseTokenizerType


@@ -75,6 +75,9 @@ class RemoteTable(Table):
        self._connection_state = connection_state
        self._namespace_path = list(namespace_path or [])
        self._checkout_version: Optional[int] = None
+        # The branch this handle is scoped to (None == main). Persisted so a
+        # fork/pickle reopen restores the branch instead of reverting to main.
+        self._branch: Optional[str] = None
        self._pid = os.getpid()

    def _serialized_connection_state(self) -> str:
@@ -109,9 +112,14 @@ class RemoteTable(Table):
        from lancedb import deserialize_conn

        db = deserialize_conn(self._serialized_connection_state(), for_worker=True)
-        table = db.open_table(self._name, namespace_path=self._namespace_path)
-        if self._checkout_version is not None:
-            table.checkout(self._checkout_version)
+        # Reopen on the same branch and pinned version (branch=None / version=None
+        # reproduce the plain main-latest open).
+        table = db.open_table(
+            self._name,
+            namespace_path=self._namespace_path,
+            branch=self._branch,
+            version=self._checkout_version,
+        )

        self._table_handle = table._table
        self.db_name = table.db_name
@@ -124,6 +132,7 @@ class RemoteTable(Table):
            "name": self.name,
            "namespace_path": self._namespace_path,
            "checkout_version": self._checkout_version,
+            "branch": self._branch,
        }

    def __setstate__(self, state: dict) -> None:
@@ -133,6 +142,7 @@ class RemoteTable(Table):
        self._connection_state = state["connection_state"]
        self._namespace_path = state["namespace_path"]
        self._checkout_version = state["checkout_version"]
+        self._branch = state.get("branch")
        self._pid = None

    @property
@@ -160,6 +170,34 @@ class RemoteTable(Table):
    def tags(self) -> Tags:
        return Tags(self._table)

+    @property
+    def branches(self) -> Branches:
+        """Branch management for the table.
+
+        ``create``/``checkout`` return a new table handle scoped to the branch;
+        writes on it do not affect ``main``.
+        """
+        return Branches(self)
+
+    def current_branch(self) -> Optional[str]:
+        """The branch this table handle is scoped to, or ``None`` for ``main``."""
+        return self._table.current_branch()
+
+    def _wrap_branch_handle(
+        self, async_table: AsyncTable, version: Optional[int] = None
+    ) -> "RemoteTable":
+        # A branch handle stays a RemoteTable with the same connection context.
+        # Record the branch and version pin so a fork/pickle reopen restores both.
+        handle = RemoteTable(
+            async_table,
+            self.db_name,
+            connection_state=self._connection_state,
+            namespace_path=self._namespace_path,
+        )
+        handle._branch = async_table.current_branch()
+        handle._checkout_version = version
+        return handle
+
    @cached_property
    def embedding_functions(self) -> Dict[str, EmbeddingFunctionConfig]:
        """
@@ -807,7 +845,8 @@ class RemoteTable(Table):
        """
        warnings.warn(
            "cleanup_old_versions() is a no-op on LanceDB Cloud. "
-            "Tables are automatically cleaned up and optimized."
+            "Tables are automatically cleaned up and optimized.",
+            stacklevel=2,
        )
        pass

@@ -819,7 +858,8 @@ class RemoteTable(Table):
        """
        warnings.warn(
            "compact_files() is a no-op on LanceDB Cloud. "
-            "Tables are automatically compacted and optimized."
+            "Tables are automatically compacted and optimized.",
+            stacklevel=2,
        )
        pass

@@ -836,7 +876,8 @@ class RemoteTable(Table):
        """
        warnings.warn(
            "optimize() is a no-op on LanceDB Cloud. "
-            "Indices are optimized automatically."
+            "Indices are optimized automatically.",
+            stacklevel=2,
        )
        pass

@@ -871,6 +912,10 @@ class RemoteTable(Table):
        """Not supported on LanceDB Cloud."""
        return LOOP.run(self._table.unset_lsm_write_spec())

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

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

        combined = pa.concat_tables(vector_results, **self._concat_tables_args)
@@ -86,7 +86,10 @@ def _from_list(data: list) -> Scannable:

@to_scannable.register(dict)
 def _from_dict(data: dict) -> Scannable:
-    raise ValueError("Cannot add a single dictionary to a table. Use a list.")
+    raise ValueError(
+        "Cannot create or add rows from a single dictionary. "
+        "Use a list of dictionaries instead."
+    )


@to_scannable.register(LanceModel)
@@ -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)
@@ -61,6 +61,7 @@ from .index import (
    HnswFlat,
    FTS,
 )
+from .expr import Expr
 from .merge import LanceMergeInsertBuilder
 from .pydantic import LanceModel, model_to_dict
 from .query import (
@@ -242,7 +243,10 @@ def _into_pyarrow_reader(
        raise ValueError("Cannot add a single LanceModel to a table. Use a list.")

    if isinstance(data, dict):
-        raise ValueError("Cannot add a single dictionary to a table. Use a list.")
+        raise ValueError(
+            "Cannot create or add rows from a single dictionary. "
+            "Use a list of dictionaries instead."
+        )

    if isinstance(data, list):
        # Handle empty list case
@@ -647,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),
@@ -795,6 +809,10 @@ class Table(ABC):
        """
        raise NotImplementedError

+    def current_branch(self) -> Optional[str]:
+        """The branch this table handle is scoped to, or ``None`` for ``main``."""
+        raise NotImplementedError
+
    def __len__(self) -> int:
        """The number of rows in this Table"""
        return self.count_rows(None)
@@ -1533,7 +1551,7 @@ class Table(ABC):
    ) -> MergeResult: ...

    @abstractmethod
-    def delete(self, where: str) -> DeleteResult:
+    def delete(self, where: Union[str, Expr]) -> DeleteResult:
        """Delete rows from the table.

        This can be used to delete a single row, many rows, all rows, or
@@ -1541,10 +1559,10 @@ class Table(ABC):

        Parameters
        ----------
-        where: str
-            The SQL where clause to use when deleting rows.
-
-            - For example, 'x = 2' or 'x IN (1, 2, 3)'.
+        where: str or :class:`~lancedb.expr.Expr`
+            The filter condition. Can be a SQL string or a type-safe
+            :class:`~lancedb.expr.Expr` built with :func:`~lancedb.expr.col`
+            and :func:`~lancedb.expr.lit`.

            The filter must not be empty, or it will error.

@@ -2014,6 +2032,7 @@ class LanceTable(Table):
        namespace_client: Optional[Any] = None,
        managed_versioning: Optional[bool] = None,
        pushdown_operations: Optional[set] = None,
+        route_pushdown_to_rust: bool = False,
        _async: AsyncTable = None,
    ):
        if namespace_path is None:
@@ -2023,6 +2042,14 @@ class LanceTable(Table):
        self._location = location  # Store location for use in _dataset_path
        self._namespace_client = namespace_client
        self._pushdown_operations = pushdown_operations or set()
+        # When the connection built the namespace client natively (e.g. an
+        # enterprise "rest" connection), the underlying Rust table already
+        # executes QueryTable pushdown itself -- and, unlike this Python urllib3
+        # path, it routes through the read-freshness context provider that emits
+        # the ``x-lancedb-min-timestamp`` header. So we must defer pushdown to
+        # Rust instead of calling the Python ``namespace_client.query_table``
+        # directly, or reads silently bypass read-freshness (stale results).
+        self._route_pushdown_to_rust = route_pushdown_to_rust
        if _async is not None:
            self._table = _async
        else:
@@ -2125,12 +2152,19 @@ class LanceTable(Table):

        branch = self.current_branch()
        version = None if branch is not None else self.version
-        if self._namespace_client is not None:
+        namespace_client = self._namespace_client
+        if namespace_client is None:
+            conn_uri = getattr(self._conn, "uri", "")
+            if get_uri_scheme(conn_uri) == "namespace":
+                namespace_client = self._conn.namespace_client()
+                self._namespace_client = namespace_client
+
+        if namespace_client is not None:
            table_id = self._namespace_path + [self.name]
            ds = lance.dataset(
                version=version,
                storage_options=self._conn.storage_options,
-                namespace_client=self._namespace_client,
+                namespace_client=namespace_client,
                table_id=table_id,
                **kwargs,
            )
@@ -2222,6 +2256,22 @@ class LanceTable(Table):
        """The branch this table handle is scoped to, or ``None`` for ``main``."""
        return self._table.current_branch()

+    def _wrap_branch_handle(
+        self, async_table: "AsyncTable", version: Optional[int] = None
+    ) -> "LanceTable":
+        # version is unused locally: the pin already lives on async_table and a
+        # local handle is not reopened via a serialized connection.
+        return LanceTable(
+            self._conn,
+            async_table.name,
+            namespace_path=self._namespace_path,
+            namespace_client=self._namespace_client,
+            pushdown_operations=self._pushdown_operations,
+            route_pushdown_to_rust=self._route_pushdown_to_rust,
+            location=self._location,
+            _async=async_table,
+        )
+
    def checkout(self, version: Union[int, str]):
        """Checkout a version of the table. This is an in-place operation.

@@ -2368,8 +2418,11 @@ class LanceTable(Table):
        Returns
        -------
        pa.Table"""
-        if _should_push_down_query_table(
-            self._namespace_client, self._pushdown_operations
+        if (
+            _should_push_down_query_table(
+                self._namespace_client, self._pushdown_operations
+            )
+            and not self._route_pushdown_to_rust
        ):
            return self._execute_query(Query()).read_all()

@@ -3321,6 +3374,7 @@ class LanceTable(Table):
        location: Optional[str] = None,
        namespace_client: Optional[Any] = None,
        pushdown_operations: Optional[set] = None,
+        route_pushdown_to_rust: bool = False,
    ):
        """
        Create a new table.
@@ -3383,21 +3437,24 @@ class LanceTable(Table):
        self._location = location
        self._namespace_client = namespace_client
        self._pushdown_operations = pushdown_operations or set()
+        self._route_pushdown_to_rust = route_pushdown_to_rust

        if data_storage_version is not None:
            warnings.warn(
-                "setting data_storage_version directly on create_table is deprecated. ",
+                "setting data_storage_version directly on create_table is deprecated. "
                "Use database_options instead.",
                DeprecationWarning,
+                stacklevel=2,
            )
            if storage_options is None:
                storage_options = {}
            storage_options["new_table_data_storage_version"] = data_storage_version
        if enable_v2_manifest_paths is not None:
            warnings.warn(
-                "setting enable_v2_manifest_paths directly on create_table is ",
+                "setting enable_v2_manifest_paths directly on create_table is "
                "deprecated. Use database_options instead.",
                DeprecationWarning,
+                stacklevel=2,
            )
            if storage_options is None:
                storage_options = {}
@@ -3423,8 +3480,9 @@ class LanceTable(Table):
        )
        return self

-    def delete(self, where: str) -> DeleteResult:
-        return LOOP.run(self._table.delete(where))
+    def delete(self, where: Union[str, Expr]) -> DeleteResult:
+        predicate = where._inner if isinstance(where, Expr) else where
+        return LOOP.run(self._table.delete(predicate))

    def update(
        self,
@@ -3493,6 +3551,7 @@ class LanceTable(Table):
            _should_push_down_query_table(
                self._namespace_client, self._pushdown_operations
            )
+            and not self._route_pushdown_to_rust
            and self.current_branch() is None
        ):
            from lancedb.namespace import _execute_server_side_query
@@ -3700,6 +3759,11 @@ class LanceTable(Table):
        [`AsyncTable.unset_lsm_write_spec`][lancedb.AsyncTable.unset_lsm_write_spec]."""
        return LOOP.run(self._table.unset_lsm_write_spec())

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

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

@@ -4234,6 +4307,7 @@ class AsyncTable:
        namespace_path: Optional[List[str]] = None,
        namespace_client: Optional[Any] = None,
        pushdown_operations: Optional[set] = None,
+        route_pushdown_to_rust: bool = False,
    ):
        """Create a new AsyncTable object.

@@ -4246,6 +4320,9 @@ class AsyncTable:
        self._namespace_path = namespace_path or []
        self._namespace_client = namespace_client
        self._pushdown_operations = pushdown_operations or set()
+        # See LanceTable.__init__: defer QueryTable pushdown to Rust (which emits
+        # the read-freshness header) for natively-built namespace clients.
+        self._route_pushdown_to_rust = route_pushdown_to_rust

    def _set_namespace_context(
        self,
@@ -4253,10 +4330,12 @@ class AsyncTable:
        namespace_path: Optional[List[str]] = None,
        namespace_client: Optional[Any] = None,
        pushdown_operations: Optional[set] = None,
+        route_pushdown_to_rust: bool = False,
    ) -> "AsyncTable":
        self._namespace_path = namespace_path or []
        self._namespace_client = namespace_client
        self._pushdown_operations = pushdown_operations or set()
+        self._route_pushdown_to_rust = route_pushdown_to_rust
        return self

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

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

@@ -4466,8 +4556,11 @@ class AsyncTable:
        -------
        pa.Table
        """
-        if _should_push_down_query_table(
-            self._namespace_client, self._pushdown_operations
+        if (
+            _should_push_down_query_table(
+                self._namespace_client, self._pushdown_operations
+            )
+            and not self._route_pushdown_to_rust
        ):
            return (await self._execute_query(Query())).read_all()

@@ -5151,8 +5244,11 @@ class AsyncTable:
        batch_size: Optional[int] = None,
        timeout: Optional[timedelta] = None,
    ) -> pa.RecordBatchReader:
-        if _should_push_down_query_table(
-            self._namespace_client, self._pushdown_operations
+        if (
+            _should_push_down_query_table(
+                self._namespace_client, self._pushdown_operations
+            )
+            and not self._route_pushdown_to_rust
        ):
            from lancedb.namespace import _execute_server_side_query

@@ -5214,6 +5310,7 @@ class AsyncTable:
                when_not_matched_insert_all=merge._when_not_matched_insert_all,
                when_not_matched_by_source_delete=merge._when_not_matched_by_source_delete,
                when_not_matched_by_source_condition=merge._when_not_matched_by_source_condition,
+                when_not_matched_by_source_condition_expr=merge._when_not_matched_by_source_condition_expr,
                timeout=merge._timeout,
                use_index=merge._use_index,
                use_lsm_write=merge._use_lsm_write,
@@ -5221,7 +5318,7 @@ class AsyncTable:
            ),
        )

-    async def delete(self, where: str) -> DeleteResult:
+    async def delete(self, where: Union[str, Expr]) -> DeleteResult:
        """Delete rows from the table.

        This can be used to delete a single row, many rows, all rows, or
@@ -5229,10 +5326,10 @@ class AsyncTable:

        Parameters
        ----------
-        where: str
-            The SQL where clause to use when deleting rows.
-
-            - For example, 'x = 2' or 'x IN (1, 2, 3)'.
+        where: str or :class:`~lancedb.expr.Expr`
+            The filter condition. Can be a SQL string or a type-safe
+            :class:`~lancedb.expr.Expr` built with :func:`~lancedb.expr.col`
+            and :func:`~lancedb.expr.lit`.

            The filter must not be empty, or it will error.

@@ -5271,7 +5368,8 @@ class AsyncTable:
           x      vector
        0  3  [5.0, 6.0]
        """
-        return await self._inner.delete(where)
+        predicate = where._inner if isinstance(where, Expr) else where
+        return await self._inner.delete(predicate)

    async def update(
        self,
@@ -5636,6 +5734,7 @@ class AsyncTable:
                "The 'retrain' parameter is deprecated and will be removed in a "
                "future version.",
                DeprecationWarning,
+                stacklevel=2,
            )

        return await self._inner.optimize(
@@ -5930,7 +6029,7 @@ class Branches:
        name: str,
        from_ref: Optional[str] = None,
        from_version: Optional[int] = None,
-    ) -> "LanceTable":
+    ) -> "Table":
        """Create a branch and return a handle scoped to it.

        Parameters
@@ -5947,7 +6046,7 @@ class Branches:
        )
        return self._wrap(async_table)

-    def checkout(self, name: str, version: Optional[int] = None) -> "LanceTable":
+    def checkout(self, name: str, version: Optional[int] = None) -> "Table":
        """Check out an existing branch and return a handle scoped to it.

        Parameters
@@ -5960,25 +6059,19 @@ class Branches:
            the branch's latest and stays writable.
        """
        async_table = LOOP.run(self._table.branches.checkout(name, version))
-        return self._wrap(async_table)
+        return self._wrap(async_table, version)

    def delete(self, name: str) -> None:
        """Delete a branch."""
        LOOP.run(self._table.branches.delete(name))

-    def _wrap(self, async_table: "AsyncTable") -> "LanceTable":
-        # Reuse the parent's connection + namespace context; from_inner would drop
-        # it and break identity/query routing for namespace-backed tables.
-        parent = self._parent
-        return LanceTable(
-            parent._conn,
-            async_table.name,
-            namespace_path=parent._namespace_path,
-            namespace_client=parent._namespace_client,
-            pushdown_operations=parent._pushdown_operations,
-            location=parent._location,
-            _async=async_table,
-        )
+    def _wrap(
+        self, async_table: "AsyncTable", version: Optional[int] = None
+    ) -> "Table":
+        # Delegate to the parent so the branch handle keeps its concrete type
+        # (LanceTable / RemoteTable) and connection context; `version` is the
+        # explicit pin so a remote handle can restore branch+version on reopen.
+        return self._parent._wrap_branch_handle(async_table, version)


 class AsyncTags:
@@ -177,7 +177,10 @@ def flatten_columns(tbl: pa.Table, flatten: Optional[Union[int, bool]] = None):
                continue
            else:
                break
-    elif isinstance(flatten, int):
+    # `bool` is a subclass of `int`, so guard against it explicitly: `flatten=False`
+    # (and `None`) must mean "do not flatten" rather than falling into the integer
+    # branch and raising on the `flatten <= 0` check.
+    elif isinstance(flatten, int) and not isinstance(flatten, bool):
        if flatten <= 0:
            raise ValueError(
                "Please specify a positive integer for flatten or the boolean "
@@ -373,9 +376,15 @@ def _(value: list):
@value_to_sql.register(dict)
 def _(value: dict):
    # https://datafusion.apache.org/user-guide/sql/scalar_functions.html#named-struct
+    # Render the field name through value_to_sql(str(...)) as well so that keys
+    # containing characters meaningful in SQL (e.g. a single quote) are escaped
+    # the same way string values are. A bare f"'{k}'" would emit invalid SQL for
+    # a key like "it's".
    return (
        "named_struct("
-        + ", ".join(f"'{k}', {value_to_sql(v)}" for k, v in value.items())
+        + ", ".join(
+            f"{value_to_sql(str(k))}, {value_to_sql(v)}" for k, v in value.items()
+        )
        + ")"
    )

@@ -509,3 +518,38 @@ def batch_to_tensor_rows(batch: pa.RecordBatch):
    stacked = torch.tensor(numpy.column_stack(columns))
    rows = list(stacked.unbind(dim=0))
    return rows
+
+
+def batch_to_tensor_dict(batch: pa.RecordBatch):
+    """
+    Convert a PyArrow RecordBatch to a list of per-row dicts of PyTorch Tensors.
+
+    Each column is first converted to a 1-D tensor (zero-copy via DLPack), then
+    sliced per row. The result is a list whose length is ``batch.num_rows`` and
+    whose items are dicts keyed by column name. This shape composes directly
+    with PyTorch's default ``DataLoader`` collate, which stacks the per-row
+    dicts back into a dict of batched tensors.
+
+    Fails if torch is not installed.
+    Fails if a column's data type is not supported by PyTorch.
+
+    Parameters
+    ----------
+    batch : pa.RecordBatch
+        The record batch to convert.
+
+    Returns
+    -------
+    list[dict[str, torch.Tensor]]
+        One per-row dict per row in the batch. Each dict maps column name to a
+        0-D tensor view into the column.
+    """
+    torch = attempt_import_or_raise("torch", "torch")
+    tensors = {
+        name: torch.from_dlpack(col)
+        for name, col in zip(batch.schema.names, batch.columns)
+    }
+    return [
+        {name: tensor[i] for name, tensor in tensors.items()}
+        for i in range(batch.num_rows)
+    ]
@@ -2,6 +2,7 @@
 # SPDX-FileCopyrightText: Copyright The LanceDB Authors

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

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


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


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

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

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


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

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

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

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

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


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


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


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


@@ -91,7 +91,9 @@ async def test_create_scalar_index(some_table: AsyncTable):
    # Can recreate if replace=True
    await some_table.create_index("id", replace=True)
    indices = await some_table.list_indices()
-    assert str(indices) == '[Index(BTree, columns=["id"], name="id_idx")]'
+    assert str(indices).startswith(
+        '[IndexConfig(name="id_idx", index_type="BTree", columns=["id"]'
+    )
    assert len(indices) == 1
    assert indices[0].index_type == "BTree"
    assert indices[0].columns == ["id"]
@@ -106,6 +108,27 @@ async def test_create_scalar_index(some_table: AsyncTable):
    assert len(indices) == 0


+@pytest.mark.asyncio
+async def test_index_config_repr(db_async):
+    # Use >= 1000 rows so the thousands separator in the repr is exercised.
+    nrows = 1500
+    table = await db_async.create_table(
+        "repr_table", pa.Table.from_pydict({"id": list(range(nrows))})
+    )
+    await table.create_index("id", config=BTree())
+    indices = await table.list_indices()
+    assert len(indices) == 1
+
+    r = repr(indices[0])
+    assert r.startswith('IndexConfig(name="id_idx", index_type="BTree", columns=["id"]')
+    # Integer counts use `_` thousands separators (valid Python int syntax).
+    assert "num_indexed_rows=1_500" in r
+    assert "num_unindexed_rows=0" in r
+    # created_at renders as a datetime so the value round-trips.
+    assert "created_at=datetime.datetime(" in r
+    assert r.endswith(")")
+
+
@pytest.mark.asyncio
 async def test_create_nested_scalar_index_lists_canonical_paths(db_async):
    metadata_type = pa.struct(
@@ -198,7 +221,9 @@ async def test_create_nested_scalar_index_lists_canonical_paths(db_async):
 async def test_create_fixed_size_binary_index(some_table: AsyncTable):
    await some_table.create_index("fsb", config=BTree())
    indices = await some_table.list_indices()
-    assert str(indices) == '[Index(BTree, columns=["fsb"], name="fsb_idx")]'
+    assert str(indices).startswith(
+        '[IndexConfig(name="fsb_idx", index_type="BTree", columns=["fsb"]'
+    )
    assert len(indices) == 1
    assert indices[0].index_type == "BTree"
    assert indices[0].columns == ["fsb"]
@@ -247,7 +272,9 @@ async def test_create_bitmap_index(some_table: AsyncTable):
 async def test_create_label_list_index(some_table: AsyncTable):
    await some_table.create_index("tags", config=LabelList())
    indices = await some_table.list_indices()
-    assert str(indices) == '[Index(LabelList, columns=["tags"], name="tags_idx")]'
+    assert str(indices).startswith(
+        '[IndexConfig(name="tags_idx", index_type="LabelList", columns=["tags"]'
+    )
    plan = await some_table.query().where("array_has(tags, 'tag0')").explain_plan()
    assert "ScalarIndexQuery" in plan

@@ -262,7 +289,9 @@ async def test_create_large_list_label_list_index(db_async):

    await table.create_index("tags", config=LabelList())
    indices = await table.list_indices()
-    assert str(indices) == '[Index(LabelList, columns=["tags"], name="tags_idx")]'
+    assert str(indices).startswith(
+        '[IndexConfig(name="tags_idx", index_type="LabelList", columns=["tags"]'
+    )
    plan = await table.query().where("array_has(tags, 'shared')").explain_plan()
    assert "ScalarIndexQuery" in plan

@@ -299,7 +328,9 @@ async def test_create_label_list_index_rejects_list_struct(db_async):
 async def test_full_text_search_index(some_table: AsyncTable):
    await some_table.create_index("tags", config=FTS(with_position=False))
    indices = await some_table.list_indices()
-    assert str(indices) == '[Index(FTS, columns=["tags"], name="tags_idx")]'
+    assert str(indices).startswith(
+        '[IndexConfig(name="tags_idx", index_type="FTS", columns=["tags"]'
+    )

    await some_table.prewarm_index("tags_idx")

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

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

 import tempfile
 import shutil
-import sys
+import importlib
 import pytest
 import pyarrow as pa
 import lancedb
 from lance_namespace.errors import NamespaceNotEmptyError, TableNotFoundError
+from lancedb.namespace import _MAX_QUERY_K
 from lancedb.table import AsyncTable, LanceTable


@@ -65,6 +66,9 @@ def _namespace_lance_table(namespace_client: _NamespaceClient) -> LanceTable:
    table._namespace_path = ["geneva"]
    table._namespace_client = namespace_client
    table._pushdown_operations = {"QueryTable"}
+    # This test exercises the Python-side pushdown path (non-native client), so
+    # pushdown is not routed to Rust.
+    table._route_pushdown_to_rust = False
    return table


@@ -100,6 +104,40 @@ class TestNamespaceConnection:
        assert isinstance(db, lancedb.LanceNamespaceDBConnection)
        assert len(list(db.table_names())) == 0

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

-    def test_rename_table_not_supported(self):
-        """Test that rename_table raises NotImplementedError."""
+    def test_rename_table(self):
+        """Test that rename_table renames a table in the namespace.
+
+        The `dir` namespace implementation in lance-namespace-impls does not
+        implement `rename_table` yet (only the `rest` backend does), so it
+        currently falls back to the default trait method which raises
+        NotSupported. This is expected to start passing once the `dir`
+        backend gains rename_table support upstream.
+        """
        db = lancedb.connect_namespace("dir", {"root": self.temp_dir})

        # Create a child namespace first
@@ -273,9 +318,14 @@ class TestNamespaceConnection:
        )
        db.create_table("old_name", schema=schema, namespace_path=["test_ns"])

-        # Rename should raise NotImplementedError
-        with pytest.raises(NotImplementedError, match="rename_table is not supported"):
-            db.rename_table("old_name", "new_name")
+        # Rename the table within the same namespace
+        with pytest.raises(NotImplementedError, match="rename_table not implemented"):
+            db.rename_table(
+                "old_name",
+                "new_name",
+                cur_namespace_path=["test_ns"],
+                new_namespace_path=["test_ns"],
+            )

    def test_drop_all_tables(self):
        """Test dropping all tables through namespace."""
@@ -549,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):
@@ -793,6 +898,39 @@ class TestPushdownOperations:
        db = lancedb.connect_namespace("dir", {"root": self.temp_dir})
        assert len(db._namespace_client_pushdown_operations) == 0

+    def test_route_pushdown_to_rust_for_native_rest(self):
+        """A natively-built rest connection must defer QueryTable pushdown to
+        Rust so reads carry the x-lancedb-min-timestamp read-freshness header."""
+        db = lancedb.connect_namespace(
+            "rest",
+            {"uri": "http://localhost:12345"},
+            namespace_client_pushdown_operations=["QueryTable"],
+        )
+        assert db._route_pushdown_to_rust is True
+
+    def test_route_pushdown_to_rust_for_native_dir(self):
+        """The sync dir connection is natively built and defers QueryTable
+        pushdown to Rust."""
+        db = lancedb.connect_namespace("dir", {"root": self.temp_dir})
+        assert db._route_pushdown_to_rust is True
+
+    def test_async_route_pushdown_to_rust_for_native_rest(self):
+        """The async connection must not silently bypass the read-freshness fix:
+        a natively-built rest connection defers pushdown to Rust (regression test
+        for the async path omitting the freshness header)."""
+        db = lancedb.connect_namespace_async(
+            "rest",
+            {"uri": "http://localhost:12345"},
+            namespace_client_pushdown_operations=["QueryTable"],
+        )
+        assert db._route_pushdown_to_rust is True
+
+    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 True
+
    def test_lance_table_to_arrow_uses_query_pushdown(self):
        namespace_client = _NamespaceClient()
        table = _namespace_lance_table(namespace_client)
@@ -804,10 +942,13 @@ class TestPushdownOperations:
            ["geneva", "hist"],
            ["geneva", "hist"],
        ]
+        # Unlimited reads cap k at i32::MAX (the namespace query_table `k`
+        # field is i32); sys.maxsize would overflow the Rust binding.
        assert [request.k for request in namespace_client.requests] == [
-            sys.maxsize,
-            sys.maxsize,
+            _MAX_QUERY_K,
+            _MAX_QUERY_K,
        ]
+        assert all(r.k <= 2**31 - 1 for r in namespace_client.requests)


@pytest.mark.asyncio
@@ -862,10 +1003,13 @@ class TestAsyncPushdownOperations:
            ["geneva", "hist"],
            ["geneva", "hist"],
        ]
+        # Unlimited reads cap k at i32::MAX (the namespace query_table `k`
+        # field is i32); sys.maxsize would overflow the Rust binding.
        assert [request.k for request in namespace_client.requests] == [
-            sys.maxsize,
-            sys.maxsize,
+            _MAX_QUERY_K,
+            _MAX_QUERY_K,
        ]
+        assert all(r.k <= 2**31 - 1 for r in namespace_client.requests)


 def test_local_table_to_arrow_and_to_pandas_are_unchanged(tmp_path):
@@ -0,0 +1,686 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+"""Regression matrix for nested field support across LanceDB Python APIs.
+
+Covers the lifecycle described in lancedb/lancedb#3406:
+  - Nested scalar, vector, and FTS index creation with full dotted paths
+  - list_indices / index_stats return canonical full paths (not leaf names)
+  - search, filter, append, optimize behaviour
+  - Field-name edge cases: mixed case, literal-dot field names, same-name leaves
+  - Both sync and async Python table APIs
+
+The matrix uses the following field-name variants from the acceptance criteria:
+  - rowId              (camelCase top-level)
+  - `row-id`           (hyphenated top-level, escaped)
+  - parent.`leaf.name` (struct leaf whose name contains a literal dot)
+  - MetaData.userId    (mixed-case nested path)
+  - `meta-data`.`user-id`  (hyphenated struct with hyphenated leaf)
+
+Note: Lance forbids top-level field names that contain a '.', so the literal-dot
+edge case is exercised via a struct leaf field (parent.`leaf.name`) instead.
+"""
+
+from datetime import timedelta
+
+import pyarrow as pa
+import pytest
+import pytest_asyncio
+
+import lancedb
+from lancedb.db import AsyncConnection, DBConnection
+from lancedb.index import BTree, FTS, IvfPq
+
+
+# ---------------------------------------------------------------------------
+# Constants
+# ---------------------------------------------------------------------------
+
+DIM = 8
+# IvfPq requires at least num_partitions * 256 rows by default; keeping rows
+# small means we must drop num_sub_vectors and num_partitions very low.
+NROWS = 256
+
+
+def _vec(row: int) -> list:
+    return [float((row * DIM + i) % 256) for i in range(DIM)]
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+def sync_db(tmp_path) -> DBConnection:
+    return lancedb.connect(tmp_path)
+
+
+@pytest_asyncio.fixture
+async def async_db(tmp_path) -> AsyncConnection:
+    return await lancedb.connect_async(
+        tmp_path, read_consistency_interval=timedelta(seconds=0)
+    )
+
+
+# ---------------------------------------------------------------------------
+# Schema / data builders
+# ---------------------------------------------------------------------------
+
+
+def _nested_scalar_schema() -> pa.Schema:
+    """Schema with nested scalar fields covering the acceptance-criteria names.
+
+    Top-level columns:
+      - rowId       int32  (camelCase top-level)
+      - row-id      int32  (hyphenated top-level name)
+      - MetaData    struct{userId int32}   (mixed-case nested path)
+      - meta-data   struct{user-id int32}  (hyphenated struct + hyphenated leaf)
+
+    Lance disallows top-level field names that contain '.' (e.g. a field
+    literally named 'a.b'), so that edge case is tested separately using
+    _literal_dot_schema() below.
+    """
+    return pa.schema(
+        [
+            pa.field("rowId", pa.int32()),
+            pa.field("row-id", pa.int32()),
+            pa.field(
+                "MetaData",
+                pa.struct([pa.field("userId", pa.int32())]),
+            ),
+            pa.field(
+                "meta-data",
+                pa.struct([pa.field("user-id", pa.int32())]),
+            ),
+        ]
+    )
+
+
+def _nested_scalar_data(nrows: int = NROWS) -> pa.Table:
+    schema = _nested_scalar_schema()
+    return pa.table(
+        {
+            "rowId": pa.array(list(range(nrows)), pa.int32()),
+            "row-id": pa.array(list(range(nrows)), pa.int32()),
+            "MetaData": pa.array(
+                [{"userId": i} for i in range(nrows)],
+                type=pa.struct([pa.field("userId", pa.int32())]),
+            ),
+            "meta-data": pa.array(
+                [{"user-id": i} for i in range(nrows)],
+                type=pa.struct([pa.field("user-id", pa.int32())]),
+            ),
+        },
+        schema=schema,
+    )
+
+
+def _literal_dot_schema() -> pa.Schema:
+    """Schema where a struct *leaf* field is named with a literal dot.
+
+    The path used in the index API is ``parent.`leaf.name` ``.
+    """
+    return pa.schema(
+        [
+            pa.field("id", pa.int32()),
+            pa.field(
+                "parent",
+                pa.struct([pa.field("leaf.name", pa.int32())]),
+            ),
+        ]
+    )
+
+
+def _literal_dot_data(nrows: int = NROWS) -> pa.Table:
+    parent_type = pa.struct([pa.field("leaf.name", pa.int32())])
+    return pa.table(
+        {
+            "id": pa.array(list(range(nrows)), pa.int32()),
+            "parent": pa.array(
+                [{"leaf.name": i} for i in range(nrows)],
+                type=parent_type,
+            ),
+        },
+        schema=_literal_dot_schema(),
+    )
+
+
+def _same_leaf_schema() -> pa.Schema:
+    return pa.schema(
+        [
+            pa.field("StructA", pa.struct([pa.field("userId", pa.int32())])),
+            pa.field("StructB", pa.struct([pa.field("userId", pa.int32())])),
+        ]
+    )
+
+
+def _same_leaf_data(nrows: int = NROWS) -> pa.Table:
+    t = pa.struct([pa.field("userId", pa.int32())])
+    return pa.table(
+        {
+            "StructA": pa.array([{"userId": i} for i in range(nrows)], type=t),
+            "StructB": pa.array([{"userId": i * 10} for i in range(nrows)], type=t),
+        },
+        schema=_same_leaf_schema(),
+    )
+
+
+def _nested_vector_schema() -> pa.Schema:
+    return pa.schema(
+        [
+            pa.field("id", pa.int32()),
+            pa.field(
+                "image",
+                pa.struct([pa.field("embedding", pa.list_(pa.float32(), DIM))]),
+            ),
+            pa.field(
+                "MetaData",
+                pa.struct([pa.field("userId", pa.int32())]),
+            ),
+        ]
+    )
+
+
+def _nested_vector_data(nrows: int = NROWS) -> pa.Table:
+    embedding_type = pa.list_(pa.float32(), DIM)
+    image_type = pa.struct([pa.field("embedding", embedding_type)])
+    meta_type = pa.struct([pa.field("userId", pa.int32())])
+    return pa.table(
+        {
+            "id": pa.array(list(range(nrows)), pa.int32()),
+            "image": pa.array(
+                [{"embedding": _vec(i)} for i in range(nrows)],
+                type=image_type,
+            ),
+            "MetaData": pa.array(
+                [{"userId": i} for i in range(nrows)],
+                type=meta_type,
+            ),
+        },
+        schema=_nested_vector_schema(),
+    )
+
+
+def _nested_fts_schema() -> pa.Schema:
+    return pa.schema(
+        [
+            pa.field("id", pa.int32()),
+            pa.field(
+                "payload",
+                pa.struct([pa.field("text", pa.utf8())]),
+            ),
+            pa.field(
+                "MetaData",
+                pa.struct([pa.field("userId", pa.int32())]),
+            ),
+        ]
+    )
+
+
+def _nested_fts_data(nrows: int = NROWS) -> pa.Table:
+    words = ["alpha", "bravo", "charlie", "delta", "echo"]
+    payload_type = pa.struct([pa.field("text", pa.utf8())])
+    meta_type = pa.struct([pa.field("userId", pa.int32())])
+    return pa.table(
+        {
+            "id": pa.array(list(range(nrows)), pa.int32()),
+            "payload": pa.array(
+                [{"text": words[i % len(words)]} for i in range(nrows)],
+                type=payload_type,
+            ),
+            "MetaData": pa.array(
+                [{"userId": i} for i in range(nrows)],
+                type=meta_type,
+            ),
+        },
+        schema=_nested_fts_schema(),
+    )
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _columns_by_name_sync(tbl) -> dict:
+    return {idx.name: idx.columns for idx in tbl.list_indices()}
+
+
+async def _columns_by_name_async(tbl) -> dict:
+    return {idx.name: idx.columns for idx in await tbl.list_indices()}
+
+
+# ===========================================================================
+# SYNC TESTS
+# ===========================================================================
+#
+# The sync LanceTable API uses:
+#   - create_scalar_index(column, ...)  for scalar (BTree/Bitmap/LabelList) indices
+#   - create_fts_index(column, ...)     for full-text-search indices
+#   - create_index(...)                 for vector indices (older positional API)
+# ===========================================================================
+
+
+class TestNestedScalarIndexSync:
+    """Sync regression matrix for nested scalar (BTree) indices."""
+
+    def test_top_level_camelcase_field(self, sync_db):
+        """list_indices must return the full camelCase field name."""
+        tbl = sync_db.create_table("t", _nested_scalar_data())
+        tbl.create_scalar_index("rowId", index_type="BTREE", name="rowid_idx")
+        col_map = _columns_by_name_sync(tbl)
+        assert col_map["rowid_idx"] == ["rowId"], (
+            "list_indices must return 'rowId', not a truncated leaf name"
+        )
+
+    def test_top_level_hyphenated_field_escaped(self, sync_db):
+        """Top-level field 'row-id' (hyphenated) accessed via escaped path."""
+        tbl = sync_db.create_table("t", _nested_scalar_data())
+        tbl.create_scalar_index("`row-id`", index_type="BTREE", name="rowid_hyph_idx")
+        col_map = _columns_by_name_sync(tbl)
+        assert col_map["rowid_hyph_idx"] == ["`row-id`"], (
+            "list_indices must return escaped path '`row-id`'"
+        )
+
+    def test_struct_leaf_literal_dot_field_escaped(self, sync_db):
+        """Struct leaf with a literal-dot name: parent.`leaf.name`.
+
+        The index listing must use the full escaped path, not just the leaf.
+        """
+        tbl = sync_db.create_table("t", _literal_dot_data())
+        tbl.create_scalar_index(
+            "parent.`leaf.name`", index_type="BTREE", name="leaf_dot_idx"
+        )
+        col_map = _columns_by_name_sync(tbl)
+        assert col_map["leaf_dot_idx"] == ["parent.`leaf.name`"], (
+            "list_indices must return 'parent.`leaf.name`', not just '`leaf.name`'"
+        )
+
+    def test_nested_mixed_case_path(self, sync_db):
+        """Nested path MetaData.userId (mixed case) must appear as full path."""
+        tbl = sync_db.create_table("t", _nested_scalar_data())
+        tbl.create_scalar_index(
+            "MetaData.userId", index_type="BTREE", name="metadata_userid_idx"
+        )
+        col_map = _columns_by_name_sync(tbl)
+        assert col_map["metadata_userid_idx"] == ["MetaData.userId"], (
+            "list_indices must return 'MetaData.userId', not leaf 'userId'"
+        )
+
+    def test_nested_hyphenated_path_escaped(self, sync_db):
+        """`meta-data`.`user-id` path with both parts escaped."""
+        tbl = sync_db.create_table("t", _nested_scalar_data())
+        tbl.create_scalar_index(
+            "`meta-data`.`user-id`", index_type="BTREE", name="metauid_idx"
+        )
+        col_map = _columns_by_name_sync(tbl)
+        assert col_map["metauid_idx"] == ["`meta-data`.`user-id`"], (
+            "list_indices must return '`meta-data`.`user-id`', not 'user-id'"
+        )
+
+    def test_filter_on_nested_mixed_case(self, sync_db):
+        """WHERE filter on a nested dotted path works after index creation."""
+        tbl = sync_db.create_table("t", _nested_scalar_data())
+        tbl.create_scalar_index(
+            "MetaData.userId", index_type="BTREE", name="metadata_userid_idx"
+        )
+        rows = tbl.search().where("MetaData.userId = 5").to_list()
+        assert len(rows) == 1
+        assert rows[0]["MetaData"]["userId"] == 5
+
+    def test_append_and_list_indices_stable(self, sync_db):
+        """After appending rows the index listing must remain unchanged."""
+        tbl = sync_db.create_table("t", _nested_scalar_data())
+        tbl.create_scalar_index(
+            "MetaData.userId", index_type="BTREE", name="meta_uid_idx"
+        )
+        tbl.add(_nested_scalar_data(nrows=4))
+        col_map = _columns_by_name_sync(tbl)
+        assert col_map["meta_uid_idx"] == ["MetaData.userId"]
+
+    def test_optimize_and_list_indices_stable(self, tmp_path):
+        """After optimize the index listing must still show full paths."""
+        db = lancedb.connect(tmp_path / "opt_db")
+        tbl = db.create_table("t", _nested_scalar_data())
+        tbl.create_scalar_index(
+            "MetaData.userId", index_type="BTREE", name="meta_uid_idx"
+        )
+        tbl.add(_nested_scalar_data(nrows=4))
+        tbl.optimize()
+        col_map = _columns_by_name_sync(tbl)
+        assert col_map["meta_uid_idx"] == ["MetaData.userId"]
+
+    def test_same_name_leaves_are_distinct(self, sync_db):
+        """Two structs sharing a leaf name must produce distinct index paths."""
+        tbl = sync_db.create_table("same_leaf", _same_leaf_data())
+        tbl.create_scalar_index(
+            "StructA.userId", index_type="BTREE", name="a_userid_idx"
+        )
+        tbl.create_scalar_index(
+            "StructB.userId", index_type="BTREE", name="b_userid_idx"
+        )
+        col_map = _columns_by_name_sync(tbl)
+        assert col_map["a_userid_idx"] == ["StructA.userId"]
+        assert col_map["b_userid_idx"] == ["StructB.userId"]
+
+    def test_index_stats_canonical_path(self, sync_db):
+        """index_stats round-trip: create on nested field, verify row count."""
+        tbl = sync_db.create_table("t", _nested_scalar_data())
+        tbl.create_scalar_index(
+            "MetaData.userId", index_type="BTREE", name="meta_uid_idx"
+        )
+        stats = tbl.index_stats("meta_uid_idx")
+        assert stats is not None
+        assert stats.index_type == "BTREE"
+        assert stats.num_indexed_rows == NROWS
+
+
+class TestNestedVectorIndexSync:
+    """Sync regression matrix for nested vector (IvfPq) indices."""
+
+    def test_nested_vector_index_full_path(self, sync_db):
+        """Listing after vector index creation must use the full dotted path."""
+        tbl = sync_db.create_table("vt", _nested_vector_data())
+        tbl.create_index(
+            num_partitions=2,
+            num_sub_vectors=2,
+            vector_column_name="image.embedding",
+            name="image_emb_idx",
+        )
+        col_map = _columns_by_name_sync(tbl)
+        assert col_map["image_emb_idx"] == ["image.embedding"], (
+            "list_indices must return 'image.embedding', not leaf 'embedding'"
+        )
+
+    def test_nested_vector_search(self, sync_db):
+        """Vector search on nested embedding field must return results."""
+        tbl = sync_db.create_table("vt", _nested_vector_data())
+        tbl.create_index(
+            num_partitions=2,
+            num_sub_vectors=2,
+            vector_column_name="image.embedding",
+            name="image_emb_idx",
+        )
+        results = (
+            tbl.search(_vec(0), vector_column_name="image.embedding").limit(5).to_list()
+        )
+        assert len(results) > 0
+
+    def test_nested_vector_index_stats(self, sync_db):
+        """index_stats for a nested vector index must reflect correct row count."""
+        tbl = sync_db.create_table("vt", _nested_vector_data())
+        tbl.create_index(
+            num_partitions=2,
+            num_sub_vectors=2,
+            vector_column_name="image.embedding",
+            name="image_emb_idx",
+        )
+        stats = tbl.index_stats("image_emb_idx")
+        assert stats is not None
+        assert stats.num_indexed_rows == NROWS
+
+    def test_nested_vector_append_optimize(self, tmp_path):
+        """After append and optimize the vector index listing must be stable."""
+        db = lancedb.connect(tmp_path / "vec_opt_db")
+        tbl = db.create_table("vt", _nested_vector_data())
+        tbl.create_index(
+            num_partitions=2,
+            num_sub_vectors=2,
+            vector_column_name="image.embedding",
+            name="image_emb_idx",
+        )
+        tbl.add(_nested_vector_data(nrows=4))
+        tbl.optimize()
+        col_map = _columns_by_name_sync(tbl)
+        assert col_map["image_emb_idx"] == ["image.embedding"]
+
+
+class TestNestedFTSIndexSync:
+    """Sync regression matrix for nested FTS indices."""
+
+    def test_nested_fts_index_full_path(self, sync_db):
+        """FTS index on payload.text must be listed with the full path."""
+        tbl = sync_db.create_table("ft", _nested_fts_data())
+        tbl.create_fts_index("payload.text", name="payload_text_idx")
+        col_map = _columns_by_name_sync(tbl)
+        assert col_map["payload_text_idx"] == ["payload.text"], (
+            "list_indices must return 'payload.text', not leaf 'text'"
+        )
+
+    def test_nested_fts_search(self, sync_db):
+        """FTS search on a nested text field must return correct results."""
+        tbl = sync_db.create_table("ft", _nested_fts_data())
+        tbl.create_fts_index("payload.text", name="payload_text_idx")
+        results = (
+            tbl.search("alpha", query_type="fts", fts_columns="payload.text")
+            .limit(10)
+            .to_list()
+        )
+        assert len(results) > 0
+        assert all(row["payload"]["text"] == "alpha" for row in results)
+
+    def test_nested_fts_append_optimize(self, tmp_path):
+        """After append and optimize the FTS index listing must be stable."""
+        db = lancedb.connect(tmp_path / "fts_opt_db")
+        tbl = db.create_table("ft", _nested_fts_data())
+        tbl.create_fts_index("payload.text", name="payload_text_idx")
+        tbl.add(_nested_fts_data(nrows=4))
+        tbl.optimize()
+        col_map = _columns_by_name_sync(tbl)
+        assert col_map["payload_text_idx"] == ["payload.text"]
+
+
+# ===========================================================================
+# ASYNC TESTS
+# ===========================================================================
+#
+# The async AsyncTable API uses create_index(column, config=...) uniformly
+# for scalar, vector, and FTS indices.
+# ===========================================================================
+
+
+class TestNestedScalarIndexAsync:
+    """Async regression matrix for nested scalar (BTree) indices."""
+
+    @pytest.mark.asyncio
+    async def test_top_level_camelcase_field(self, async_db):
+        """list_indices must return the full camelCase field name."""
+        tbl = await async_db.create_table("t", _nested_scalar_data())
+        await tbl.create_index("rowId", config=BTree(), name="rowid_idx")
+        col_map = await _columns_by_name_async(tbl)
+        assert col_map["rowid_idx"] == ["rowId"]
+
+    @pytest.mark.asyncio
+    async def test_top_level_hyphenated_field_escaped(self, async_db):
+        """Hyphenated top-level field accessed via escaped path."""
+        tbl = await async_db.create_table("t", _nested_scalar_data())
+        await tbl.create_index("`row-id`", config=BTree(), name="rowid_hyph_idx")
+        col_map = await _columns_by_name_async(tbl)
+        assert col_map["rowid_hyph_idx"] == ["`row-id`"]
+
+    @pytest.mark.asyncio
+    async def test_struct_leaf_literal_dot_field_escaped(self, async_db):
+        """Struct leaf with a literal-dot name: parent.`leaf.name`."""
+        tbl = await async_db.create_table("t", _literal_dot_data())
+        await tbl.create_index(
+            "parent.`leaf.name`", config=BTree(), name="leaf_dot_idx"
+        )
+        col_map = await _columns_by_name_async(tbl)
+        assert col_map["leaf_dot_idx"] == ["parent.`leaf.name`"]
+
+    @pytest.mark.asyncio
+    async def test_nested_mixed_case_path(self, async_db):
+        """Mixed-case nested path MetaData.userId must appear as full path."""
+        tbl = await async_db.create_table("t", _nested_scalar_data())
+        await tbl.create_index(
+            "MetaData.userId", config=BTree(), name="metadata_userid_idx"
+        )
+        col_map = await _columns_by_name_async(tbl)
+        assert col_map["metadata_userid_idx"] == ["MetaData.userId"]
+
+    @pytest.mark.asyncio
+    async def test_nested_hyphenated_path_escaped(self, async_db):
+        """`meta-data`.`user-id` path with both parts escaped."""
+        tbl = await async_db.create_table("t", _nested_scalar_data())
+        await tbl.create_index(
+            "`meta-data`.`user-id`", config=BTree(), name="metauid_idx"
+        )
+        col_map = await _columns_by_name_async(tbl)
+        assert col_map["metauid_idx"] == ["`meta-data`.`user-id`"]
+
+    @pytest.mark.asyncio
+    async def test_filter_on_nested_mixed_case(self, async_db):
+        """WHERE filter on a nested dotted path works after index creation."""
+        tbl = await async_db.create_table("t", _nested_scalar_data())
+        await tbl.create_index(
+            "MetaData.userId", config=BTree(), name="metadata_userid_idx"
+        )
+        rows = await tbl.query().where("MetaData.userId = 5").to_list()
+        assert len(rows) == 1
+        assert rows[0]["MetaData"]["userId"] == 5
+
+    @pytest.mark.asyncio
+    async def test_index_stats_canonical_path(self, async_db):
+        """index_stats round-trip: create on nested field, verify stats."""
+        tbl = await async_db.create_table("t", _nested_scalar_data())
+        await tbl.create_index("MetaData.userId", config=BTree(), name="meta_uid_idx")
+        stats = await tbl.index_stats("meta_uid_idx")
+        assert stats is not None
+        assert stats.index_type == "BTREE"
+        assert stats.num_indexed_rows == NROWS
+
+    @pytest.mark.asyncio
+    async def test_append_and_list_indices_stable(self, async_db):
+        """After appending rows the index listing must remain unchanged."""
+        tbl = await async_db.create_table("t", _nested_scalar_data())
+        await tbl.create_index("MetaData.userId", config=BTree(), name="meta_uid_idx")
+        await tbl.add(_nested_scalar_data(nrows=4))
+        col_map = await _columns_by_name_async(tbl)
+        assert col_map["meta_uid_idx"] == ["MetaData.userId"]
+
+    @pytest.mark.asyncio
+    async def test_optimize_and_list_indices_stable(self, tmp_path):
+        """After optimize the index listing must still show full paths."""
+        db = await lancedb.connect_async(
+            tmp_path / "opt_db", read_consistency_interval=timedelta(seconds=0)
+        )
+        tbl = await db.create_table("t", _nested_scalar_data())
+        await tbl.create_index("MetaData.userId", config=BTree(), name="meta_uid_idx")
+        await tbl.add(_nested_scalar_data(nrows=4))
+        await tbl.optimize()
+        col_map = await _columns_by_name_async(tbl)
+        assert col_map["meta_uid_idx"] == ["MetaData.userId"]
+
+    @pytest.mark.asyncio
+    async def test_same_name_leaves_are_distinct(self, async_db):
+        """Two structs sharing a leaf name must produce distinct index paths."""
+        tbl = await async_db.create_table("same_leaf", _same_leaf_data())
+        await tbl.create_index("StructA.userId", config=BTree(), name="a_userid_idx")
+        await tbl.create_index("StructB.userId", config=BTree(), name="b_userid_idx")
+        col_map = await _columns_by_name_async(tbl)
+        assert col_map["a_userid_idx"] == ["StructA.userId"]
+        assert col_map["b_userid_idx"] == ["StructB.userId"]
+
+
+class TestNestedVectorIndexAsync:
+    """Async regression matrix for nested vector (IvfPq) indices."""
+
+    @pytest.mark.asyncio
+    async def test_nested_vector_index_full_path(self, async_db):
+        """Listing after vector index creation must use the full dotted path."""
+        tbl = await async_db.create_table("vt", _nested_vector_data())
+        await tbl.create_index(
+            "image.embedding",
+            config=IvfPq(num_partitions=2, num_sub_vectors=2),
+            name="image_emb_idx",
+        )
+        col_map = await _columns_by_name_async(tbl)
+        assert col_map["image_emb_idx"] == ["image.embedding"]
+
+    @pytest.mark.asyncio
+    async def test_nested_vector_search(self, async_db):
+        """Vector search on nested embedding field must return results."""
+        tbl = await async_db.create_table("vt", _nested_vector_data())
+        await tbl.create_index(
+            "image.embedding",
+            config=IvfPq(num_partitions=2, num_sub_vectors=2),
+            name="image_emb_idx",
+        )
+        results = (
+            await tbl.query()
+            .nearest_to(_vec(0))
+            .column("image.embedding")
+            .limit(5)
+            .to_list()
+        )
+        assert len(results) > 0
+
+    @pytest.mark.asyncio
+    async def test_nested_vector_index_stats(self, async_db):
+        """index_stats for a nested vector index must reflect correct row count."""
+        tbl = await async_db.create_table("vt", _nested_vector_data())
+        await tbl.create_index(
+            "image.embedding",
+            config=IvfPq(num_partitions=2, num_sub_vectors=2),
+            name="image_emb_idx",
+        )
+        stats = await tbl.index_stats("image_emb_idx")
+        assert stats is not None
+        assert stats.num_indexed_rows == NROWS
+
+    @pytest.mark.asyncio
+    async def test_nested_vector_append_optimize(self, tmp_path):
+        """After append and optimize the vector index listing must be stable."""
+        db = await lancedb.connect_async(
+            tmp_path / "vec_opt_db", read_consistency_interval=timedelta(seconds=0)
+        )
+        tbl = await db.create_table("vt", _nested_vector_data())
+        await tbl.create_index(
+            "image.embedding",
+            config=IvfPq(num_partitions=2, num_sub_vectors=2),
+            name="image_emb_idx",
+        )
+        await tbl.add(_nested_vector_data(nrows=4))
+        await tbl.optimize()
+        col_map = await _columns_by_name_async(tbl)
+        assert col_map["image_emb_idx"] == ["image.embedding"]
+
+
+class TestNestedFTSIndexAsync:
+    """Async regression matrix for nested FTS indices."""
+
+    @pytest.mark.asyncio
+    async def test_nested_fts_index_full_path(self, async_db):
+        """FTS index on payload.text must be listed with the full path."""
+        tbl = await async_db.create_table("ft", _nested_fts_data())
+        await tbl.create_index("payload.text", config=FTS(), name="payload_text_idx")
+        col_map = await _columns_by_name_async(tbl)
+        assert col_map["payload_text_idx"] == ["payload.text"]
+
+    @pytest.mark.asyncio
+    async def test_nested_fts_search(self, async_db):
+        """FTS search on a nested text field must return correct results."""
+        tbl = await async_db.create_table("ft", _nested_fts_data())
+        await tbl.create_index("payload.text", config=FTS(), name="payload_text_idx")
+        results = (
+            await tbl.query()
+            .nearest_to_text("alpha", columns="payload.text")
+            .limit(10)
+            .to_list()
+        )
+        assert len(results) > 0
+        assert all(row["payload"]["text"] == "alpha" for row in results)
+
+    @pytest.mark.asyncio
+    async def test_nested_fts_append_optimize(self, tmp_path):
+        """After append and optimize the FTS index listing must be stable."""
+        db = await lancedb.connect_async(
+            tmp_path / "fts_opt_db", read_consistency_interval=timedelta(seconds=0)
+        )
+        tbl = await db.create_table("ft", _nested_fts_data())
+        await tbl.create_index("payload.text", config=FTS(), name="payload_text_idx")
+        await tbl.add(_nested_fts_data(nrows=4))
+        await tbl.optimize()
+        col_map = await _columns_by_name_async(tbl)
+        assert col_map["payload_text_idx"] == ["payload.text"]
@@ -935,14 +935,41 @@ def test_transform_fn(mem_db):
    try:
        import torch

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


+def test_bare_generic_raises_type_error():
+    # A bare, unparameterised List/Tuple has no element type to map to Arrow.
+    # It should raise a clear TypeError, not crash with AttributeError: __args__.
+    for bare in (List, Tuple):
+
+        class TestModel(pydantic.BaseModel):
+            items: bare
+
+        with pytest.raises(TypeError, match="unsupported type"):
+            pydantic_to_schema(TestModel)
+
+
 def test_nested_struct_list_optional():
    class SplitInfo(pydantic.BaseModel):
        start_frame: int
@@ -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()
@@ -154,50 +154,116 @@ async def test_async_checkout():
        assert await table.count_rows() == 300


+def _branch_open_handler(request):
+    if "/branches/list" in request.path:
+        body = json.dumps(
+            {
+                "branches": {
+                    "exp": {
+                        "parentBranch": None,
+                        "parentVersion": 1,
+                        "createAt": 1,
+                        "manifestSize": 1,
+                    }
+                }
+            }
+        ).encode()
+    else:
+        # describe (table open + version/branch validation)
+        body = json.dumps({"version": 2, "schema": {"fields": []}}).encode()
+    request.send_response(200)
+    request.send_header("Content-Type", "application/json")
+    request.end_headers()
+    request.wfile.write(body)
+
+
 def test_remote_open_table_branch_and_version():
+    with mock_lancedb_connection(_branch_open_handler) as db:
+        # version-only (and "main" + version) time-travels the main chain
+        assert db.open_table("test", version=2) is not None
+        assert db.open_table("test", branch="main", version=2).current_branch() is None
+
+        # a non-main branch opens a handle scoped to that branch, with or
+        # without a version
+        assert db.open_table("test", branch="exp").current_branch() == "exp"
+        assert db.open_table("test", branch="exp", version=2).current_branch() == "exp"
+
+
+def test_remote_table_branches_sync():
+    # Branch CRUD + current_branch on the sync RemoteTable. The handle returned
+    # by create/checkout must stay a RemoteTable scoped to the branch.
+    from lancedb.remote.table import RemoteTable
+
    def handler(request):
-        # describe (table open + version validation) always succeeds
+        if "/branches/list" in request.path:
+            body = json.dumps(
+                {
+                    "branches": {
+                        "exp": {
+                            "parentBranch": None,
+                            "parentVersion": 1,
+                            "createAt": 1,
+                            "manifestSize": 1,
+                        }
+                    }
+                }
+            ).encode()
+        elif "/branches/create" in request.path or "/branches/delete" in request.path:
+            body = b"{}"
+        else:
+            # describe (table open + checkout validation)
+            body = json.dumps({"version": 1, "schema": {"fields": []}}).encode()
        request.send_response(200)
        request.send_header("Content-Type", "application/json")
        request.end_headers()
-        request.wfile.write(
-            json.dumps({"version": 2, "schema": {"fields": []}}).encode()
-        )
+        request.wfile.write(body)

    with mock_lancedb_connection(handler) as db:
-        # version-only (and "main" + version) is allowed: remote supports
-        # version time-travel even though it has no branches
-        assert db.open_table("test", version=2) is not None
-        assert db.open_table("test", branch="main", version=2) is not None
+        table = db.open_table("test")
+        assert isinstance(table, RemoteTable)
+        assert table.current_branch() is None

-        # a non-main branch is rejected, with or without a version
-        with pytest.raises(NotImplementedError, match="branching"):
-            db.open_table("test", branch="exp")
-        with pytest.raises(NotImplementedError, match="branching"):
-            db.open_table("test", branch="exp", version=2)
+        branch = table.branches.create("exp")
+        assert isinstance(branch, RemoteTable)
+        assert branch.current_branch() == "exp"
+
+        # list + checkout round trip; checkout also yields a branch-scoped handle
+        assert "exp" in table.branches.list()
+        checked = table.branches.checkout("exp")
+        assert isinstance(checked, RemoteTable)
+        assert checked.current_branch() == "exp"
+
+        table.branches.delete("exp")


@pytest.mark.asyncio
 async def test_async_remote_open_table_branch_and_version():
-    def handler(request):
-        request.send_response(200)
-        request.send_header("Content-Type", "application/json")
-        request.end_headers()
-        request.wfile.write(
-            json.dumps({"version": 2, "schema": {"fields": []}}).encode()
-        )
-
-    async with mock_lancedb_connection_async(handler) as db:
-        # version-only (and "main" + version) is allowed: "main" is the default
-        # branch, so it must not hit the unsupported remote branch path
+    async with mock_lancedb_connection_async(_branch_open_handler) as db:
+        # version-only (and "main" + version) time-travels the main chain
        assert await db.open_table("test", version=2) is not None
-        assert await db.open_table("test", branch="main", version=2) is not None
+        main_v2 = await db.open_table("test", branch="main", version=2)
+        assert main_v2.current_branch() is None

-        # a non-main branch is rejected, with or without a version
-        with pytest.raises(NotImplementedError, match="branching"):
-            await db.open_table("test", branch="exp")
-        with pytest.raises(NotImplementedError, match="branching"):
-            await db.open_table("test", branch="exp", version=2)
+        # a non-main branch opens a handle scoped to that branch
+        exp = await db.open_table("test", branch="exp")
+        assert exp.current_branch() == "exp"
+        exp_v2 = await db.open_table("test", branch="exp", version=2)
+        assert exp_v2.current_branch() == "exp"
+
+
+def test_remote_table_branch_survives_pickle():
+    # Regression: a branch-scoped handle must keep its branch across a
+    # pickle/fork round-trip (it used to reopen on main).
+    with mock_lancedb_connection(_branch_open_handler) as db:
+        branch = db.open_table("test", branch="exp")
+        assert branch.current_branch() == "exp"
+        restored = pickle.loads(pickle.dumps(branch))
+        assert restored.current_branch() == "exp"
+
+        # the pinned version is carried through as well
+        branch_v2 = db.open_table("test", branch="exp", version=2)
+        restored_v2 = pickle.loads(pickle.dumps(branch_v2))
+        assert restored_v2.current_branch() == "exp"


 def test_table_len_sync():
@@ -346,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)
--- a/Show More
+++ b/Show More