refactor(remote): drop x-lancedb-min-version; use timestamp baseline for read-your-write

2026-06-02 20:00:46 +00:00 · 2026-05-28 20:05:10 -07:00
60 changed files with 792 additions and 4694 deletions
--- a/.agents/skills/README.md
+++ b/.agents/skills/README.md
@@ -1,7 +0,0 @@
-# Agent Skills
-
-This directory contains repo-scoped code agent skills for the LanceDB project.
-
-Each skill is a folder that contains a required `SKILL.md` and optional bundled resources.
-
-Codex discovers skills from `.agents/skills` in the current working directory and parent directories.
--- a/.agents/skills/lancedb-update-lance-dependency/SKILL.md
+++ b/.agents/skills/lancedb-update-lance-dependency/SKILL.md
@@ -1,98 +0,0 @@
---
-name: lancedb-update-lance-dependency
-description: Update LanceDB to a specific Lance release or tag. Use when bumping Lance dependencies in the lancedb repository, including Rust workspace Lance crates, Java lance-core, validation, branch creation, commit, push, and PR creation when requested.
---
-
-# LanceDB Update Lance Dependency
-
-## Scope
-
-Use this skill in the `lancedb/lancedb` repository when updating the Lance dependency to a specific Lance version or tag.
-
-Inputs can be a version (`7.2.0-beta.1`), a tag (`v7.2.0-beta.1`), a tag ref (`refs/tags/v7.2.0-beta.1`), or `latest`.
-
-## Workflow
-
-1. Confirm the worktree status with `git status --short`.
-2. Resolve the target Lance version:
-
-   - If the input is `latest`, empty, or omitted, run:
-
-     ```bash
-     python3 ci/check_lance_release.py
-     ```
-
-     Parse the JSON output. If `needs_update` is not `true`, stop without creating a PR. Otherwise use `latest_tag`.
-
-   - If the input is explicit, use it directly.
-
-3. Compute update metadata without changing files:
-
-   ```bash
-   python3 ci/update_lance_dependency.py "$TAG_OR_VERSION" --metadata-only
-   ```
-
-   Before making changes, check for an existing open PR with the emitted `pr_title`:
-
-   ```bash
-   gh pr list --search "\"$PR_TITLE\" in:title" --state open --limit 1 --json number,url,title
-   ```
-
-   If a matching open PR exists, stop and report it instead of creating a duplicate.
-
-4. Run the deterministic update entrypoint:
-
-   ```bash
-   python3 ci/update_lance_dependency.py "$TAG_OR_VERSION"
-   ```
-
-   This updates the Rust workspace Lance dependencies through `ci/set_lance_version.py`, updates `java/pom.xml`, refreshes Cargo metadata, and prints JSON metadata containing `branch_name`, `commit_message`, and `pr_title`.
-
-5. Run validation:
-
-   ```bash
-   cargo clippy --quiet --workspace --tests --all-features -- -D warnings
-   cargo fmt --all --quiet
-   ```
-
-   Fix real diagnostics and rerun clippy until it succeeds. Do not skip warnings.
-
-6. Inspect `git status --short` and `git diff` to ensure only the Lance dependency update and required compatibility fixes are present.
-
-7. If the task only asks to prepare local changes, stop here and report the changed files and validation result.
-
-8. If the task asks to publish the update, create a branch using the printed `branch_name`, stage all relevant files, and commit using the printed `commit_message`. Do not amend or rewrite existing commits.
-
-9. Push to `origin`. Before creating the PR, check that the current token has push permission:
-
-   ```bash
-   gh api repos/lancedb/lancedb --jq .permissions.push
-   ```
-
-   If the remote branch already exists for the same generated branch name, delete the remote ref with `gh api -X DELETE repos/lancedb/lancedb/git/refs/heads/$BRANCH_NAME`, then push. Do not force-push.
-
-10. Create a PR targeting `main` with the printed `pr_title`. If there is no PR template, keep the body to two or three concise sentences: state the Lance dependency bump, note any required compatibility fixes, and link the triggering Lance tag or release.
-
-11. Read back the remote PR title after creation. If it is not a Conventional Commit title, fix it immediately.
-
-12. When running in GitHub Actions after creating the LanceDB PR, trigger the Sophon dependency update:
-
-    ```bash
-    gh workflow run codex-bump-lancedb-lance.yml \
-      --repo lancedb/sophon \
-      -f lance_ref="$LANCE_TAG" \
-      -f lancedb_ref="$BRANCH_NAME"
-    gh run list --repo lancedb/sophon --workflow codex-bump-lancedb-lance.yml --limit 1 --json databaseId,url,displayTitle
-    ```
-
-    Use the emitted metadata `tag` value as `LANCE_TAG`. Do this only after a new LanceDB PR has been created. If the update was skipped because no update is needed or an open PR already exists, do not trigger Sophon.
-
-## GitHub Actions
-
-When this skill is used from GitHub Actions, `TAG`, `GH_TOKEN`, and `GITHUB_TOKEN` may already be set. Resolve `latest` first when `TAG` is empty. Once an explicit tag or version is known, use:
-
-```bash
-python3 ci/update_lance_dependency.py "$TAG" --github-output "$GITHUB_OUTPUT"
-```
-
-Then use the emitted `branch_name`, `commit_message`, and `pr_title` values for branch, commit, and PR creation.
--- a/.bumpversion.toml
+++ b/.bumpversion.toml
@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "0.30.1-beta.0"
+current_version = "0.30.0-beta.1"
 parse = """(?x)
    (?P<major>0|[1-9]\\d*)\\.
    (?P<minor>0|[1-9]\\d*)\\.
--- a/.github/workflows/codex-update-lance-dependency.yml
+++ b/.github/workflows/codex-update-lance-dependency.yml
@@ -4,16 +4,14 @@ on:
  workflow_call:
    inputs:
      tag:
-        description: "Tag name from Lance. If omitted, the skill will use the latest Lance release that needs an update."
-        required: false
-        default: ""
+        description: "Tag name from Lance"
+        required: true
        type: string
  workflow_dispatch:
    inputs:
      tag:
-        description: "Tag name from Lance. Leave empty to use the latest Lance release that needs an update."
-        required: false
-        default: ""
+        description: "Tag name from Lance"
+        required: true
        type: string

 permissions:
@@ -27,7 +25,7 @@ jobs:
    steps:
      - name: Show inputs
        run: |
-          echo "tag = ${{ inputs.tag || 'latest' }}"
+          echo "tag = ${{ inputs.tag }}"

      - name: Checkout Repo LanceDB
        uses: actions/checkout@v4
@@ -73,21 +71,65 @@ jobs:
          OPENAI_API_KEY: ${{ secrets.CODEX_TOKEN }}
        run: |
          set -euo pipefail
-          TARGET_TAG="${TAG:-latest}"
+          VERSION="${TAG#refs/tags/}"
+          VERSION="${VERSION#v}"
+          BRANCH_NAME="codex/update-lance-${VERSION//[^a-zA-Z0-9]/-}"
+
+          # Use "chore" for beta/rc versions, "feat" for stable releases
+          if [[ "${VERSION}" == *beta* ]] || [[ "${VERSION}" == *rc* ]]; then
+            COMMIT_TYPE="chore"
+          else
+            COMMIT_TYPE="feat"
+          fi

          cat <<EOF >/tmp/codex-prompt.txt
-          You are running inside the lancedb repository on a GitHub Actions runner.
+          You are running inside the lancedb repository on a GitHub Actions runner. Update the Lance dependency to version ${VERSION} and prepare a pull request for maintainers to review.

-          Use \$lancedb-update-lance-dependency with target "${TARGET_TAG}".
+          Follow these steps exactly:
+          1. Use script "ci/set_lance_version.py" to update Lance Rust dependencies. The script already refreshes Cargo metadata, so allow it to finish even if it takes time.
+          2. Update the Java lance-core dependency version in "java/pom.xml": change the "<lance-core.version>...</lance-core.version>" property to "${VERSION}".
+          3. Run "cargo clippy --workspace --tests --all-features -- -D warnings". If diagnostics appear, fix them yourself and rerun clippy until it exits cleanly. Do not skip any warnings.
+          4. After clippy succeeds, run "cargo fmt --all" to format the workspace.
+          5. Ensure the repository is clean except for intentional changes. Inspect "git status --short" and "git diff" to confirm the dependency update and any required fixes.
+          6. Create and switch to a new branch named "${BRANCH_NAME}" (replace any duplicated hyphens if necessary).
+          7. Stage all relevant files with "git add -A". Commit using the message "${COMMIT_TYPE}: update lance dependency to v${VERSION}".
+          8. Push the branch to origin. If the remote branch already exists, delete it first with "gh api -X DELETE repos/lancedb/lancedb/git/refs/heads/${BRANCH_NAME}" then push with "git push origin ${BRANCH_NAME}". Do NOT use "git push --force" or "git push -f".
+          9. env "GH_TOKEN" is available, use "gh" tools for github related operations like creating pull request.
+          10. Create a pull request targeting "main" with title "${COMMIT_TYPE}: update lance dependency to v${VERSION}". First, write the PR body to /tmp/pr-body.md using a heredoc (cat <<'EOF' > /tmp/pr-body.md). The body should summarize the dependency bump, clippy/fmt verification, and link the triggering tag (${TAG}). Then run "gh pr create --body-file /tmp/pr-body.md".
+          11. After creating the PR, display the PR URL, "git status --short", and a concise summary of the commands run and their results.

          Constraints:
-          - Use env "GH_TOKEN" for GitHub operations.
-          - Do not merge the pull request.
-          - Do not force-push.
-          - Do not create a duplicate pull request if an open PR already exists for the target Lance version.
-          - If any command fails, diagnose and fix the root cause instead of aborting.
-          - After creating the PR, display the PR URL, "git status --short", and a concise summary of the commands run and their results.
+          - Use bash commands; avoid modifying GitHub workflow files other than through the scripted task above.
+          - Do not merge the PR.
+          - If any command fails, diagnose and fix the issue instead of aborting.
          EOF

          printenv OPENAI_API_KEY | codex login --with-api-key
          codex --config shell_environment_policy.ignore_default_excludes=true exec --dangerously-bypass-approvals-and-sandbox "$(cat /tmp/codex-prompt.txt)"
+
+      - name: Trigger sophon dependency update
+        env:
+          TAG: ${{ inputs.tag }}
+          GH_TOKEN: ${{ secrets.ROBOT_TOKEN }}
+        run: |
+          set -euo pipefail
+          VERSION="${TAG#refs/tags/}"
+          VERSION="${VERSION#v}"
+          LANCEDB_BRANCH="codex/update-lance-${VERSION//[^a-zA-Z0-9]/-}"
+
+          echo "Triggering sophon workflow with:"
+          echo "  lance_ref: ${TAG#refs/tags/}"
+          echo "  lancedb_ref: ${LANCEDB_BRANCH}"
+
+          gh workflow run codex-bump-lancedb-lance.yml \
+            --repo lancedb/sophon \
+            -f lance_ref="${TAG#refs/tags/}" \
+            -f lancedb_ref="${LANCEDB_BRANCH}"
+
+      - name: Show latest sophon workflow run
+        env:
+          GH_TOKEN: ${{ secrets.ROBOT_TOKEN }}
+        run: |
+          set -euo pipefail
+          echo "Latest sophon workflow run:"
+          gh run list --repo lancedb/sophon --workflow codex-bump-lancedb-lance.yml --limit 1 --json databaseId,url,displayTitle
--- a/.github/workflows/lance-release-timer.yml
+++ b/.github/workflows/lance-release-timer.yml
@@ -0,0 +1,62 @@
+name: Lance Release Timer
+
+on:
+  schedule:
+    - cron: "*/10 * * * *"
+  workflow_dispatch:
+
+permissions:
+  contents: read
+  actions: write
+
+concurrency:
+  group: lance-release-timer
+  cancel-in-progress: false
+
+jobs:
+  trigger-update:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Check for new Lance tag
+        id: check
+        env:
+          GH_TOKEN: ${{ secrets.ROBOT_TOKEN }}
+        run: |
+          python3 ci/check_lance_release.py --github-output "$GITHUB_OUTPUT"
+
+      - name: Look for existing PR
+        if: steps.check.outputs.needs_update == 'true'
+        id: pr
+        env:
+          GH_TOKEN: ${{ secrets.ROBOT_TOKEN }}
+        run: |
+          set -euo pipefail
+          TITLE="chore: update lance dependency to v${{ steps.check.outputs.latest_version }}"
+          COUNT=$(gh pr list --search "\"$TITLE\" in:title" --state open --limit 1 --json number --jq 'length')
+          if [ "$COUNT" -gt 0 ]; then
+            echo "Open PR already exists for $TITLE"
+            echo "pr_exists=true" >> "$GITHUB_OUTPUT"
+          else
+            echo "No existing PR for $TITLE"
+            echo "pr_exists=false" >> "$GITHUB_OUTPUT"
+          fi
+
+      - name: Trigger codex update workflow
+        if: steps.check.outputs.needs_update == 'true' && steps.pr.outputs.pr_exists != 'true'
+        env:
+          GH_TOKEN: ${{ secrets.ROBOT_TOKEN }}
+        run: |
+          set -euo pipefail
+          TAG=${{ steps.check.outputs.latest_tag }}
+          gh workflow run codex-update-lance-dependency.yml -f tag=refs/tags/$TAG
+
+      - name: Show latest codex workflow run
+        if: steps.check.outputs.needs_update == 'true' && steps.pr.outputs.pr_exists != 'true'
+        env:
+          GH_TOKEN: ${{ secrets.ROBOT_TOKEN }}
+        run: |
+          set -euo pipefail
+          gh run list --workflow codex-update-lance-dependency.yml --limit 1 --json databaseId,url,displayTitle
--- a/Cargo.lock
+++ b/Cargo.lock
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -13,20 +13,20 @@ categories = ["database-implementations"]
 rust-version = "1.91.0"

 [workspace.dependencies]
-lance = { "version" = "=7.2.0-beta.3", default-features = false, "tag" = "v7.2.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
-lance-core = { "version" = "=7.2.0-beta.3", "tag" = "v7.2.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
-lance-datagen = { "version" = "=7.2.0-beta.3", "tag" = "v7.2.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
-lance-file = { "version" = "=7.2.0-beta.3", "tag" = "v7.2.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
-lance-io = { "version" = "=7.2.0-beta.3", default-features = false, "tag" = "v7.2.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
-lance-index = { "version" = "=7.2.0-beta.3", "tag" = "v7.2.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
-lance-linalg = { "version" = "=7.2.0-beta.3", "tag" = "v7.2.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace = { "version" = "=7.2.0-beta.3", "tag" = "v7.2.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace-impls = { "version" = "=7.2.0-beta.3", default-features = false, "tag" = "v7.2.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
-lance-table = { "version" = "=7.2.0-beta.3", "tag" = "v7.2.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
-lance-testing = { "version" = "=7.2.0-beta.3", "tag" = "v7.2.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
-lance-datafusion = { "version" = "=7.2.0-beta.3", "tag" = "v7.2.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
-lance-encoding = { "version" = "=7.2.0-beta.3", "tag" = "v7.2.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
-lance-arrow = { "version" = "=7.2.0-beta.3", "tag" = "v7.2.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
+lance = { "version" = "=7.1.0-beta.4", default-features = false, "tag" = "v7.1.0-beta.4", "git" = "https://github.com/lance-format/lance.git" }
+lance-core = { "version" = "=7.1.0-beta.4", "tag" = "v7.1.0-beta.4", "git" = "https://github.com/lance-format/lance.git" }
+lance-datagen = { "version" = "=7.1.0-beta.4", "tag" = "v7.1.0-beta.4", "git" = "https://github.com/lance-format/lance.git" }
+lance-file = { "version" = "=7.1.0-beta.4", "tag" = "v7.1.0-beta.4", "git" = "https://github.com/lance-format/lance.git" }
+lance-io = { "version" = "=7.1.0-beta.4", default-features = false, "tag" = "v7.1.0-beta.4", "git" = "https://github.com/lance-format/lance.git" }
+lance-index = { "version" = "=7.1.0-beta.4", "tag" = "v7.1.0-beta.4", "git" = "https://github.com/lance-format/lance.git" }
+lance-linalg = { "version" = "=7.1.0-beta.4", "tag" = "v7.1.0-beta.4", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace = { "version" = "=7.1.0-beta.4", "tag" = "v7.1.0-beta.4", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace-impls = { "version" = "=7.1.0-beta.4", default-features = false, "tag" = "v7.1.0-beta.4", "git" = "https://github.com/lance-format/lance.git" }
+lance-table = { "version" = "=7.1.0-beta.4", "tag" = "v7.1.0-beta.4", "git" = "https://github.com/lance-format/lance.git" }
+lance-testing = { "version" = "=7.1.0-beta.4", "tag" = "v7.1.0-beta.4", "git" = "https://github.com/lance-format/lance.git" }
+lance-datafusion = { "version" = "=7.1.0-beta.4", "tag" = "v7.1.0-beta.4", "git" = "https://github.com/lance-format/lance.git" }
+lance-encoding = { "version" = "=7.1.0-beta.4", "tag" = "v7.1.0-beta.4", "git" = "https://github.com/lance-format/lance.git" }
+lance-arrow = { "version" = "=7.1.0-beta.4", "tag" = "v7.1.0-beta.4", "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 }
--- a/ci/update_lance_dependency.py
+++ b/ci/update_lance_dependency.py
@@ -1,126 +0,0 @@
-#!/usr/bin/env python3
-"""Prepare a Lance dependency update for LanceDB."""
-
-from __future__ import annotations
-
-import argparse
-import json
-import re
-import subprocess
-import sys
-from pathlib import Path
-from typing import Sequence
-
-try:
-    from check_lance_release import parse_semver
-except ModuleNotFoundError:
-    # Supports importing as ci.update_lance_dependency from tests or ad hoc checks.
-    from ci.check_lance_release import parse_semver  # type: ignore
-
-
-def normalize_version(raw: str) -> str:
-    value = raw.strip()
-    value = value.removeprefix("refs/tags/")
-    value = value.removeprefix("v")
-    try:
-        parse_semver(value)
-    except ValueError:
-        raise ValueError(f"Unsupported Lance version or tag: {raw}")
-    return value
-
-
-def normalized_tag(version: str) -> str:
-    return f"v{version}"
-
-
-def branch_name(version: str) -> str:
-    suffix = re.sub(r"[^a-zA-Z0-9]+", "-", version).strip("-")
-    suffix = re.sub(r"-+", "-", suffix)
-    return f"codex/update-lance-{suffix}"
-
-
-def commit_type(version: str) -> str:
-    prerelease = version.split("-", maxsplit=1)[1] if "-" in version else ""
-    return "chore" if "beta" in prerelease or "rc" in prerelease else "feat"
-
-
-def metadata_for(version: str) -> dict[str, str]:
-    kind = commit_type(version)
-    message = f"{kind}: update lance dependency to v{version}"
-    return {
-        "version": version,
-        "tag": normalized_tag(version),
-        "branch_name": branch_name(version),
-        "commit_type": kind,
-        "commit_message": message,
-        "pr_title": message,
-    }
-
-
-def run_command(cmd: Sequence[str], *, cwd: Path) -> None:
-    subprocess.run(cmd, cwd=cwd, check=True)
-
-
-def update_java_lance_core_version(repo_root: Path, version: str) -> None:
-    pom_path = repo_root / "java" / "pom.xml"
-    contents = pom_path.read_text(encoding="utf-8")
-    updated, count = re.subn(
-        r"(<lance-core\.version>)[^<]+(</lance-core\.version>)",
-        rf"\g<1>{version}\g<2>",
-        contents,
-        count=1,
-    )
-    if count != 1:
-        raise RuntimeError(
-            "Expected exactly one <lance-core.version> entry in java/pom.xml"
-        )
-    pom_path.write_text(updated, encoding="utf-8")
-
-
-def write_github_outputs(path: str | None, payload: dict[str, str]) -> None:
-    if not path:
-        return
-    with open(path, "a", encoding="utf-8") as output:
-        for key, value in payload.items():
-            output.write(f"{key}={value}\n")
-
-
-def main(argv: Sequence[str] | None = None) -> int:
-    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument(
-        "tag_or_version",
-        help="Lance tag or version, for example refs/tags/v7.2.0-beta.1 or 7.2.0",
-    )
-    parser.add_argument(
-        "--repo-root",
-        type=Path,
-        default=Path(__file__).resolve().parents[1],
-        help="Path to the lancedb repository root",
-    )
-    parser.add_argument(
-        "--github-output",
-        default=None,
-        help="Optional GitHub Actions output file to receive metadata fields",
-    )
-    parser.add_argument(
-        "--metadata-only",
-        action="store_true",
-        help="Only print derived metadata; do not modify dependency files",
-    )
-    args = parser.parse_args(argv)
-
-    repo_root = args.repo_root.resolve()
-    version = normalize_version(args.tag_or_version)
-    payload = metadata_for(version)
-
-    if not args.metadata_only:
-        run_command([sys.executable, "ci/set_lance_version.py", version], cwd=repo_root)
-        update_java_lance_core_version(repo_root, version)
-
-    write_github_outputs(args.github_output, payload)
-    print(json.dumps(payload, sort_keys=True))
-    return 0
-
-
-if __name__ == "__main__":
-    sys.exit(main())
--- a/docs/src/java/java.md
+++ b/docs/src/java/java.md
@@ -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.0</version>
+    <version>0.30.0-beta.1</version>
 </dependency>
 ```

--- a/docs/src/js/classes/MergeInsertBuilder.md
+++ b/docs/src/js/classes/MergeInsertBuilder.md
@@ -76,57 +76,6 @@ the query optimizer chooses a suboptimal path.

 ***

-### useLsmWrite()
-
-```ts
-useLsmWrite(useLsmWrite): MergeInsertBuilder
-```
-
-Controls whether the merge uses the MemWAL LSM write path.
-
-By default (unset), a `mergeInsert` on a table with an LSM write spec is
-routed through Lance's MemWAL shard writer, and a table without one uses
-the standard path. Pass `false` to force the standard path even when a
-spec is set. Pass `true` to require a spec — `mergeInsert` rejects if none
-is installed.
-
-#### Parameters
-
-* **useLsmWrite**: `boolean`
-    Whether to use the LSM write path.
-
-#### Returns
-
-[`MergeInsertBuilder`](MergeInsertBuilder.md)
-
-***
-
-### validateSingleShard()
-
-```ts
-validateSingleShard(validateSingleShard): MergeInsertBuilder
-```
-
-Controls how an LSM merge checks that its input targets a single shard.
-
-When a table has an LSM write spec, every row in a `mergeInsert` call must
-route to the same shard. When `true` (the default), every row is inspected
-to verify this. When `false`, only the first row is inspected and the
-shard it routes to is used for the whole input — a faster path for callers
-that have already pre-sharded their input. Has no effect on tables without
-an LSM write spec.
-
-#### Parameters
-
-* **validateSingleShard**: `boolean`
-    Whether to check every row routes to one shard. Defaults to `true`.
-
-#### Returns
-
-[`MergeInsertBuilder`](MergeInsertBuilder.md)
-
-***
-
 ### whenMatchedUpdateAll()

 ```ts
--- a/docs/src/js/classes/Table.md
+++ b/docs/src/js/classes/Table.md
@@ -187,25 +187,6 @@ Any attempt to use the table after it is closed will result in an error.

 ***

-### closeLsmWriters()
-
-```ts
-abstract closeLsmWriters(): Promise<void>
-```
-
-Drain and close any cached MemWAL shard writers held for this table.
-
-When an [LsmWriteSpec](../interfaces/LsmWriteSpec.md) is installed, `mergeInsert` opens MemWAL
-shard writers and caches them for reuse across calls. This closes them,
-flushing pending data; writers reopen lazily on the next `mergeInsert`.
-It is a no-op when no writers are cached.
-
-#### Returns
-
-`Promise`&lt;`void`&gt;
-
-***
-
 ### countRows()

 ```ts
--- a/docs/src/js/interfaces/LsmWriteSpec.md
+++ b/docs/src/js/interfaces/LsmWriteSpec.md
@@ -11,10 +11,7 @@ Specification selecting Lance's MemWAL LSM-style write path for

 `specType` is `"bucket"`, `"identity"`, or `"unsharded"`. For `"bucket"`,
 `column` and `numBuckets` are required; for `"identity"`, `column` is
-required and must be a deterministic function of the unenforced primary
-key (every row with a given primary key must always produce the same
-`column` value, or upserts of that key can land in different shards and a
-stale version can win).
+required.

 ## Properties

--- a/docs/src/js/interfaces/MergeResult.md
+++ b/docs/src/js/interfaces/MergeResult.md
@@ -32,14 +32,6 @@ numInsertedRows: number;

 ***

-### numRows
-
-```ts
-numRows: number;
-```
-
-***
-
 ### numUpdatedRows

 ```ts
--- a/java/lancedb-core/pom.xml
+++ b/java/lancedb-core/pom.xml
@@ -8,7 +8,7 @@
    <parent>
      <groupId>com.lancedb</groupId>
      <artifactId>lancedb-parent</artifactId>
-      <version>0.30.1-beta.0</version>
+      <version>0.30.0-beta.1</version>
      <relativePath>../pom.xml</relativePath>
    </parent>

--- a/java/pom.xml
+++ b/java/pom.xml
@@ -6,7 +6,7 @@

    <groupId>com.lancedb</groupId>
    <artifactId>lancedb-parent</artifactId>
-    <version>0.30.1-beta.0</version>
+    <version>0.30.0-beta.1</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>7.2.0-beta.1</lance-core.version>
+        <lance-core.version>7.1.0-beta.4</lance-core.version>
        <spotless.skip>false</spotless.skip>
        <spotless.version>2.30.0</spotless.version>
        <spotless.java.googlejavaformat.version>1.7</spotless.java.googlejavaformat.version>
--- a/nodejs/Cargo.toml
+++ b/nodejs/Cargo.toml
@@ -1,7 +1,7 @@
 [package]
 name = "lancedb-nodejs"
 edition.workspace = true
-version = "0.30.1-beta.0"
+version = "0.30.0-beta.1"
 publish = false
 license.workspace = true
 description.workspace = true
--- a/nodejs/test/table.test.ts
+++ b/nodejs/test/table.test.ts
@@ -2625,97 +2625,3 @@ describe("setLsmWriteSpec / unsetLsmWriteSpec", () => {
    ).rejects.toThrow();
  });
 });
-
-describe("LSM merge insert", () => {
-  let tmpDir: tmp.DirResult;
-
-  beforeEach(() => {
-    tmpDir = tmp.dirSync({ unsafeCleanup: true });
-  });
-  afterEach(() => tmpDir.removeCallback());
-
-  async function bucketTable(conn: Connection): Promise<Table> {
-    // The primary key column must be non-nullable.
-    const table = await conn.createEmptyTable(
-      "t",
-      new arrow.Schema([
-        new arrow.Field("id", new arrow.Utf8(), false),
-        new arrow.Field("value", new arrow.Float64(), true),
-      ]),
-    );
-    await table.add([
-      { id: "a", value: 1 },
-      { id: "b", value: 2 },
-    ]);
-    await table.setUnenforcedPrimaryKey("id");
-    // numBuckets = 1: every row routes to the single bucket.
-    await table.setLsmWriteSpec({
-      specType: "bucket",
-      column: "id",
-      numBuckets: 1,
-    });
-    return table;
-  }
-
-  it("routes merge_insert through the shard writer", async () => {
-    const conn = await connect(tmpDir.name);
-    const table = await bucketTable(conn);
-
-    const res = await table
-      .mergeInsert("id")
-      .whenMatchedUpdateAll()
-      .whenNotMatchedInsertAll()
-      .execute([
-        { id: "c", value: 3 },
-        { id: "d", value: 4 },
-      ]);
-    // LSM path: rows go to the MemWAL, so only numRows is populated.
-    expect(res.numRows).toBe(2);
-    expect(res.version).toBe(0);
-    expect(res.numInsertedRows).toBe(0);
-
-    await table.closeLsmWriters();
-  });
-
-  it("falls back to the standard path with useLsmWrite(false)", async () => {
-    const conn = await connect(tmpDir.name);
-    const table = await bucketTable(conn);
-
-    const res = await table
-      .mergeInsert("id")
-      .whenNotMatchedInsertAll()
-      .useLsmWrite(false)
-      .execute([
-        { id: "b", value: 9 },
-        { id: "e", value: 5 },
-      ]);
-    // Standard path commits: id="e" inserted ("b" already exists).
-    expect(res.numInsertedRows).toBe(1);
-    expect(await table.countRows()).toBe(3);
-  });
-
-  it("supports validateSingleShard(false)", async () => {
-    const conn = await connect(tmpDir.name);
-    const table = await bucketTable(conn);
-
-    const res = await table
-      .mergeInsert("id")
-      .whenMatchedUpdateAll()
-      .whenNotMatchedInsertAll()
-      .validateSingleShard(false)
-      .execute([{ id: "f", value: 6 }]);
-    expect(res.numRows).toBe(1);
-  });
-
-  it("rejects a non-upsert merge under an LSM spec", async () => {
-    const conn = await connect(tmpDir.name);
-    const table = await bucketTable(conn);
-
-    await expect(
-      table
-        .mergeInsert("id")
-        .whenNotMatchedInsertAll()
-        .execute([{ id: "g", value: 7 }]),
-    ).rejects.toThrow();
-  });
-});
--- a/nodejs/lancedb/merge.ts
+++ b/nodejs/lancedb/merge.ts
@@ -87,41 +87,6 @@ export class MergeInsertBuilder {
      this.#schema,
    );
  }
-  /**
-   * Controls whether the merge uses the MemWAL LSM write path.
-   *
-   * By default (unset), a `mergeInsert` on a table with an LSM write spec is
-   * routed through Lance's MemWAL shard writer, and a table without one uses
-   * the standard path. Pass `false` to force the standard path even when a
-   * spec is set. Pass `true` to require a spec — `mergeInsert` rejects if none
-   * is installed.
-   *
-   * @param useLsmWrite - Whether to use the LSM write path.
-   */
-  useLsmWrite(useLsmWrite: boolean): MergeInsertBuilder {
-    return new MergeInsertBuilder(
-      this.#native.useLsmWrite(useLsmWrite),
-      this.#schema,
-    );
-  }
-  /**
-   * Controls how an LSM merge checks that its input targets a single shard.
-   *
-   * When a table has an LSM write spec, every row in a `mergeInsert` call must
-   * route to the same shard. When `true` (the default), every row is inspected
-   * to verify this. When `false`, only the first row is inspected and the
-   * shard it routes to is used for the whole input — a faster path for callers
-   * that have already pre-sharded their input. Has no effect on tables without
-   * an LSM write spec.
-   *
-   * @param validateSingleShard - Whether to check every row routes to one shard. Defaults to `true`.
-   */
-  validateSingleShard(validateSingleShard: boolean): MergeInsertBuilder {
-    return new MergeInsertBuilder(
-      this.#native.validateSingleShard(validateSingleShard),
-      this.#schema,
-    );
-  }
  /**
   * Executes the merge insert operation
   *
--- a/nodejs/lancedb/table.ts
+++ b/nodejs/lancedb/table.ts
@@ -161,10 +161,7 @@ export interface Version {
 *
 * `specType` is `"bucket"`, `"identity"`, or `"unsharded"`. For `"bucket"`,
 * `column` and `numBuckets` are required; for `"identity"`, `column` is
- * required and must be a deterministic function of the unenforced primary
- * key (every row with a given primary key must always produce the same
- * `column` value, or upserts of that key can land in different shards and a
- * stale version can win).
+ * required.
 */
 export interface LsmWriteSpec {
  /** One of `"bucket"`, `"identity"`, or `"unsharded"`. */
@@ -570,16 +567,6 @@ export abstract class Table {
   * @returns {Promise<void>}
   */
  abstract unsetLsmWriteSpec(): Promise<void>;
-  /**
-   * Drain and close any cached MemWAL shard writers held for this table.
-   *
-   * When an {@link LsmWriteSpec} is installed, `mergeInsert` opens MemWAL
-   * shard writers and caches them for reuse across calls. This closes them,
-   * flushing pending data; writers reopen lazily on the next `mergeInsert`.
-   * It is a no-op when no writers are cached.
-   * @returns {Promise<void>}
-   */
-  abstract closeLsmWriters(): Promise<void>;
  /** Retrieve the version of the table */

  abstract version(): Promise<number>;
@@ -1054,10 +1041,6 @@ export class LocalTable extends Table {
    return await this.inner.unsetLsmWriteSpec();
  }

-  async closeLsmWriters(): Promise<void> {
-    return await this.inner.closeLsmWriters();
-  }
-
  async version(): Promise<number> {
    return await this.inner.version();
  }
--- a/nodejs/npm/darwin-arm64/package.json
+++ b/nodejs/npm/darwin-arm64/package.json
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-darwin-arm64",
-	"version": "0.30.1-beta.0",
+	"version": "0.30.0-beta.1",
 	"os": ["darwin"],
 	"cpu": ["arm64"],
 	"main": "lancedb.darwin-arm64.node",
--- a/nodejs/npm/linux-arm64-gnu/package.json
+++ b/nodejs/npm/linux-arm64-gnu/package.json
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-gnu",
-	"version": "0.30.1-beta.0",
+	"version": "0.30.0-beta.1",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-gnu.node",
--- a/nodejs/npm/linux-arm64-musl/package.json
+++ b/nodejs/npm/linux-arm64-musl/package.json
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-musl",
-	"version": "0.30.1-beta.0",
+	"version": "0.30.0-beta.1",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-musl.node",
--- a/nodejs/npm/linux-x64-gnu/package.json
+++ b/nodejs/npm/linux-x64-gnu/package.json
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-gnu",
-	"version": "0.30.1-beta.0",
+	"version": "0.30.0-beta.1",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-gnu.node",
--- a/nodejs/npm/linux-x64-musl/package.json
+++ b/nodejs/npm/linux-x64-musl/package.json
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-musl",
-	"version": "0.30.1-beta.0",
+	"version": "0.30.0-beta.1",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-musl.node",
--- a/nodejs/npm/win32-arm64-msvc/package.json
+++ b/nodejs/npm/win32-arm64-msvc/package.json
@@ -1,6 +1,6 @@
 {
  "name": "@lancedb/lancedb-win32-arm64-msvc",
-  "version": "0.30.1-beta.0",
+  "version": "0.30.0-beta.1",
  "os": [
    "win32"
  ],
--- a/nodejs/npm/win32-x64-msvc/package.json
+++ b/nodejs/npm/win32-x64-msvc/package.json
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-win32-x64-msvc",
-	"version": "0.30.1-beta.0",
+	"version": "0.30.0-beta.1",
 	"os": ["win32"],
 	"cpu": ["x64"],
 	"main": "lancedb.win32-x64-msvc.node",
--- a/nodejs/package-lock.json
+++ b/nodejs/package-lock.json
@@ -1,12 +1,12 @@
 {
  "name": "@lancedb/lancedb",
-  "version": "0.30.1-beta.0",
+  "version": "0.30.0-beta.1",
  "lockfileVersion": 3,
  "requires": true,
  "packages": {
    "": {
      "name": "@lancedb/lancedb",
-      "version": "0.30.1-beta.0",
+      "version": "0.30.0-beta.1",
      "cpu": [
        "x64",
        "arm64"
--- a/nodejs/package.json
+++ b/nodejs/package.json
@@ -11,7 +11,7 @@
    "ann"
  ],
  "private": false,
-  "version": "0.30.1-beta.0",
+  "version": "0.30.0-beta.1",
  "main": "dist/index.js",
  "exports": {
    ".": "./dist/index.js",
--- a/nodejs/src/merge.rs
+++ b/nodejs/src/merge.rs
@@ -50,20 +50,6 @@ impl NativeMergeInsertBuilder {
        this
    }

-    #[napi]
-    pub fn use_lsm_write(&self, use_lsm_write: bool) -> Self {
-        let mut this = self.clone();
-        this.inner.use_lsm_write(use_lsm_write);
-        this
-    }
-
-    #[napi]
-    pub fn validate_single_shard(&self, validate_single_shard: bool) -> Self {
-        let mut this = self.clone();
-        this.inner.validate_single_shard(validate_single_shard);
-        this
-    }
-
    #[napi(catch_unwind)]
    pub async fn execute(&self, buf: Buffer) -> napi::Result<MergeResult> {
        let data = ipc_file_to_batches(buf.to_vec())
--- a/nodejs/src/table.rs
+++ b/nodejs/src/table.rs
@@ -391,11 +391,6 @@ impl Table {
            .default_error()
    }

-    #[napi(catch_unwind)]
-    pub async fn close_lsm_writers(&self) -> napi::Result<()> {
-        self.inner_ref()?.close_lsm_writers().await.default_error()
-    }
-
    #[napi(catch_unwind)]
    pub async fn version(&self) -> napi::Result<i64> {
        self.inner_ref()?
@@ -945,7 +940,6 @@ pub struct MergeResult {
    pub num_updated_rows: i64,
    pub num_deleted_rows: i64,
    pub num_attempts: i64,
-    pub num_rows: i64,
 }

 impl From<lancedb::table::MergeResult> for MergeResult {
@@ -956,7 +950,6 @@ impl From<lancedb::table::MergeResult> for MergeResult {
            num_updated_rows: value.num_updated_rows as i64,
            num_deleted_rows: value.num_deleted_rows as i64,
            num_attempts: value.num_attempts as i64,
-            num_rows: value.num_rows as i64,
        }
    }
 }
--- a/python/.bumpversion.toml
+++ b/python/.bumpversion.toml
@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "0.33.1-beta.0"
+current_version = "0.33.0-beta.1"
 parse = """(?x)
    (?P<major>0|[1-9]\\d*)\\.
    (?P<minor>0|[1-9]\\d*)\\.
--- a/python/Cargo.toml
+++ b/python/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "lancedb-python"
-version = "0.33.1-beta.0"
+version = "0.33.0-beta.1"
 publish = false
 edition.workspace = true
 description = "Python bindings for LanceDB"
--- a/python/python/lancedb/init.py
+++ b/python/python/lancedb/init.py
@@ -315,15 +315,6 @@ def deserialize_conn(
            manifest_enabled=parsed.get("manifest_enabled", False),
            namespace_client_properties=parsed.get("namespace_client_properties"),
        )
-    elif connection_type == "remote":
-        return RemoteDBConnection(
-            parsed["db_url"],
-            parsed["api_key"],
-            parsed.get("region", "us-east-1"),
-            host_override=parsed.get("host_override"),
-            client_config=parsed.get("client_config"),
-            storage_options=storage_options,
-        )
    else:
        raise ValueError(f"Unknown connection_type: {connection_type}")

--- a/python/python/lancedb/_lancedb.pyi
+++ b/python/python/lancedb/_lancedb.pyi
@@ -220,7 +220,6 @@ 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 close_lsm_writers(self) -> None: ...
    @property
    def tags(self) -> Tags: ...
    def query(self) -> Query: ...
@@ -421,7 +420,6 @@ class MergeResult:
    num_inserted_rows: int
    num_deleted_rows: int
    num_attempts: int
-    num_rows: int

 class LsmWriteSpec:
    """Specification selecting Lance's MemWAL LSM-style write path for
--- a/python/python/lancedb/index.py
+++ b/python/python/lancedb/index.py
@@ -281,9 +281,6 @@ class HnswPq:
    m: int = 20
    ef_construction: int = 300
    target_partition_size: Optional[int] = None
-    # Name of the accelerator (e.g. "cuda") to use for IVF training. When set,
-    # create_index() dispatches to pylance to build the index on the accelerator.
-    accelerator: Optional[str] = None


@dataclass
@@ -389,9 +386,6 @@ class HnswSq:
    m: int = 20
    ef_construction: int = 300
    target_partition_size: Optional[int] = None
-    # Name of the accelerator (e.g. "cuda") to use for IVF training. When set,
-    # create_index() dispatches to pylance to build the index on the accelerator.
-    accelerator: Optional[str] = None


@dataclass
@@ -585,9 +579,6 @@ class IvfFlat:
    max_iterations: int = 50
    sample_rate: int = 256
    target_partition_size: Optional[int] = None
-    # Name of the accelerator (e.g. "cuda") to use for IVF training. When set,
-    # create_index() dispatches to pylance to build the index on the accelerator.
-    accelerator: Optional[str] = None


@dataclass
@@ -618,9 +609,6 @@ class IvfSq:
    max_iterations: int = 50
    sample_rate: int = 256
    target_partition_size: Optional[int] = None
-    # Name of the accelerator (e.g. "cuda") to use for IVF training. When set,
-    # create_index() dispatches to pylance to build the index on the accelerator.
-    accelerator: Optional[str] = None


@dataclass
@@ -751,9 +739,6 @@ class IvfPq:
    max_iterations: int = 50
    sample_rate: int = 256
    target_partition_size: Optional[int] = None
-    # Name of the accelerator (e.g. "cuda") to use for IVF training. When set,
-    # create_index() dispatches to pylance to build the index on the accelerator.
-    accelerator: Optional[str] = None


@dataclass
@@ -807,9 +792,6 @@ class IvfRq:
    max_iterations: int = 50
    sample_rate: int = 256
    target_partition_size: Optional[int] = None
-    # Name of the accelerator (e.g. "cuda") to use for IVF training. When set,
-    # create_index() dispatches to pylance to build the index on the accelerator.
-    accelerator: Optional[str] = None


 __all__ = [
--- a/python/python/lancedb/merge.py
+++ b/python/python/lancedb/merge.py
@@ -34,8 +34,6 @@ class LanceMergeInsertBuilder(object):
        self._when_not_matched_by_source_condition = None
        self._timeout = None
        self._use_index = True
-        self._use_lsm_write = None
-        self._validate_single_shard = None

    def when_matched_update_all(
        self, *, where: Optional[str] = None
@@ -98,46 +96,6 @@ class LanceMergeInsertBuilder(object):
        self._use_index = use_index
        return self

-    def use_lsm_write(self, use_lsm_write: bool) -> LanceMergeInsertBuilder:
-        """
-        Controls whether the merge uses the MemWAL LSM write path.
-
-        By default (unset), a `merge_insert` on a table with an LSM write spec
-        is routed through Lance's MemWAL shard writer, and a table without one
-        uses the standard path. Pass `False` to force the standard path even
-        when a spec is set. Pass `True` to require a spec — `merge_insert`
-        raises an error if none is installed.
-
-        Parameters
-        ----------
-        use_lsm_write: bool
-            Whether to use the LSM write path.
-        """
-        self._use_lsm_write = use_lsm_write
-        return self
-
-    def validate_single_shard(
-        self, validate_single_shard: bool
-    ) -> LanceMergeInsertBuilder:
-        """
-        Controls how an LSM merge checks that its input targets a single shard.
-
-        When a table has an LSM write spec, every row in a `merge_insert` call
-        must route to the same shard. When `True` (the default), every row is
-        inspected to verify this. When `False`, only the first row is inspected
-        and the shard it routes to is used for the whole input — a faster path
-        for callers that have already pre-sharded their input.
-
-        Has no effect on tables without an LSM write spec.
-
-        Parameters
-        ----------
-        validate_single_shard: bool
-            Whether to check every row routes to one shard. Defaults to `True`.
-        """
-        self._validate_single_shard = validate_single_shard
-        return self
-
    def execute(
        self,
        new_data: DATA,
--- a/python/python/lancedb/permutation.py
+++ b/python/python/lancedb/permutation.py
@@ -3,13 +3,12 @@

 import copy
 import json
-import os

 from deprecation import deprecated
 import pyarrow as pa

 from ._lancedb import async_permutation_builder, PermutationReader
-from .table import LanceTable, Table
+from .table import LanceTable
 from .background_loop import LOOP
 from .util import batch_to_tensor, batch_to_tensor_rows
 from typing import Any, Callable, Iterator, Literal, Optional, TYPE_CHECKING, Union
@@ -355,49 +354,6 @@ class Transforms:
 DEFAULT_BATCH_SIZE = 100


-def _table_to_pickle_state(table: Table) -> dict[str, Any]:
-    from .remote.table import RemoteTable
-
-    if isinstance(table, RemoteTable):
-        return {
-            "kind": "remote",
-            "table": table,
-        }
-
-    if not isinstance(table, LanceTable):
-        raise ValueError(f"Cannot pickle table of type {type(table)!r}")
-
-    base_uri = table._conn.uri
-    if base_uri.startswith("memory://"):
-        return {
-            "kind": "memory",
-            "name": table.name,
-            "data": table.to_arrow(),
-        }
-
-    return {
-        "kind": "local",
-        "name": table.name,
-        "uri": base_uri,
-        "namespace": table._namespace_path,
-        "storage_options": table._conn.storage_options,
-    }
-
-
-def _table_from_pickle_state(state: dict[str, Any]) -> Table:
-    from . import connect
-
-    kind = state["kind"]
-    if kind == "remote":
-        return state["table"]
-    if kind == "memory":
-        return connect("memory://").create_table(state["name"], state["data"])
-    if kind == "local":
-        db = connect(state["uri"], storage_options=state["storage_options"])
-        return db.open_table(state["name"], namespace_path=state["namespace"] or None)
-    raise ValueError(f"Unknown table pickle state kind: {kind}")
-
-
 class Permutation:
    """
    A Permutation is a view of a dataset that can be used as input to model training
@@ -413,15 +369,15 @@ class Permutation:

    def __init__(
        self,
-        base_table: Table,
-        permutation_table: Optional[Table],
+        base_table: LanceTable,
+        permutation_table: Optional[LanceTable],
        split: int,
        selection: dict[str, str],
        batch_size: int,
        transform_fn: Callable[pa.RecordBatch, Any],
        offset: Optional[int] = None,
        limit: Optional[int] = None,
-        connection_factory: Optional[Callable[[str], Table]] = None,
+        connection_factory: Optional[Callable[[str], LanceTable]] = None,
        _reader: Optional[PermutationReader] = None,
    ):
        """
@@ -441,7 +397,6 @@ class Permutation:
        if _reader is None:
            _reader = LOOP.run(self._build_reader())
        self.reader: PermutationReader = _reader
-        self._pid = os.getpid()

    async def _build_reader(self) -> PermutationReader:
        reader = await PermutationReader.from_tables(
@@ -473,25 +428,29 @@ class Permutation:
        return new

    def with_connection_factory(
-        self, connection_factory: Callable[[str], Table]
+        self, connection_factory: Callable[[str], LanceTable]
    ) -> "Permutation":
        """
        Creates a new permutation that will use ``connection_factory`` to reopen
        the base table when this permutation is unpickled in a worker process.

-        The factory is a callable that takes a single argument, the base table
-        name, and returns a LanceDB table. It must be picklable; the worker
+        The factory is a callable that takes a single argument — the base table
+        name — and returns a [LanceTable]. It must be picklable; the worker
        will pickle it via standard ``pickle`` and call it to recover the base
        table. Picklable callables in practice means top-level (module-level)
        functions, ``functools.partial`` of such functions, or instances of
        picklable classes implementing ``__call__``. Lambdas and closures over
        local variables don't pickle with the default protocol.

-        A factory is optional for normal local and remote LanceDB connections:
-        if not set, ``__getstate__`` captures the table's own picklable reopen
-        state. Use a factory when that default state is not enough, for example
-        when credentials should be loaded from the worker environment instead
-        of being embedded in the pickle.
+        Setting a factory is necessary when the URI alone is not enough to
+        re-open the connection — most importantly for LanceDB Cloud (``db://``)
+        connections, where ``api_key`` and ``region`` aren't recoverable from
+        the connection object after construction.
+
+        For local file or cloud-storage paths the factory is optional: if not
+        set, ``__getstate__`` falls back to capturing
+        ``(uri, storage_options, namespace_path)`` and re-opening via
+        ``lancedb.connect(uri, storage_options=...)``.

        Examples
        --------
@@ -549,7 +508,7 @@ class Permutation:
        return new

    @classmethod
-    def identity(cls, table: Table) -> "Permutation":
+    def identity(cls, table: LanceTable) -> "Permutation":
        """
        Creates an identity permutation for the given table.
        """
@@ -558,8 +517,8 @@ class Permutation:
    @classmethod
    def from_tables(
        cls,
-        base_table: Table,
-        permutation_table: Optional[Table] = None,
+        base_table: LanceTable,
+        permutation_table: Optional[LanceTable] = None,
        split: Optional[Union[str, int]] = None,
    ) -> "Permutation":
        """
@@ -635,10 +594,11 @@ class Permutation:

        The base table is captured either via a user-supplied
        ``connection_factory`` (see [with_connection_factory]) or, as a
-        fallback, by the table's own picklable reopen state. The permutation
-        table is captured as a pyarrow Table (which pickles via Arrow IPC
-        natively). The reader is dropped from the wire format and rebuilt
-        lazily on first use.
+        fallback, by introspecting ``(uri, storage_options, namespace_path)``
+        on the connection. The permutation table — always an in-memory
+        LanceDB table — is captured as a pyarrow Table (which pickles via
+        Arrow IPC natively). The reader is dropped from the wire format;
+        ``__setstate__`` rebuilds it from the restored tables.
        """
        permutation_data: Optional[pa.Table] = None
        if self.permutation_table is not None:
@@ -662,9 +622,39 @@ class Permutation:
            # namespace from the existing connection.
            return common

+        # URI-introspection fallback: only viable for native (OSS) connections
+        # where (uri, storage_options) is enough to reopen. Remote / cloud
+        # connections don't expose recoverable api_key / region — those users
+        # must call with_connection_factory().
+        try:
+            base_uri = self.base_table._conn.uri
+            storage_options = self.base_table._conn.storage_options
+        except AttributeError as e:
+            raise ValueError(
+                "Cannot pickle this Permutation: the base table's connection "
+                "does not expose a uri/storage_options, which usually means it "
+                "is a remote (LanceDB Cloud) connection. Call "
+                "Permutation.with_connection_factory(...) first to provide a "
+                "picklable callable that re-opens the base table from a worker "
+                "process."
+            ) from e
+
+        if base_uri.startswith("memory://"):
+            # In-memory base tables don't exist in any worker process by
+            # default, so dump the entire base table into the pickle. This
+            # can be expensive for large datasets — users with large
+            # in-memory base tables should either persist them or set a
+            # connection_factory.
+            return {
+                **common,
+                "base_table_data": self.base_table.to_arrow(),
+            }
+
        return {
            **common,
-            "base_table_state": _table_to_pickle_state(self.base_table),
+            "base_table_uri": base_uri,
+            "base_table_namespace": self.base_table._namespace_path,
+            "base_table_storage_options": storage_options,
        }

    def __setstate__(self, state: dict[str, Any]) -> None:
@@ -673,8 +663,6 @@ class Permutation:
        connection_factory = state["connection_factory"]
        if connection_factory is not None:
            base_table = connection_factory(state["base_table_name"])
-        elif "base_table_state" in state:
-            base_table = _table_from_pickle_state(state["base_table_state"])
        elif "base_table_data" in state:
            # In-memory base table inlined into the pickle; rebuild the same
            # way we rebuild the in-memory permutation table.
@@ -692,7 +680,7 @@ class Permutation:
                namespace_path=state["base_table_namespace"] or None,
            )

-        permutation_table: Optional[Table] = None
+        permutation_table: Optional[LanceTable] = None
        if state["permutation_data"] is not None:
            mem_db = connect("memory://")
            permutation_table = mem_db.create_table(
@@ -708,28 +696,10 @@ class Permutation:
        self.offset = state["offset"]
        self.limit = state["limit"]
        self.connection_factory = connection_factory
-        self.reader = None
-        self._pid = None
-
-    def _ensure_open(self) -> None:
-        pid = os.getpid()
-        if self.reader is not None and getattr(self, "_pid", None) == pid:
-            return
-        # The reader owns Rust-side table handles. Rebuild it after unpickle or
-        # fork even though the Python table wrappers reopen themselves.
-        if hasattr(self.base_table, "_ensure_open"):
-            self.base_table._ensure_open()
-        if self.permutation_table is not None and hasattr(
-            self.permutation_table, "_ensure_open"
-        ):
-            self.permutation_table._ensure_open()
        self.reader = LOOP.run(self._build_reader())
-        self._pid = pid

    @property
    def schema(self) -> pa.Schema:
-        self._ensure_open()
-
        async def do_output_schema():
            return await self.reader.output_schema(self.selection)

@@ -747,7 +717,6 @@ class Permutation:
        """
        The number of rows in the permutation
        """
-        self._ensure_open()
        return self.reader.count_rows()

    @property
@@ -906,7 +875,6 @@ class Permutation:
        If skip_last_batch is True, the last batch will be skipped if it is not a
        multiple of batch_size.
        """
-        self._ensure_open()

        async def get_iter():
            return await self.reader.read(self.selection, batch_size=batch_size)
@@ -1008,7 +976,6 @@ class Permutation:
        so `with_format` and `with_transform` affect this method in the same way
        they affect iteration.
        """
-        self._ensure_open()

        async def do_take_offsets():
            return await self.reader.take_offsets(offsets, selection=self.selection)
@@ -1044,11 +1011,9 @@ class Permutation:
        """
        Skip the first `skip` rows of the permutation
        """
-        self._ensure_open()
        new = copy.copy(self)
        new.offset = skip
        new.reader = LOOP.run(new._build_reader())
-        new._pid = os.getpid()
        return new

    @deprecated(details="Use with_take instead")
@@ -1067,11 +1032,9 @@ class Permutation:
        """
        Limit the permutation to `limit` rows (following any `skip`)
        """
-        self._ensure_open()
        new = copy.copy(self)
        new.limit = limit
        new.reader = LOOP.run(new._build_reader())
-        new._pid = os.getpid()
        return new

    @deprecated(details="Use with_repeat instead")
--- a/python/python/lancedb/query.py
+++ b/python/python/lancedb/query.py
@@ -41,14 +41,6 @@ from .rerankers.rrf import RRFReranker
 from .rerankers.util import check_reranker_result
 from .util import flatten_columns

-BlobMode = Literal["lazy", "bytes", "descriptions"]
-
-_BLOB_MODE_TO_HANDLING = {
-    "lazy": "blobs_descriptions",
-    "bytes": "all_binary",
-    "descriptions": "blobs_descriptions",
-}
-
 if TYPE_CHECKING:
    import sys

@@ -63,7 +55,7 @@ if TYPE_CHECKING:
    from ._lancedb import VectorQuery as LanceVectorQuery
    from .common import VEC
    from .pydantic import LanceModel
-    from .table import AsyncTable, Table
+    from .table import Table

    if sys.version_info >= (3, 11):
        from typing import Self
@@ -73,147 +65,6 @@ if TYPE_CHECKING:
 T = TypeVar("T", bound="LanceModel")


-def _validate_blob_mode(blob_mode: BlobMode) -> None:
-    if blob_mode not in _BLOB_MODE_TO_HANDLING:
-        modes = ", ".join(repr(mode) for mode in _BLOB_MODE_TO_HANDLING)
-        raise ValueError(f"blob_mode must be one of {modes}, got {blob_mode!r}")
-
-
-def _field_is_blob(field: pa.Field) -> bool:
-    metadata = field.metadata or {}
-    return metadata.get(b"lance-encoding:blob") == b"true" or (
-        metadata.get("lance-encoding:blob") == "true"
-    )
-
-
-def _schema_has_blob_field(schema: pa.Schema) -> bool:
-    return any(_field_is_blob(field) for field in schema)
-
-
-def _blob_mode_requires_native_pandas(blob_mode: BlobMode, schema: pa.Schema) -> bool:
-    return blob_mode in ("lazy", "bytes") and _schema_has_blob_field(schema)
-
-
-def _unsupported_blob_pandas_error(reason: str) -> RuntimeError:
-    return RuntimeError(
-        "blob_mode='lazy' and blob_mode='bytes' require Lance native pandas "
-        f"conversion for queries that return blob columns, but {reason}. "
-        "Use blob_mode='descriptions' or remove blob columns from the projection."
-    )
-
-
-def _query_is_plain_scan(query: Query) -> bool:
-    return (
-        query.vector is None
-        and query.full_text_query is None
-        and not query.postfilter
-        and not query.order_by
-    )
-
-
-def _filter_to_sql(filter: Optional[Union[str, Expr]]) -> Optional[str]:
-    if filter is None:
-        return None
-    if isinstance(filter, Expr):
-        return filter.to_sql()
-    return filter
-
-
-def _projection_to_scanner_kwargs(
-    columns: Optional[
-        Union[
-            List[str], List[Tuple[str, Union[str, Expr]]], Dict[str, Union[str, Expr]]
-        ]
-    ],
-) -> Dict[str, Any]:
-    if columns is None:
-        return {}
-    if isinstance(columns, list):
-        if all(isinstance(column, str) for column in columns):
-            return {"columns": columns}
-        if all(isinstance(column, tuple) and len(column) == 2 for column in columns):
-            return {
-                "columns": {
-                    name: expr.to_sql() if isinstance(expr, Expr) else expr
-                    for name, expr in columns
-                }
-            }
-        # Let Lance raise the detailed projection validation error.
-        return {"columns": columns}
-
-    projection = {}
-    for name, expr in columns.items():
-        if isinstance(expr, Expr):
-            expr = expr.to_sql()
-        projection[name] = expr
-    return {"columns": projection}
-
-
-def _scanner_kwargs_for_query(query: Query, blob_mode: BlobMode) -> Dict[str, Any]:
-    kwargs = {
-        **_projection_to_scanner_kwargs(query.columns),
-        "filter": _filter_to_sql(query.filter),
-        "limit": query.limit,
-        "offset": query.offset,
-        "with_row_id": query.with_row_id,
-        "fast_search": query.fast_search,
-        "blob_handling": _BLOB_MODE_TO_HANDLING[blob_mode],
-    }
-    return {key: value for key, value in kwargs.items() if value is not None}
-
-
-def _ensure_lazy_blob_frame(
-    df: "pd.DataFrame", schema: pa.Schema, blob_mode: BlobMode
-) -> "pd.DataFrame":
-    if blob_mode != "lazy" or not _schema_has_blob_field(schema) or len(df) == 0:
-        return df
-
-    for field in schema:
-        if not _field_is_blob(field) or field.name not in df.columns:
-            continue
-        value = df[field.name].iloc[0]
-        if value is not None and not hasattr(value, "readall"):
-            raise _unsupported_blob_pandas_error(
-                "the Lance scanner did not return lazy blob files"
-            )
-    return df
-
-
-def _scanner_to_pandas(scanner: Any, blob_mode: BlobMode, **kwargs) -> "pd.DataFrame":
-    schema = getattr(scanner, "projected_schema", None)
-    if schema is None:
-        schema = getattr(scanner, "schema", None)
-    if schema is None:
-        schema = getattr(scanner, "dataset_schema", None)
-    if callable(schema):
-        schema = schema()
-    if hasattr(scanner, "to_pandas"):
-        try:
-            df = scanner.to_pandas(blob_mode=blob_mode, **kwargs)
-        except TypeError as err:
-            message = str(err)
-            if "blob_mode" not in message and "unexpected keyword" not in message:
-                raise
-            df = scanner.to_pandas(**kwargs)
-        if schema is not None:
-            return _ensure_lazy_blob_frame(df, schema, blob_mode)
-        return df
-
-    if hasattr(scanner, "to_pyarrow"):
-        reader = scanner.to_pyarrow()
-        tbl = reader.read_all()
-    elif hasattr(scanner, "to_table"):
-        tbl = scanner.to_table()
-    else:
-        reader = scanner.to_reader()
-        tbl = reader.read_all()
-    if blob_mode == "lazy" and _schema_has_blob_field(tbl.schema):
-        raise _unsupported_blob_pandas_error(
-            "the Lance scanner does not expose to_pandas"
-        )
-    return tbl.to_pandas(**kwargs)
-
-
 # Pydantic validation function for vector queries
 def ensure_vector_query(
    val: Any,
@@ -867,7 +718,6 @@ class LanceQueryBuilder(ABC):
        self,
        flatten: Optional[Union[int, bool]] = None,
        *,
-        blob_mode: BlobMode = "lazy",
        timeout: Optional[timedelta] = None,
        **kwargs,
    ) -> "pd.DataFrame":
@@ -887,31 +737,11 @@ class LanceQueryBuilder(ABC):
        timeout: Optional[timedelta]
            The maximum time to wait for the query to complete.
            If None, wait indefinitely.
-        blob_mode: str, default "lazy"
-            Controls how blob columns are returned for plain scan queries.
-            Vector, FTS, hybrid, and other non-native query shapes keep the
-            existing Arrow conversion path and only support blob descriptions.
        **kwargs
            Forwarded to pyarrow.Table.to_pandas after query execution and
            optional flattening.
        """
-        _validate_blob_mode(blob_mode)
-        native_error = None
        tbl = flatten_columns(self.to_arrow(timeout=timeout), flatten)
-        if _blob_mode_requires_native_pandas(blob_mode, tbl.schema):
-            if flatten is None and timeout is None:
-                try:
-                    df = self._plain_scan_to_pandas(blob_mode, **kwargs)
-                    if df is not None:
-                        return df
-                except Exception as err:
-                    native_error = err
-            reason = (
-                "this query shape cannot use Lance native pandas conversion"
-                if native_error is None
-                else str(native_error)
-            )
-            raise _unsupported_blob_pandas_error(reason) from native_error
        return tbl.to_pandas(**kwargs)

    @abstractmethod
@@ -1256,19 +1086,6 @@ class LanceQueryBuilder(ABC):
        """
        raise NotImplementedError

-    def _plain_scan_to_pandas(
-        self,
-        blob_mode: BlobMode,
-        **kwargs,
-    ) -> Optional["pd.DataFrame"]:
-        query = self.to_query_object()
-        if not _query_is_plain_scan(query):
-            return None
-
-        dataset = self._table.to_lance()
-        scanner = dataset.scanner(**_scanner_kwargs_for_query(query, blob_mode))
-        return _scanner_to_pandas(scanner, blob_mode, **kwargs)
-
    @abstractmethod
    def to_query_object(self) -> Query:
        """Return a serializable representation of the query
@@ -2390,11 +2207,7 @@ class AsyncQueryBase(object):
    Base class for all async queries (take, scan, vector, fts, hybrid)
    """

-    def __init__(
-        self,
-        inner: Union[LanceQuery, LanceVectorQuery, LanceTakeQuery],
-        table: Optional["AsyncTable"] = None,
-    ):
+    def __init__(self, inner: Union[LanceQuery, LanceVectorQuery, LanceTakeQuery]):
        """
        Construct an AsyncQueryBase

@@ -2402,7 +2215,6 @@ class AsyncQueryBase(object):
        [AsyncTable.query][lancedb.table.AsyncTable.query] method to create a query.
        """
        self._inner = inner
-        self._table = table

    def to_query_object(self) -> Query:
        """
@@ -2545,8 +2357,6 @@ class AsyncQueryBase(object):
        self,
        flatten: Optional[Union[int, bool]] = None,
        timeout: Optional[timedelta] = None,
-        *,
-        blob_mode: BlobMode = "lazy",
        **kwargs,
    ) -> "pd.DataFrame":
        """
@@ -2580,48 +2390,13 @@ class AsyncQueryBase(object):
            The maximum time to wait for the query to complete.
            If not specified, no timeout is applied. If the query does not
            complete within the specified time, an error will be raised.
-        blob_mode: str, default "lazy"
-            Controls how blob columns are returned for plain scan queries.
-            Vector, FTS, hybrid, and other non-native query shapes keep the
-            existing Arrow conversion path and only support blob descriptions.
        **kwargs
            Forwarded to pyarrow.Table.to_pandas after query execution and
            optional flattening.
        """
-        _validate_blob_mode(blob_mode)
-        native_error = None
-        tbl = flatten_columns(await self.to_arrow(timeout=timeout), flatten)
-        if _blob_mode_requires_native_pandas(blob_mode, tbl.schema):
-            if flatten is None and timeout is None:
-                try:
-                    df = await self._plain_scan_to_pandas(blob_mode, **kwargs)
-                    if df is not None:
-                        return df
-                except Exception as err:
-                    native_error = err
-            reason = (
-                "this query shape cannot use Lance native pandas conversion"
-                if native_error is None
-                else str(native_error)
-            )
-            raise _unsupported_blob_pandas_error(reason) from native_error
-        return tbl.to_pandas(**kwargs)
-
-    async def _plain_scan_to_pandas(
-        self,
-        blob_mode: BlobMode,
-        **kwargs,
-    ) -> Optional["pd.DataFrame"]:
-        if self._table is None:
-            return None
-
-        query = self.to_query_object()
-        if not _query_is_plain_scan(query):
-            return None
-
-        dataset = await self._table._to_lance()
-        scanner = dataset.scanner(**_scanner_kwargs_for_query(query, blob_mode))
-        return _scanner_to_pandas(scanner, blob_mode, **kwargs)
+        return (
+            flatten_columns(await self.to_arrow(timeout=timeout), flatten)
+        ).to_pandas(**kwargs)

    async def to_polars(
        self,
@@ -2728,18 +2503,14 @@ class AsyncStandardQuery(AsyncQueryBase):
    Base class for "standard" async queries (all but take currently)
    """

-    def __init__(
-        self,
-        inner: Union[LanceQuery, LanceVectorQuery],
-        table: Optional["AsyncTable"] = None,
-    ):
+    def __init__(self, inner: Union[LanceQuery, LanceVectorQuery]):
        """
        Construct an AsyncStandardQuery

        This method is not intended to be called directly.  Instead, use the
        [AsyncTable.query][lancedb.table.AsyncTable.query] method to create a query.
        """
-        super().__init__(inner, table)
+        super().__init__(inner)

    def where(self, predicate: Union[str, Expr]) -> Self:
        """
@@ -2845,14 +2616,14 @@ class AsyncStandardQuery(AsyncQueryBase):


 class AsyncQuery(AsyncStandardQuery):
-    def __init__(self, inner: LanceQuery, table: Optional["AsyncTable"] = None):
+    def __init__(self, inner: LanceQuery):
        """
        Construct an AsyncQuery

        This method is not intended to be called directly.  Instead, use the
        [AsyncTable.query][lancedb.table.AsyncTable.query] method to create a query.
        """
-        super().__init__(inner, table)
+        super().__init__(inner)
        self._inner = inner

    @classmethod
@@ -2936,11 +2707,10 @@ class AsyncQuery(AsyncStandardQuery):
            new_self = self._inner.nearest_to(query_vectors[0])
            for v in query_vectors[1:]:
                new_self.add_query_vector(v)
-            return AsyncVectorQuery(new_self, self._table)
+            return AsyncVectorQuery(new_self)
        else:
            return AsyncVectorQuery(
-                self._inner.nearest_to(AsyncQuery._query_vec_to_array(query_vector)),
-                self._table,
+                self._inner.nearest_to(AsyncQuery._query_vec_to_array(query_vector))
            )

    def nearest_to_text(
@@ -2973,18 +2743,17 @@ class AsyncQuery(AsyncStandardQuery):

        if isinstance(query, str):
            return AsyncFTSQuery(
-                self._inner.nearest_to_text({"query": query, "columns": columns}),
-                self._table,
+                self._inner.nearest_to_text({"query": query, "columns": columns})
            )
        # FullTextQuery object
-        return AsyncFTSQuery(self._inner.nearest_to_text({"query": query}), self._table)
+        return AsyncFTSQuery(self._inner.nearest_to_text({"query": query}))


 class AsyncFTSQuery(AsyncStandardQuery):
    """A query for full text search for LanceDB."""

-    def __init__(self, inner: LanceFTSQuery, table: Optional["AsyncTable"] = None):
-        super().__init__(inner, table)
+    def __init__(self, inner: LanceFTSQuery):
+        super().__init__(inner)
        self._inner = inner
        self._reranker = None

@@ -3066,11 +2835,10 @@ class AsyncFTSQuery(AsyncStandardQuery):
            new_self = self._inner.nearest_to(query_vectors[0])
            for v in query_vectors[1:]:
                new_self.add_query_vector(v)
-            return AsyncHybridQuery(new_self, self._table)
+            return AsyncHybridQuery(new_self)
        else:
            return AsyncHybridQuery(
-                self._inner.nearest_to(AsyncQuery._query_vec_to_array(query_vector)),
-                self._table,
+                self._inner.nearest_to(AsyncQuery._query_vec_to_array(query_vector))
            )

    async def to_batches(
@@ -3261,7 +3029,7 @@ class AsyncVectorQueryBase:


 class AsyncVectorQuery(AsyncStandardQuery, AsyncVectorQueryBase):
-    def __init__(self, inner: LanceVectorQuery, table: Optional["AsyncTable"] = None):
+    def __init__(self, inner: LanceVectorQuery):
        """
        Construct an AsyncVectorQuery

@@ -3271,7 +3039,7 @@ class AsyncVectorQuery(AsyncStandardQuery, AsyncVectorQueryBase):
        a vector query.  Or you can use
        [AsyncTable.vector_search][lancedb.table.AsyncTable.vector_search]
        """
-        super().__init__(inner, table)
+        super().__init__(inner)
        self._inner = inner
        self._reranker = None
        self._query_string = None
@@ -3325,13 +3093,10 @@ class AsyncVectorQuery(AsyncStandardQuery, AsyncVectorQueryBase):

        if isinstance(query, str):
            return AsyncHybridQuery(
-                self._inner.nearest_to_text({"query": query, "columns": columns}),
-                self._table,
+                self._inner.nearest_to_text({"query": query, "columns": columns})
            )
        # FullTextQuery object
-        return AsyncHybridQuery(
-            self._inner.nearest_to_text({"query": query}), self._table
-        )
+        return AsyncHybridQuery(self._inner.nearest_to_text({"query": query}))

    async def to_batches(
        self,
@@ -3358,8 +3123,8 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
    in the `rerank` method to convert the scores to ranks and then normalize them.
    """

-    def __init__(self, inner: LanceHybridQuery, table: Optional["AsyncTable"] = None):
-        super().__init__(inner, table)
+    def __init__(self, inner: LanceHybridQuery):
+        super().__init__(inner)
        self._inner = inner
        self._norm = "score"
        self._reranker = RRFReranker()
@@ -3400,8 +3165,8 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
        max_batch_length: Optional[int] = None,
        timeout: Optional[timedelta] = None,
    ) -> AsyncRecordBatchReader:
-        fts_query = AsyncFTSQuery(self._inner.to_fts_query(), self._table)
-        vec_query = AsyncVectorQuery(self._inner.to_vector_query(), self._table)
+        fts_query = AsyncFTSQuery(self._inner.to_fts_query())
+        vec_query = AsyncVectorQuery(self._inner.to_vector_query())

        # save the row ID choice that was made on the query builder and force it
        # to actually fetch the row ids because we need this for reranking
@@ -3501,15 +3266,8 @@ class AsyncTakeQuery(AsyncQueryBase):
    Builder for parameterizing and executing take queries.
    """

-    def __init__(self, inner: LanceTakeQuery, table: Optional["AsyncTable"] = None):
-        super().__init__(inner, table)
-
-    async def _plain_scan_to_pandas(
-        self,
-        blob_mode: BlobMode,
-        **kwargs,
-    ) -> Optional["pd.DataFrame"]:
-        return None
+    def __init__(self, inner: LanceTakeQuery):
+        super().__init__(inner)


 class BaseQueryBuilder(object):
@@ -3642,8 +3400,6 @@ class BaseQueryBuilder(object):
        self,
        flatten: Optional[Union[int, bool]] = None,
        timeout: Optional[timedelta] = None,
-        *,
-        blob_mode: BlobMode = "lazy",
        **kwargs,
    ) -> "pd.DataFrame":
        """
@@ -3677,15 +3433,11 @@ class BaseQueryBuilder(object):
            The maximum time to wait for the query to complete.
            If not specified, no timeout is applied. If the query does not
            complete within the specified time, an error will be raised.
-        blob_mode: str, default "lazy"
-            Controls how blob columns are returned for plain scan queries.
        **kwargs
            Forwarded to pyarrow.Table.to_pandas after query execution and
            optional flattening.
        """
-        return LOOP.run(
-            self._inner.to_pandas(flatten, timeout, blob_mode=blob_mode, **kwargs)
-        )
+        return LOOP.run(self._inner.to_pandas(flatten, timeout, **kwargs))

    def to_polars(
        self,
--- a/python/python/lancedb/remote/db.py
+++ b/python/python/lancedb/remote/db.py
@@ -3,7 +3,6 @@


 from datetime import timedelta
-import json
 import logging
 from concurrent.futures import ThreadPoolExecutor
 import sys
@@ -18,7 +17,7 @@ else:

 # Remove this import to fix circular dependency
 # from lancedb import connect_async
-from lancedb.remote import ClientConfig, RetryConfig, TimeoutConfig, TlsConfig
+from lancedb.remote import ClientConfig
 import pyarrow as pa

 from ..common import DATA
@@ -37,64 +36,6 @@ from ..table import Table
 from ..util import validate_table_name


-def _duration_seconds(value: Optional[timedelta]) -> Optional[float]:
-    return value.total_seconds() if value is not None else None
-
-
-def _timeout_config_to_dict(
-    config: Optional[TimeoutConfig],
-) -> Optional[dict[str, Any]]:
-    if config is None:
-        return None
-    return {
-        "timeout": _duration_seconds(config.timeout),
-        "connect_timeout": _duration_seconds(config.connect_timeout),
-        "read_timeout": _duration_seconds(config.read_timeout),
-        "pool_idle_timeout": _duration_seconds(config.pool_idle_timeout),
-    }
-
-
-def _retry_config_to_dict(config: RetryConfig) -> dict[str, Any]:
-    return {
-        "retries": config.retries,
-        "connect_retries": config.connect_retries,
-        "read_retries": config.read_retries,
-        "backoff_factor": config.backoff_factor,
-        "backoff_jitter": config.backoff_jitter,
-        "statuses": config.statuses,
-    }
-
-
-def _tls_config_to_dict(config: Optional[TlsConfig]) -> Optional[dict[str, Any]]:
-    if config is None:
-        return None
-    return {
-        "cert_file": config.cert_file,
-        "key_file": config.key_file,
-        "ssl_ca_cert": config.ssl_ca_cert,
-        "assert_hostname": config.assert_hostname,
-    }
-
-
-def _client_config_to_dict(config: ClientConfig) -> dict[str, Any]:
-    if config.header_provider is not None:
-        raise ValueError(
-            "Cannot serialize a remote connection with a header_provider. "
-            "Use static api_key/extra_headers or provide a worker-side "
-            "connection factory instead."
-        )
-    return {
-        "user_agent": config.user_agent,
-        "retry_config": _retry_config_to_dict(config.retry_config),
-        "timeout_config": _timeout_config_to_dict(config.timeout_config),
-        "extra_headers": config.extra_headers,
-        "id_delimiter": config.id_delimiter,
-        "tls_config": _tls_config_to_dict(config.tls_config),
-        "header_provider": None,
-        "user_id": config.user_id,
-    }
-
-
 class RemoteDBConnection(DBConnection):
    """A connection to a remote LanceDB database."""

@@ -148,11 +89,6 @@ class RemoteDBConnection(DBConnection):
        parsed = urlparse(db_url)
        if parsed.scheme != "db":
            raise ValueError(f"Invalid scheme: {parsed.scheme}, only accepts db://")
-        self.db_url = db_url
-        self.api_key = api_key
-        self.region = region
-        self.host_override = host_override
-        self.storage_options = storage_options
        self.db_name = parsed.netloc

        self.client_config = client_config
@@ -175,20 +111,6 @@ class RemoteDBConnection(DBConnection):
    def __repr__(self) -> str:
        return f"RemoteConnect(name={self.db_name})"

-    @override
-    def serialize(self) -> str:
-        return json.dumps(
-            {
-                "connection_type": "remote",
-                "db_url": self.db_url,
-                "api_key": self.api_key,
-                "region": self.region,
-                "host_override": self.host_override,
-                "client_config": _client_config_to_dict(self.client_config),
-                "storage_options": self.storage_options,
-            }
-        )
-
    @override
    def list_namespaces(
        self,
@@ -409,12 +331,7 @@ class RemoteDBConnection(DBConnection):
            )

        table = LOOP.run(self._conn.open_table(name, namespace_path=namespace_path))
-        return RemoteTable(
-            table,
-            self.db_name,
-            connection_state=self.serialize,
-            namespace_path=namespace_path,
-        )
+        return RemoteTable(table, self.db_name)

    def clone_table(
        self,
@@ -463,12 +380,7 @@ class RemoteDBConnection(DBConnection):
                is_shallow=is_shallow,
            )
        )
-        return RemoteTable(
-            table,
-            self.db_name,
-            connection_state=self.serialize,
-            namespace_path=target_namespace_path,
-        )
+        return RemoteTable(table, self.db_name)

    @override
    def create_table(
@@ -613,12 +525,7 @@ class RemoteDBConnection(DBConnection):
                fill_value=fill_value,
            )
        )
-        return RemoteTable(
-            table,
-            self.db_name,
-            connection_state=self.serialize,
-            namespace_path=namespace_path,
-        )
+        return RemoteTable(table, self.db_name)

    @override
    def drop_table(self, name: str, namespace_path: Optional[List[str]] = None):
--- a/python/python/lancedb/remote/table.py
+++ b/python/python/lancedb/remote/table.py
@@ -2,25 +2,11 @@
 # SPDX-FileCopyrightText: Copyright The LanceDB Authors

 from datetime import timedelta
-import deprecation
 import logging
 from functools import cached_property
-import os
-from typing import (
-    Any,
-    Callable,
-    Dict,
-    Iterable,
-    List,
-    Optional,
-    Union,
-    Literal,
-    overload,
-)
+from typing import Any, Callable, Dict, Iterable, List, Optional, Union, Literal
 import warnings

-from lancedb import __version__
-
 from lancedb._lancedb import (
    AddColumnsResult,
    AddResult,
@@ -46,7 +32,6 @@ from lancedb.index import (
    LabelList,
 )
 from lancedb.remote.db import LOOP
-from lancedb.table import IndexConfigType, KNOWN_METRICS
 import pyarrow as pa

 from lancedb.common import DATA, VEC, VECTOR_COLUMN_NAME
@@ -64,80 +49,14 @@ class RemoteTable(Table):
        self,
        table: AsyncTable,
        db_name: str,
-        *,
-        connection_state: Optional[Union[str, Callable[[], str]]] = None,
-        namespace_path: Optional[List[str]] = None,
    ):
-        self._table_handle = table
-        self._name = table.name
+        self._table = table
        self.db_name = db_name
-        self._connection_state = connection_state
-        self._namespace_path = list(namespace_path or [])
-        self._checkout_version: Optional[int] = None
-        self._pid = os.getpid()
-
-    def _serialized_connection_state(self) -> str:
-        if self._connection_state is None:
-            raise RuntimeError(
-                "Cannot reopen this remote table because it does not carry "
-                "serialized connection state"
-            )
-        if callable(self._connection_state):
-            self._connection_state = self._connection_state()
-        return self._connection_state
-
-    @property
-    def _table(self) -> AsyncTable:
-        self._ensure_open()
-        assert self._table_handle is not None
-        return self._table_handle
-
-    @_table.setter
-    def _table(self, table: AsyncTable) -> None:
-        self._table_handle = table
-        self._name = table.name
-        self._pid = os.getpid()
-
-    def _ensure_open(self) -> None:
-        pid = os.getpid()
-        if self._table_handle is not None and self._pid == pid:
-            return
-
-        # Pickle clears the handle; fork inherits a handle created in the
-        # parent process. In both cases reopen before touching the Rust client.
-        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)
-
-        self._table_handle = table._table
-        self.db_name = table.db_name
-        self._pid = pid
-
-    def __getstate__(self) -> dict:
-        return {
-            "connection_state": self._serialized_connection_state(),
-            "db_name": self.db_name,
-            "name": self.name,
-            "namespace_path": self._namespace_path,
-            "checkout_version": self._checkout_version,
-        }
-
-    def __setstate__(self, state: dict) -> None:
-        self._table_handle = None
-        self._name = state["name"]
-        self.db_name = state["db_name"]
-        self._connection_state = state["connection_state"]
-        self._namespace_path = state["namespace_path"]
-        self._checkout_version = state["checkout_version"]
-        self._pid = None

    @property
    def name(self) -> str:
        """The name of the table"""
-        return self._name
+        return self._table.name

    def __repr__(self) -> str:
        return f"RemoteTable({self.db_name}.{self.name})"
@@ -187,19 +106,13 @@ class RemoteTable(Table):
        raise NotImplementedError("to_pandas() is not yet supported on LanceDB cloud.")

    def checkout(self, version: Union[int, str]):
-        result = LOOP.run(self._table.checkout(version))
-        self._checkout_version = self.version
-        return result
+        return LOOP.run(self._table.checkout(version))

    def checkout_latest(self):
-        result = LOOP.run(self._table.checkout_latest())
-        self._checkout_version = None
-        return result
+        return LOOP.run(self._table.checkout_latest())

    def restore(self, version: Optional[Union[int, str]] = None):
-        result = LOOP.run(self._table.restore(version))
-        self._checkout_version = None
-        return result
+        return LOOP.run(self._table.restore(version))

    def list_indices(self) -> Iterable[IndexConfig]:
        """List all the indices on the table"""
@@ -209,11 +122,6 @@ class RemoteTable(Table):
        """List all the stats of a specified index"""
        return LOOP.run(self._table.index_stats(index_uuid))

-    @deprecation.deprecated(
-        deprecated_in="0.25.0",
-        current_version=__version__,
-        details="Use create_index() with config=BTree()/Bitmap()/LabelList() instead.",
-    )
    def create_scalar_index(
        self,
        column: str,
@@ -223,12 +131,7 @@ class RemoteTable(Table):
        wait_timeout: Optional[timedelta] = None,
        name: Optional[str] = None,
    ):
-        """Creates a scalar index.
-
-        .. deprecated:: 0.25.0
-            Use :meth:`create_index` with a BTree, Bitmap, or LabelList config instead.
-            Example: ``table.create_index("column", config=BTree())``
-
+        """Creates a scalar index
        Parameters
        ----------
        column : str
@@ -259,11 +162,6 @@ class RemoteTable(Table):
            )
        )

-    @deprecation.deprecated(
-        deprecated_in="0.25.0",
-        current_version=__version__,
-        details="Use create_index() with config=FTS() instead.",
-    )
    def create_fts_index(
        self,
        column: str,
@@ -284,12 +182,6 @@ class RemoteTable(Table):
        prefix_only: bool = False,
        name: Optional[str] = None,
    ):
-        """Create a full-text search index on a column.
-
-        .. deprecated:: 0.25.0
-            Use :meth:`create_index` with an FTS config instead.
-            Example: ``table.create_index("text_column", config=FTS())``
-        """
        config = FTS(
            with_position=with_position,
            base_tokenizer=base_tokenizer,
@@ -313,43 +205,9 @@ class RemoteTable(Table):
            )
        )

-    # New unified API overload
-    @overload
    def create_index(
        self,
-        column: str,
-        /,
-        *,
-        config: IndexConfigType,
-        wait_timeout: Optional[timedelta] = ...,
-        name: Optional[str] = ...,
-        train: bool = ...,
-    ) -> None: ...
-
-    # Legacy API overload (deprecated)
-    @overload
-    def create_index(
-        self,
-        metric: Literal["l2", "cosine", "dot", "hamming"] = ...,
-        vector_column_name: str = ...,
-        index_cache_size: Optional[int] = ...,
-        num_partitions: Optional[int] = ...,
-        num_sub_vectors: Optional[int] = ...,
-        replace: Optional[bool] = ...,
-        accelerator: Optional[str] = ...,
-        index_type: Literal[
-            "VECTOR", "IVF_FLAT", "IVF_SQ", "IVF_PQ", "IVF_HNSW_SQ", "IVF_HNSW_PQ"
-        ] = ...,
-        wait_timeout: Optional[timedelta] = ...,
-        *,
-        num_bits: int = ...,
-        name: Optional[str] = ...,
-        train: bool = ...,
-    ) -> None: ...
-
-    def create_index(
-        self,
-        metric: str = "l2",
+        metric="l2",
        vector_column_name: str = VECTOR_COLUMN_NAME,
        index_cache_size: Optional[int] = None,
        num_partitions: Optional[int] = None,
@@ -360,113 +218,89 @@ class RemoteTable(Table):
        wait_timeout: Optional[timedelta] = None,
        *,
        num_bits: int = 8,
-        config: Optional[IndexConfigType] = None,
        name: Optional[str] = None,
        train: bool = True,
    ):
-        """Create an index on a column.
+        """Create an index on the table.

-        This method supports both the new unified API and the legacy API
-        for backwards compatibility. The new API takes the column name as the
-        first positional argument and an index configuration object via
-        ``config``; the legacy API takes the distance metric as the first
-        argument plus separate ``vector_column_name`` / ``num_partitions`` /
-        etc. parameters, and emits a ``DeprecationWarning``.
+        Parameters
+        ----------
+        metric : str
+            The metric to use for the index. Default is "l2".
+        vector_column_name : str
+            The name of the vector column. Default is "vector".

        Examples
        --------
-        New API (recommended):
-
-        >>> table.create_index(  # doctest: +SKIP
-        ...     "vector", config=IvfPq(distance_type="l2")
+        >>> import lancedb
+        >>> import uuid
+        >>> from lancedb.schema import vector
+        >>> db = lancedb.connect("db://...", api_key="...", # doctest: +SKIP
+        ...                      region="...") # doctest: +SKIP
+        >>> table_name = uuid.uuid4().hex
+        >>> schema = pa.schema(
+        ...     [
+        ...             pa.field("id", pa.uint32(), False),
+        ...            pa.field("vector", vector(128), False),
+        ...             pa.field("s", pa.string(), False),
+        ...     ]
        ... )
-        >>> table.create_index("category", config=BTree())  # doctest: +SKIP
-        >>> table.create_index("content", config=FTS())  # doctest: +SKIP
-
-        Legacy API (deprecated):
-
-        >>> table.create_index(  # doctest: +SKIP
-        ...     "l2", vector_column_name="vector"
+        >>> table = db.create_table( # doctest: +SKIP
+        ...     table_name, # doctest: +SKIP
+        ...     schema=schema, # doctest: +SKIP
        ... )
+        >>> table.create_index("l2", "vector") # doctest: +SKIP
        """
-        # Detect whether this is a legacy API call
-        is_legacy = self._is_legacy_create_index_call(
-            metric,
-            config,
-            num_partitions,
-            num_sub_vectors,
-            vector_column_name,
-            accelerator,
-            index_cache_size,
-            replace,
-        )

-        if is_legacy:
-            warnings.warn(
-                "The create_index() API with metric/num_partitions parameters is "
-                "deprecated and will be removed in a future version. "
-                "Please migrate to the new unified API:\n"
-                "  # Old (deprecated):\n"
-                "  table.create_index('l2', vector_column_name='my_vector')\n"
-                "  # New (recommended):\n"
-                "  table.create_index('my_vector', config=IvfPq(distance_type='l2'))",
-                DeprecationWarning,
-                stacklevel=2,
+        if accelerator is not None:
+            logging.warning(
+                "GPU accelerator is not yet supported on LanceDB cloud."
+                "If you have 100M+ vectors to index,"
+                "please contact us at contact@lancedb.com"
+            )
+        if replace is not None:
+            logging.warning(
+                "replace is not supported on LanceDB cloud."
+                "Existing indexes will always be replaced."
            )

-            column = vector_column_name
-
-            if accelerator is not None:
-                logging.warning(
-                    "GPU accelerator is not yet supported on LanceDB cloud."
-                    "If you have 100M+ vectors to index,"
-                    "please contact us at contact@lancedb.com"
-                )
-            if replace is not None:
-                logging.warning(
-                    "replace is not supported on LanceDB cloud."
-                    "Existing indexes will always be replaced."
-                )
-
-            idx_type = index_type.upper()
-            if idx_type == "VECTOR" or idx_type == "IVF_PQ":
-                config = IvfPq(
-                    distance_type=metric,
-                    num_partitions=num_partitions,
-                    num_sub_vectors=num_sub_vectors,
-                    num_bits=num_bits,
-                )
-            elif idx_type == "IVF_RQ":
-                config = IvfRq(
-                    distance_type=metric,
-                    num_partitions=num_partitions,
-                    num_bits=num_bits,
-                )
-            elif idx_type == "IVF_SQ":
-                config = IvfSq(distance_type=metric, num_partitions=num_partitions)
-            elif idx_type == "IVF_HNSW_PQ":
-                raise ValueError(
-                    "IVF_HNSW_PQ is not supported on LanceDB cloud."
-                    "Please use IVF_HNSW_SQ instead."
-                )
-            elif idx_type == "IVF_HNSW_SQ":
-                config = HnswSq(distance_type=metric, num_partitions=num_partitions)
-            elif idx_type == "IVF_HNSW_FLAT":
-                config = HnswFlat(distance_type=metric, num_partitions=num_partitions)
-            elif idx_type == "IVF_FLAT":
-                config = IvfFlat(distance_type=metric, num_partitions=num_partitions)
-            else:
-                raise ValueError(
-                    f"Unknown vector index type: {idx_type}. Valid options are"
-                    " 'IVF_FLAT', 'IVF_PQ', 'IVF_RQ', 'IVF_SQ',"
-                    " 'IVF_HNSW_PQ', 'IVF_HNSW_SQ', 'IVF_HNSW_FLAT'"
-                )
+        index_type = index_type.upper()
+        if index_type == "VECTOR" or index_type == "IVF_PQ":
+            config = IvfPq(
+                distance_type=metric,
+                num_partitions=num_partitions,
+                num_sub_vectors=num_sub_vectors,
+                num_bits=num_bits,
+            )
+        elif index_type == "IVF_RQ":
+            config = IvfRq(
+                distance_type=metric,
+                num_partitions=num_partitions,
+                num_bits=num_bits,
+            )
+        elif index_type == "IVF_SQ":
+            config = IvfSq(distance_type=metric, num_partitions=num_partitions)
+        elif index_type == "IVF_HNSW_PQ":
+            raise ValueError(
+                "IVF_HNSW_PQ is not supported on LanceDB cloud."
+                "Please use IVF_HNSW_SQ instead."
+            )
+        elif index_type == "IVF_HNSW_SQ":
+            config = HnswSq(distance_type=metric, num_partitions=num_partitions)
+        elif index_type == "IVF_HNSW_FLAT":
+            config = HnswFlat(distance_type=metric, num_partitions=num_partitions)
+        elif index_type == "IVF_FLAT":
+            config = IvfFlat(distance_type=metric, num_partitions=num_partitions)
        else:
-            column = metric
+            raise ValueError(
+                f"Unknown vector index type: {index_type}. Valid options are"
+                " 'IVF_FLAT', 'IVF_PQ', 'IVF_RQ', 'IVF_SQ',"
+                " 'IVF_HNSW_PQ', 'IVF_HNSW_SQ', 'IVF_HNSW_FLAT'"
+            )

        LOOP.run(
            self._table.create_index(
-                column,
+                vector_column_name,
                config=config,
                wait_timeout=wait_timeout,
                name=name,
@@ -474,37 +308,6 @@ class RemoteTable(Table):
            )
        )

-    def _is_legacy_create_index_call(
-        self,
-        first_arg: str,
-        config: Optional[IndexConfigType],
-        num_partitions: Optional[int],
-        num_sub_vectors: Optional[int],
-        vector_column_name: str,
-        accelerator: Optional[str],
-        index_cache_size: Optional[int],
-        replace: Optional[bool],
-    ) -> bool:
-        """Detect if this is a legacy create_index call."""
-        if config is not None:
-            return False
-        if any(
-            x is not None
-            for x in (
-                num_partitions,
-                num_sub_vectors,
-                accelerator,
-                index_cache_size,
-                replace,
-            )
-        ):
-            return True
-        if vector_column_name != VECTOR_COLUMN_NAME:
-            return True
-        if first_arg.lower() in KNOWN_METRICS:
-            return True
-        return False
-
    def add(
        self,
        data: DATA,
@@ -865,10 +668,6 @@ class RemoteTable(Table):
        """Not supported on LanceDB Cloud."""
        return LOOP.run(self._table.unset_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())
-
    def drop_index(self, index_name: str):
        return LOOP.run(self._table.drop_index(index_name))

--- a/python/python/lancedb/table.py
+++ b/python/python/lancedb/table.py
@@ -89,18 +89,6 @@ from .index import lang_mapping

 BlobMode = Literal["lazy", "bytes", "descriptions"]

-
-def _field_is_blob(field: pa.Field) -> bool:
-    metadata = field.metadata or {}
-    return metadata.get(b"lance-encoding:blob") == b"true" or (
-        metadata.get("lance-encoding:blob") == "true"
-    )
-
-
-def _schema_has_blob_field(schema: pa.Schema) -> bool:
-    return any(_field_is_blob(field) for field in schema)
-
-
 _MODEL_BACKED_TOKENIZER_PREFIXES = ("jieba", "lindera")
 _MODEL_BACKED_TOKENIZER_ERRORS = (
    "unknown base tokenizer",
@@ -186,24 +174,6 @@ if TYPE_CHECKING:
        DistanceType,
    )

-# Type alias for index configuration objects
-IndexConfigType = Union[
-    IvfFlat,
-    IvfPq,
-    IvfSq,
-    IvfRq,
-    HnswFlat,
-    HnswPq,
-    HnswSq,
-    BTree,
-    Bitmap,
-    LabelList,
-    FTS,
-]
-
-# Known distance metrics for legacy API detection
-KNOWN_METRICS = {"l2", "cosine", "dot", "hamming"}
-

 def _into_pyarrow_reader(
    data, schema: Optional[pa.Schema] = None
@@ -837,49 +807,11 @@ class Table(ABC):
        """
        raise NotImplementedError

-    # New unified API overload
-    @overload
    def create_index(
        self,
-        column: str,
-        /,
-        *,
-        config: IndexConfigType,
-        replace: bool = ...,
-        wait_timeout: Optional[timedelta] = ...,
-        name: Optional[str] = ...,
-        train: bool = ...,
-    ) -> None: ...
-
-    # Legacy API overload (deprecated)
-    @overload
-    def create_index(
-        self,
-        metric: Literal["l2", "cosine", "dot", "hamming"] = ...,
-        num_partitions: Optional[int] = ...,
-        num_sub_vectors: Optional[int] = ...,
-        vector_column_name: str = ...,
-        replace: bool = ...,
-        accelerator: Optional[str] = ...,
-        index_cache_size: Optional[int] = ...,
-        *,
-        index_type: VectorIndexType = ...,
-        wait_timeout: Optional[timedelta] = ...,
-        num_bits: int = ...,
-        max_iterations: int = ...,
-        sample_rate: int = ...,
-        m: int = ...,
-        ef_construction: int = ...,
-        name: Optional[str] = ...,
-        train: bool = ...,
-        target_partition_size: Optional[int] = ...,
-    ) -> None: ...
-
-    def create_index(
-        self,
-        metric: DistanceType = "l2",
-        num_partitions: Optional[int] = None,
-        num_sub_vectors: Optional[int] = None,
+        metric="l2",
+        num_partitions=256,
+        num_sub_vectors=96,
        vector_column_name: str = VECTOR_COLUMN_NAME,
        replace: bool = True,
        accelerator: Optional[str] = None,
@@ -892,53 +824,46 @@ class Table(ABC):
        sample_rate: int = 256,
        m: int = 20,
        ef_construction: int = 300,
-        config: Optional[IndexConfigType] = None,
        name: Optional[str] = None,
        train: bool = True,
        target_partition_size: Optional[int] = None,
    ):
-        """Create an index on a column.
-
-        This method supports both the new unified API and the legacy API
-        for backwards compatibility. The new API takes the column name as the
-        first positional argument and an index configuration object via
-        ``config``; the legacy API takes the distance metric as the first
-        argument plus separate ``vector_column_name`` / ``num_partitions`` /
-        etc. parameters, and emits a ``DeprecationWarning``.
+        """Create an index on the table.

        Parameters
        ----------
-        metric : str
-            For new API: the column name to index.
-            For legacy API: the distance metric ("l2", "cosine", "dot", "hamming").
-        config : IndexConfigType, optional
-            The index configuration object. If provided, uses the new unified API.
-            Can be one of: IvfFlat, IvfPq, IvfSq, IvfRq, HnswPq, HnswSq,
-            BTree, Bitmap, LabelList, FTS.
-        replace : bool, default True
-            Whether to replace an existing index on this column.
-        wait_timeout : timedelta, optional
-            Timeout to wait for async indexing to complete.
-        name : str, optional
-            Custom name for the index.
-        train : bool, default True
-            Whether to train the index with existing data.
+        metric: str, default "l2"
+            The distance metric to use when creating the index.
+            Valid values are "l2", "cosine", "dot", or "hamming".
+            l2 is euclidean distance.
+            Hamming is available only for binary vectors.
+        num_partitions: int, default 256
+            The number of IVF partitions to use when creating the index.
+            Default is 256.
+        num_sub_vectors: int, default 96
+            The number of PQ sub-vectors to use when creating the index.
+            Default is 96.
+        vector_column_name: str, default "vector"
+            The vector column name to create the index.
+        replace: bool, default True
+            - If True, replace the existing index if it exists.

-        Examples
-        --------
-        New API (recommended):
-
-        >>> table.create_index(  # doctest: +SKIP
-        ...     "vector", config=IvfPq(distance_type="l2")
-        ... )
-        >>> table.create_index("category", config=BTree())  # doctest: +SKIP
-        >>> table.create_index("content", config=FTS())  # doctest: +SKIP
-
-        Legacy API (deprecated):
-
-        >>> table.create_index(  # doctest: +SKIP
-        ...     "l2", vector_column_name="vector"
-        ... )
+            - If False, raise an error if duplicate index exists.
+        accelerator: str, default None
+            If set, use the given accelerator to create the index.
+            Only support "cuda" for now.
+        index_cache_size : int, optional
+            The size of the index cache in number of entries. Default value is 256.
+        num_bits: int
+            The number of bits to encode sub-vectors. Only used with the IVF_PQ index.
+            Only 4 and 8 are supported.
+        wait_timeout: timedelta, optional
+            The timeout to wait if indexing is asynchronous.
+        name: str, optional
+            The name of the index. If not provided, a default name will be generated.
+        train: bool, default True
+            Whether to train the index with existing data. Vector indices always train
+            with existing data.
        """
        raise NotImplementedError

@@ -1263,7 +1188,7 @@ class Table(ABC):
        ...      .when_not_matched_insert_all() \\
        ...      .execute(new_data)
        >>> res
-        MergeResult(version=2, num_updated_rows=2, num_inserted_rows=1, num_deleted_rows=0, num_attempts=1, num_rows=3)
+        MergeResult(version=2, num_updated_rows=2, num_inserted_rows=1, num_deleted_rows=0, num_attempts=1)
        >>> # The order of new rows is non-deterministic since we use
        >>> # a hash-join as part of this operation and so we sort here
        >>> table.to_arrow().sort_by("a").to_pandas()
@@ -2282,13 +2207,9 @@ class LanceTable(Table):
        -------
        pd.DataFrame
        """
-        if blob_mode == "descriptions" or not _schema_has_blob_field(self.schema):
-            return self.to_arrow().to_pandas(**kwargs)
-
-        if (
-            blob_mode == "lazy"
-            and self._namespace_client is None
-            and get_uri_scheme(self._dataset_path) == "memory"
+        if blob_mode == "lazy" and (
+            self._namespace_client is not None
+            or get_uri_scheme(self._dataset_path) == "memory"
        ):
            return self.to_arrow().to_pandas(**kwargs)

@@ -2329,51 +2250,11 @@ class LanceTable(Table):
            dataset, allow_pyarrow_filter=False, batch_size=batch_size
        )

-    # New unified API overload
-    @overload
    def create_index(
        self,
-        column: str,
-        /,
-        *,
-        config: IndexConfigType,
-        replace: bool = ...,
-        wait_timeout: Optional[timedelta] = ...,
-        name: Optional[str] = ...,
-        train: bool = ...,
-    ) -> None: ...
-
-    # Legacy API overload (deprecated)
-    @overload
-    def create_index(
-        self,
-        metric: Literal["l2", "cosine", "dot", "hamming"] = ...,
-        num_partitions: Optional[int] = ...,
-        num_sub_vectors: Optional[int] = ...,
-        vector_column_name: str = ...,
-        replace: bool = ...,
-        accelerator: Optional[str] = ...,
-        index_cache_size: Optional[int] = ...,
-        num_bits: int = ...,
-        index_type: Literal[
-            "IVF_FLAT", "IVF_SQ", "IVF_PQ", "IVF_RQ", "IVF_HNSW_SQ", "IVF_HNSW_PQ"
-        ] = ...,
-        max_iterations: int = ...,
-        sample_rate: int = ...,
-        m: int = ...,
-        ef_construction: int = ...,
-        *,
-        wait_timeout: Optional[timedelta] = ...,
-        name: Optional[str] = ...,
-        train: bool = ...,
-        target_partition_size: Optional[int] = ...,
-    ) -> None: ...
-
-    def create_index(
-        self,
-        metric: str = "l2",
-        num_partitions: Optional[int] = None,
-        num_sub_vectors: Optional[int] = None,
+        metric: DistanceType = "l2",
+        num_partitions=None,
+        num_sub_vectors=None,
        vector_column_name: str = VECTOR_COLUMN_NAME,
        replace: bool = True,
        accelerator: Optional[str] = None,
@@ -2393,232 +2274,47 @@ class LanceTable(Table):
        m: int = 20,
        ef_construction: int = 300,
        *,
-        config: Optional[IndexConfigType] = None,
-        wait_timeout: Optional[timedelta] = None,
        name: Optional[str] = None,
        train: bool = True,
        target_partition_size: Optional[int] = None,
    ):
-        """Create an index on a column.
-
-        This method supports both the new unified API and the legacy API
-        for backwards compatibility. The new API takes the column name as the
-        first positional argument and an index configuration object via
-        ``config``; the legacy API takes the distance metric as the first
-        argument plus separate ``vector_column_name`` / ``num_partitions`` /
-        etc. parameters, and emits a ``DeprecationWarning``.
-
-        Parameters
-        ----------
-        metric : str
-            For new API: the column name to index.
-            For legacy API: the distance metric ("l2", "cosine", "dot", "hamming").
-        config : IndexConfigType, optional
-            The index configuration object. If provided, uses the new unified API.
-            Can be one of: IvfFlat, IvfPq, IvfSq, IvfRq, HnswPq, HnswSq,
-            BTree, Bitmap, LabelList, FTS.
-        replace : bool, default True
-            Whether to replace an existing index on this column.
-        wait_timeout : timedelta, optional
-            Timeout to wait for async indexing to complete.
-        name : str, optional
-            Custom name for the index.
-        train : bool, default True
-            Whether to train the index with existing data.
-
-        Examples
-        --------
-        New API (recommended):
-
-        >>> table.create_index(  # doctest: +SKIP
-        ...     "vector", config=IvfPq(distance_type="l2")
-        ... )
-        >>> table.create_index("category", config=BTree())  # doctest: +SKIP
-        >>> table.create_index("content", config=FTS())  # doctest: +SKIP
-
-        Legacy API (deprecated):
-
-        >>> table.create_index(  # doctest: +SKIP
-        ...     "l2", vector_column_name="vector"
-        ... )
-        """
-        # Detect whether this is a legacy API call
-        is_legacy = self._is_legacy_create_index_call(
-            metric,
-            config,
-            num_partitions,
-            num_sub_vectors,
-            vector_column_name,
-            accelerator,
-            index_cache_size,
-        )
-
-        if is_legacy:
-            warnings.warn(
-                "The create_index() API with metric/num_partitions parameters is "
-                "deprecated and will be removed in a future version. "
-                "Please migrate to the new unified API:\n"
-                "  # Old (deprecated):\n"
-                "  table.create_index('l2', vector_column_name='my_vector')\n"
-                "  # New (recommended):\n"
-                "  table.create_index('my_vector', config=IvfPq(distance_type='l2'))",
-                DeprecationWarning,
-                stacklevel=2,
-            )
-
-            # Legacy API: first arg is the distance metric
-            column = vector_column_name
-
-            # Build config from legacy parameters
-            config = self._build_vector_config_from_legacy_params(
-                metric=metric,
+        """Create an index on the table."""
+        if accelerator is not None:
+            # accelerator is only supported through pylance.
+            self.to_lance().create_index(
+                column=vector_column_name,
                index_type=index_type,
+                metric=metric,
                num_partitions=num_partitions,
                num_sub_vectors=num_sub_vectors,
+                replace=replace,
+                accelerator=accelerator,
+                index_cache_size=index_cache_size,
                num_bits=num_bits,
-                max_iterations=max_iterations,
-                sample_rate=sample_rate,
                m=m,
                ef_construction=ef_construction,
                target_partition_size=target_partition_size,
-                accelerator=accelerator,
            )
-
-            # Handle accelerator through pylance
-            if accelerator is not None:
-                self.to_lance().create_index(
-                    column=column,
-                    index_type=index_type,
-                    metric=metric,
-                    num_partitions=num_partitions,
-                    num_sub_vectors=num_sub_vectors,
-                    replace=replace,
-                    accelerator=accelerator,
-                    index_cache_size=index_cache_size,
-                    num_bits=num_bits,
-                    m=m,
-                    ef_construction=ef_construction,
-                    target_partition_size=target_partition_size,
-                )
-                self.checkout_latest()
-                return
-        else:
-            # New API: metric is the column name
-            column = metric
-
-            # Check if config has accelerator set and dispatch to pylance
-            if config is not None and hasattr(config, "accelerator"):
-                acc = getattr(config, "accelerator", None)
-                if acc is not None:
-                    # Dispatch to pylance for GPU acceleration
-                    index_type_map = {
-                        "IvfFlat": "IVF_FLAT",
-                        "IvfSq": "IVF_SQ",
-                        "IvfPq": "IVF_PQ",
-                        "IvfRq": "IVF_RQ",
-                        "HnswPq": "IVF_HNSW_PQ",
-                        "HnswSq": "IVF_HNSW_SQ",
-                    }
-                    cfg_type = type(config).__name__
-                    lance_index_type = index_type_map.get(cfg_type, "IVF_PQ")
-
-                    self.to_lance().create_index(
-                        column=column,
-                        index_type=lance_index_type,
-                        metric=getattr(config, "distance_type", "l2"),
-                        num_partitions=getattr(config, "num_partitions", None),
-                        num_sub_vectors=getattr(config, "num_sub_vectors", None),
-                        replace=replace,
-                        accelerator=acc,
-                        num_bits=getattr(config, "num_bits", 8),
-                        m=getattr(config, "m", 20),
-                        ef_construction=getattr(config, "ef_construction", 300),
-                        target_partition_size=getattr(
-                            config, "target_partition_size", None
-                        ),
-                    )
-                    self.checkout_latest()
-                    return
-
-        return LOOP.run(
-            self._table.create_index(
-                column,
-                replace=replace,
-                config=config,
-                wait_timeout=wait_timeout,
-                name=name,
-                train=train,
-            )
-        )
-
-    def _is_legacy_create_index_call(
-        self,
-        first_arg: str,
-        config: Optional[IndexConfigType],
-        num_partitions: Optional[int],
-        num_sub_vectors: Optional[int],
-        vector_column_name: str,
-        accelerator: Optional[str],
-        index_cache_size: Optional[int],
-    ) -> bool:
-        """Detect if this is a legacy create_index call."""
-        # If config is provided, it's definitely the new API
-        if config is not None:
-            return False
-
-        # If old-style parameters were explicitly set, it's legacy
-        if any(
-            x is not None
-            for x in (num_partitions, num_sub_vectors, accelerator, index_cache_size)
-        ):
-            return True
-
-        # If vector_column_name differs from default, it's legacy
-        if vector_column_name != VECTOR_COLUMN_NAME:
-            return True
-
-        # If first arg is a known metric, assume legacy
-        if first_arg.lower() in KNOWN_METRICS:
-            return True
-
-        # Otherwise assume new API
-        return False
-
-    def _build_vector_config_from_legacy_params(
-        self,
-        metric: str,
-        index_type: str,
-        num_partitions: Optional[int],
-        num_sub_vectors: Optional[int],
-        num_bits: int,
-        max_iterations: int,
-        sample_rate: int,
-        m: int,
-        ef_construction: int,
-        target_partition_size: Optional[int],
-        accelerator: Optional[str],
-    ) -> IndexConfigType:
-        """Build an index config object from legacy parameters."""
-        if index_type == "IVF_FLAT":
-            return IvfFlat(
+            self.checkout_latest()
+            return
+        elif index_type == "IVF_FLAT":
+            config = IvfFlat(
                distance_type=metric,
                num_partitions=num_partitions,
                max_iterations=max_iterations,
                sample_rate=sample_rate,
                target_partition_size=target_partition_size,
-                accelerator=accelerator,
            )
        elif index_type == "IVF_SQ":
-            return IvfSq(
+            config = IvfSq(
                distance_type=metric,
                num_partitions=num_partitions,
                max_iterations=max_iterations,
                sample_rate=sample_rate,
                target_partition_size=target_partition_size,
-                accelerator=accelerator,
            )
        elif index_type == "IVF_PQ":
-            return IvfPq(
+            config = IvfPq(
                distance_type=metric,
                num_partitions=num_partitions,
                num_sub_vectors=num_sub_vectors,
@@ -2626,20 +2322,18 @@ class LanceTable(Table):
                max_iterations=max_iterations,
                sample_rate=sample_rate,
                target_partition_size=target_partition_size,
-                accelerator=accelerator,
            )
        elif index_type == "IVF_RQ":
-            return IvfRq(
+            config = IvfRq(
                distance_type=metric,
                num_partitions=num_partitions,
                num_bits=num_bits,
                max_iterations=max_iterations,
                sample_rate=sample_rate,
                target_partition_size=target_partition_size,
-                accelerator=accelerator,
            )
        elif index_type == "IVF_HNSW_PQ":
-            return HnswPq(
+            config = HnswPq(
                distance_type=metric,
                num_partitions=num_partitions,
                num_sub_vectors=num_sub_vectors,
@@ -2649,10 +2343,9 @@ class LanceTable(Table):
                m=m,
                ef_construction=ef_construction,
                target_partition_size=target_partition_size,
-                accelerator=accelerator,
            )
        elif index_type == "IVF_HNSW_SQ":
-            return HnswSq(
+            config = HnswSq(
                distance_type=metric,
                num_partitions=num_partitions,
                max_iterations=max_iterations,
@@ -2660,10 +2353,9 @@ class LanceTable(Table):
                m=m,
                ef_construction=ef_construction,
                target_partition_size=target_partition_size,
-                accelerator=accelerator,
            )
        elif index_type == "IVF_HNSW_FLAT":
-            return HnswFlat(
+            config = HnswFlat(
                distance_type=metric,
                num_partitions=num_partitions,
                max_iterations=max_iterations,
@@ -2675,6 +2367,16 @@ class LanceTable(Table):
        else:
            raise ValueError(f"Unknown index type {index_type}")

+        return LOOP.run(
+            self._table.create_index(
+                vector_column_name,
+                replace=replace,
+                config=config,
+                name=name,
+                train=train,
+            )
+        )
+
    def drop_index(self, name: str) -> None:
        """
        Drops an index from the table
@@ -2774,11 +2476,6 @@ class LanceTable(Table):
        """
        return LOOP.run(self._table.latest_storage_options())

-    @deprecation.deprecated(
-        deprecated_in="0.25.0",
-        current_version=__version__,
-        details="Use create_index() with config=BTree()/Bitmap()/LabelList() instead.",
-    )
    def create_scalar_index(
        self,
        column: str,
@@ -2787,12 +2484,6 @@ class LanceTable(Table):
        index_type: ScalarIndexType = "BTREE",
        name: Optional[str] = None,
    ):
-        """Create a scalar index on a column.
-
-        .. deprecated:: 0.25.0
-            Use :meth:`create_index` with a BTree, Bitmap, or LabelList config instead.
-            Example: ``table.create_index("column", config=BTree())``
-        """
        if index_type == "BTREE":
            config = BTree()
        elif index_type == "BITMAP":
@@ -2805,11 +2496,6 @@ class LanceTable(Table):
            self._table.create_index(column, replace=replace, config=config, name=name)
        )

-    @deprecation.deprecated(
-        deprecated_in="0.25.0",
-        current_version=__version__,
-        details="Use create_index() with config=FTS() instead.",
-    )
    def create_fts_index(
        self,
        field_names: Union[str, List[str]],
@@ -2833,12 +2519,6 @@ class LanceTable(Table):
        prefix_only: bool = False,
        name: Optional[str] = None,
    ):
-        """Create a full-text search index on a column.
-
-        .. deprecated:: 0.25.0
-            Use :meth:`create_index` with an FTS config instead.
-            Example: ``table.create_index("text_column", config=FTS())``
-        """
        self._ensure_no_legacy_fts_index()

        if use_tantivy:
@@ -3617,11 +3297,6 @@ class LanceTable(Table):
        [`AsyncTable.unset_lsm_write_spec`][lancedb.AsyncTable.unset_lsm_write_spec]."""
        return LOOP.run(self._table.unset_lsm_write_spec())

-    def close_lsm_writers(self) -> None:
-        """Close cached MemWAL shard writers. See
-        [`AsyncTable.close_lsm_writers`][lancedb.AsyncTable.close_lsm_writers]."""
-        return LOOP.run(self._table.close_lsm_writers())
-
    def uses_v2_manifest_paths(self) -> bool:
        """
        Check if the table is using the new v2 manifest paths.
@@ -4230,16 +3905,6 @@ class AsyncTable:
        """
        await self._inner.unset_lsm_write_spec()

-    async def close_lsm_writers(self) -> None:
-        """Drain and close any cached MemWAL shard writers for this table.
-
-        When an LSM write spec is installed, `merge_insert` opens MemWAL shard
-        writers and caches them for reuse across calls. This closes them,
-        flushing pending data; writers reopen lazily on the next
-        `merge_insert`. It is a no-op when no writers are cached.
-        """
-        await self._inner.close_lsm_writers()
-
    @property
    def name(self) -> str:
        """The name of the table."""
@@ -4296,7 +3961,7 @@ class AsyncTable:
        can be executed with methods like [to_arrow][lancedb.query.AsyncQuery.to_arrow],
        [to_pandas][lancedb.query.AsyncQuery.to_pandas] and more.
        """
-        return AsyncQuery(self._inner.query(), self)
+        return AsyncQuery(self._inner.query())

    async def _to_lance(self, **kwargs) -> lance.LanceDataset:
        try:
@@ -4328,12 +3993,7 @@ class AsyncTable:
        -------
        pd.DataFrame
        """
-        if blob_mode == "descriptions" or not _schema_has_blob_field(
-            await self.schema()
-        ):
-            return (await self.to_arrow()).to_pandas(**kwargs)
-
-        if blob_mode == "lazy" and get_uri_scheme(await self.uri()) == "memory":
+        if blob_mode == "lazy":
            return (await self.to_arrow()).to_pandas(**kwargs)
        return (await self._to_lance()).to_pandas(blob_mode=blob_mode, **kwargs)

@@ -4695,7 +4355,7 @@ class AsyncTable:
        ...      .when_not_matched_insert_all() \\
        ...      .execute(new_data)
        >>> res
-        MergeResult(version=2, num_updated_rows=2, num_inserted_rows=1, num_deleted_rows=0, num_attempts=1, num_rows=3)
+        MergeResult(version=2, num_updated_rows=2, num_inserted_rows=1, num_deleted_rows=0, num_attempts=1)
        >>> # The order of new rows is non-deterministic since we use
        >>> # a hash-join as part of this operation and so we sort here
        >>> table.to_arrow().sort_by("a").to_pandas()
@@ -5075,8 +4735,6 @@ class AsyncTable:
                when_not_matched_by_source_condition=merge._when_not_matched_by_source_condition,
                timeout=merge._timeout,
                use_index=merge._use_index,
-                use_lsm_write=merge._use_lsm_write,
-                validate_single_shard=merge._validate_single_shard,
            ),
        )

@@ -5370,7 +5028,7 @@ class AsyncTable:
        pa.RecordBatch
            A record batch containing the rows at the given offsets.
        """
-        return AsyncTakeQuery(self._inner.take_offsets(offsets), self)
+        return AsyncTakeQuery(self._inner.take_offsets(offsets))

    def take_row_ids(self, row_ids: list[int]) -> AsyncTakeQuery:
        """
@@ -5399,7 +5057,7 @@ class AsyncTable:
        AsyncTakeQuery
            A query object that can be executed to get the rows.
        """
-        return AsyncTakeQuery(self._inner.take_row_ids(row_ids), self)
+        return AsyncTakeQuery(self._inner.take_row_ids(row_ids))

    @property
    def tags(self) -> AsyncTags:
--- a/python/python/tests/docs/test_merge_insert.py
+++ b/python/python/tests/docs/test_merge_insert.py
@@ -57,7 +57,7 @@ async def test_upsert_async(mem_db_async):
    await table.count_rows()  # 3
    res
    # MergeResult(version=2, num_updated_rows=1,
-    # num_inserted_rows=1, num_deleted_rows=0, num_rows=2)
+    # num_inserted_rows=1, num_deleted_rows=0)
    # --8<-- [end:upsert_basic_async]
    assert await table.count_rows() == 3
    assert res.version == 2
@@ -86,7 +86,7 @@ def test_insert_if_not_exists(mem_db):
    table.count_rows()  # 3
    res
    # MergeResult(version=2, num_updated_rows=0,
-    # num_inserted_rows=1, num_deleted_rows=0, num_rows=1)
+    # num_inserted_rows=1, num_deleted_rows=0)
    # --8<-- [end:insert_if_not_exists]
    assert table.count_rows() == 3
    assert res.version == 2
@@ -116,7 +116,7 @@ async def test_insert_if_not_exists_async(mem_db_async):
    await table.count_rows()  # 3
    res
    # MergeResult(version=2, num_updated_rows=0,
-    # num_inserted_rows=1, num_deleted_rows=0, num_rows=1)
+    # num_inserted_rows=1, num_deleted_rows=0)
    # --8<-- [end:insert_if_not_exists]
    assert await table.count_rows() == 3
    assert res.version == 2
@@ -150,7 +150,7 @@ def test_replace_range(mem_db):
    table.count_rows("doc_id = 1")  # 1
    res
    # MergeResult(version=2, num_updated_rows=1,
-    # num_inserted_rows=0, num_deleted_rows=1, num_rows=1)
+    # num_inserted_rows=0, num_deleted_rows=1)
    # --8<-- [end:insert_if_not_exists]
    assert table.count_rows("doc_id = 1") == 1
    assert res.version == 2
@@ -185,7 +185,7 @@ async def test_replace_range_async(mem_db_async):
    await table.count_rows("doc_id = 1")  # 1
    res
    # MergeResult(version=2, num_updated_rows=1,
-    # num_inserted_rows=0, num_deleted_rows=1, num_rows=1)
+    # num_inserted_rows=0, num_deleted_rows=1)
    # --8<-- [end:insert_if_not_exists]
    assert await table.count_rows("doc_id = 1") == 1
    assert res.version == 2
--- a/python/python/tests/test_fts.py
+++ b/python/python/tests/test_fts.py
@@ -215,12 +215,11 @@ def test_reject_legacy_tantivy_index(table):

@pytest.mark.parametrize("with_position", [True, False])
 def test_create_inverted_index(table, with_position):
-    with pytest.warns(DeprecationWarning, match="create_fts_index"):
-        table.create_fts_index(
-            "text",
-            with_position=with_position,
-            name="custom_fts_index",
-        )
+    table.create_fts_index(
+        "text",
+        with_position=with_position,
+        name="custom_fts_index",
+    )
    indices = table.list_indices()
    fts_indices = [i for i in indices if i.index_type == "FTS"]
    assert any(i.name == "custom_fts_index" for i in fts_indices)
--- a/python/python/tests/test_index.py
+++ b/python/python/tests/test_index.py
@@ -162,13 +162,12 @@ async def test_create_bitmap_index(some_table: AsyncTable):
    await some_table.create_index("data", config=Bitmap())
    indices = await some_table.list_indices()
    assert len(indices) == 3
-    # list_indices returns indices in alphabetical order by name
    assert indices[0].index_type == "Bitmap"
-    assert indices[0].columns == ["data"]
+    assert indices[0].columns == ["id"]
    assert indices[1].index_type == "Bitmap"
-    assert indices[1].columns == ["id"]
+    assert indices[1].columns == ["is_active"]
    assert indices[2].index_type == "Bitmap"
-    assert indices[2].columns == ["is_active"]
+    assert indices[2].columns == ["data"]

    index_name = indices[0].name
    stats = await some_table.index_stats(index_name)
--- a/python/python/tests/test_merge_insert_lsm.py
+++ b/python/python/tests/test_merge_insert_lsm.py
@@ -1,196 +0,0 @@
-# SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright The LanceDB Authors
-
-"""Tests for the MemWAL LSM ``merge_insert`` dispatch."""
-
-from datetime import timedelta
-
-import lancedb
-import pyarrow as pa
-import pytest
-from lancedb._lancedb import LsmWriteSpec
-
-SCHEMA = pa.schema(
-    [
-        pa.field("id", pa.int64(), nullable=False),
-        pa.field("value", pa.int64(), nullable=False),
-    ]
-)
-
-REGION_SCHEMA = pa.schema(
-    [
-        pa.field("id", pa.int64(), nullable=False),
-        pa.field("region", pa.utf8(), nullable=False),
-    ]
-)
-
-
-def _reader(ids):
-    batch = pa.RecordBatch.from_arrays(
-        [
-            pa.array(ids, type=pa.int64()),
-            pa.array(list(range(len(ids))), type=pa.int64()),
-        ],
-        schema=SCHEMA,
-    )
-    return pa.RecordBatchReader.from_batches(SCHEMA, [batch])
-
-
-def _region_reader(rows):
-    batch = pa.RecordBatch.from_arrays(
-        [
-            pa.array([row[0] for row in rows], type=pa.int64()),
-            pa.array([row[1] for row in rows], type=pa.utf8()),
-        ],
-        schema=REGION_SCHEMA,
-    )
-    return pa.RecordBatchReader.from_batches(REGION_SCHEMA, [batch])
-
-
-def _bucket_table(tmp_path):
-    """A table with ``id`` as the primary key and a single-bucket LSM spec."""
-    db = lancedb.connect(tmp_path, read_consistency_interval=timedelta(seconds=0))
-    table = db.create_table("t", _reader([1, 2, 3]))
-    table.set_unenforced_primary_key("id")
-    # num_buckets = 1: every row routes to the single bucket.
-    table.set_lsm_write_spec(LsmWriteSpec.bucket("id", 1))
-    return table
-
-
-def test_lsm_merge_insert_bucket(tmp_path):
-    table = _bucket_table(tmp_path)
-    # Empty `on` defaults to the primary key.
-    result = (
-        table.merge_insert([])
-        .when_matched_update_all()
-        .when_not_matched_insert_all()
-        .execute(_reader([3, 4, 5]))
-    )
-    # LSM path: rows go to the MemWAL, so only num_rows is populated.
-    assert result.num_rows == 3
-    assert result.version == 0
-    assert result.num_inserted_rows == 0
-    assert result.num_updated_rows == 0
-
-
-def test_lsm_merge_insert_unsharded(tmp_path):
-    db = lancedb.connect(tmp_path, read_consistency_interval=timedelta(seconds=0))
-    table = db.create_table("t", _reader([1, 2, 3]))
-    table.set_unenforced_primary_key("id")
-    table.set_lsm_write_spec(LsmWriteSpec.unsharded())
-    result = (
-        table.merge_insert("id")
-        .when_matched_update_all()
-        .when_not_matched_insert_all()
-        .execute(_reader([10, 11, 12, 13]))
-    )
-    assert result.num_rows == 4
-
-
-def test_lsm_merge_insert_identity(tmp_path):
-    db = lancedb.connect(tmp_path, read_consistency_interval=timedelta(seconds=0))
-    table = db.create_table("t", _region_reader([(1, "us"), (2, "us")]))
-    table.set_unenforced_primary_key("id")
-    table.set_lsm_write_spec(LsmWriteSpec.identity("region"))
-    # All rows share one identity value, so they route to one shard.
-    result = (
-        table.merge_insert([])
-        .when_matched_update_all()
-        .when_not_matched_insert_all()
-        .execute(_region_reader([(3, "us"), (4, "us")]))
-    )
-    assert result.num_rows == 2
-
-
-def test_lsm_merge_insert_use_lsm_write_false(tmp_path):
-    table = _bucket_table(tmp_path)  # rows id = 1, 2, 3
-    # use_lsm_write(False) opts out: the standard path runs and commits.
-    result = (
-        table.merge_insert("id")
-        .when_not_matched_insert_all()
-        .use_lsm_write(False)
-        .execute(_reader([3, 4, 5]))
-    )
-    assert result.num_inserted_rows == 2
-    assert table.count_rows() == 5
-
-
-def test_lsm_merge_insert_validate_single_shard_off(tmp_path):
-    table = _bucket_table(tmp_path)
-    result = (
-        table.merge_insert([])
-        .when_matched_update_all()
-        .when_not_matched_insert_all()
-        .validate_single_shard(False)
-        .execute(_reader([6, 7, 8]))
-    )
-    assert result.num_rows == 3
-
-
-def test_lsm_merge_insert_use_lsm_write_true_requires_spec(tmp_path):
-    # A table with a primary key but no LSM write spec installed.
-    db = lancedb.connect(tmp_path, read_consistency_interval=timedelta(seconds=0))
-    table = db.create_table("t", _reader([1, 2, 3]))
-    table.set_unenforced_primary_key("id")
-    with pytest.raises(Exception, match="use_lsm_write"):
-        (
-            table.merge_insert("id")
-            .when_matched_update_all()
-            .when_not_matched_insert_all()
-            .use_lsm_write(True)
-            .execute(_reader([4]))
-        )
-
-
-def test_lsm_merge_insert_rejects_on_not_primary_key(tmp_path):
-    table = _bucket_table(tmp_path)
-    with pytest.raises(Exception, match="primary key"):
-        (
-            table.merge_insert("value")
-            .when_matched_update_all()
-            .when_not_matched_insert_all()
-            .execute(_reader([1]))
-        )
-
-
-def test_lsm_merge_insert_rejects_non_upsert(tmp_path):
-    table = _bucket_table(tmp_path)
-    # Insert-only (no when_matched_update_all) is not the upsert shape.
-    with pytest.raises(Exception, match="upsert"):
-        table.merge_insert([]).when_not_matched_insert_all().execute(_reader([4]))
-
-
-def test_lsm_close_writers(tmp_path):
-    table = _bucket_table(tmp_path)
-    (
-        table.merge_insert([])
-        .when_matched_update_all()
-        .when_not_matched_insert_all()
-        .execute(_reader([7, 8]))
-    )
-    table.close_lsm_writers()
-    # The writer reopens lazily on the next merge_insert.
-    result = (
-        table.merge_insert([])
-        .when_matched_update_all()
-        .when_not_matched_insert_all()
-        .execute(_reader([9]))
-    )
-    assert result.num_rows == 1
-
-
-@pytest.mark.asyncio
-async def test_async_lsm_merge_insert(tmp_path):
-    db = await lancedb.connect_async(
-        tmp_path, read_consistency_interval=timedelta(seconds=0)
-    )
-    table = await db.create_table("t", _reader([1, 2, 3]))
-    await table.set_unenforced_primary_key("id")
-    await table.set_lsm_write_spec(LsmWriteSpec.bucket("id", 1))
-
-    builder = (
-        table.merge_insert([]).when_matched_update_all().when_not_matched_insert_all()
-    )
-    result = await builder.execute(_reader([3, 4, 5]))
-    assert result.num_rows == 3
-    await table.close_lsm_writers()
--- a/python/python/tests/test_namespace.py
+++ b/python/python/tests/test_namespace.py
@@ -76,35 +76,6 @@ class TestNamespaceConnection:
        assert len(result) == 0
        assert list(result.columns) == ["id", "vector", "text"]

-    def test_table_to_pandas_blob_lazy_through_namespace(self):
-        """Namespace-backed tables should use Lance blob-aware pandas conversion."""
-        pytest.importorskip("lance")
-        db = lancedb.connect_namespace("dir", {"root": self.temp_dir})
-        db.create_namespace(["test_ns"])
-        data = pa.table(
-            {
-                "id": pa.array([1, 2], pa.int64()),
-                "blob": pa.array([b"hello", b"world"], pa.large_binary()),
-            },
-            schema=pa.schema(
-                [
-                    pa.field("id", pa.int64()),
-                    pa.field(
-                        "blob",
-                        pa.large_binary(),
-                        metadata={"lance-encoding:blob": "true"},
-                    ),
-                ]
-            ),
-        )
-
-        table = db.create_table("blob_table", data, namespace_path=["test_ns"])
-        df = table.to_pandas(blob_mode="lazy").sort_values("id")
-
-        blob = df["blob"].iloc[0]
-        assert hasattr(blob, "readall")
-        assert blob.readall() == b"hello"
-
    def test_open_table_through_namespace(self):
        """Test opening an existing table through namespace."""
        db = lancedb.connect_namespace("dir", {"root": self.temp_dir})
--- a/python/python/tests/test_query.py
+++ b/python/python/tests/test_query.py
@@ -39,35 +39,6 @@ from utils import exception_output
 from importlib.util import find_spec


-def _blob_query_data():
-    return pa.table(
-        {
-            "id": pa.array([1, 2, 3, 4], pa.int64()),
-            "tag": pa.array(["drop", "keep", "keep", "keep"], pa.utf8()),
-            "vector": pa.array(
-                [[1.0, 0.0], [2.0, 0.0], [3.0, 0.0], [4.0, 0.0]],
-                type=pa.list_(pa.float32(), list_size=2),
-            ),
-            "blob": pa.array([b"one", b"two", b"three", b"four"], pa.large_binary()),
-        },
-        schema=pa.schema(
-            [
-                pa.field("id", pa.int64()),
-                pa.field("tag", pa.utf8()),
-                pa.field("vector", pa.list_(pa.float32(), list_size=2)),
-                pa.field(
-                    "blob", pa.large_binary(), metadata={"lance-encoding:blob": "true"}
-                ),
-            ]
-        ),
-    )
-
-
-def _assert_lazy_blob(value, expected: bytes):
-    assert hasattr(value, "readall")
-    assert value.readall() == expected
-
-
@pytest.fixture(scope="module")
 def table(tmpdir_factory) -> lancedb.table.Table:
    tmp_path = str(tmpdir_factory.mktemp("data"))
@@ -210,97 +181,6 @@ async def test_query_to_pandas_kwargs(table, table_async):
    assert async_df["id"].tolist() == [1, 2]


-@pytest.mark.parametrize("blob_mode", ["lazy", "bytes", "descriptions"])
-def test_plain_scan_query_to_pandas_blob_modes(tmp_db, blob_mode):
-    pytest.importorskip("lance")
-    table = tmp_db.create_table(
-        f"test_query_to_pandas_blob_{blob_mode}", _blob_query_data()
-    )
-
-    df = (
-        table.search()
-        .select(["id", "blob"])
-        .where("id = 1")
-        .to_pandas(blob_mode=blob_mode)
-    )
-
-    assert df["id"].tolist() == [1]
-    if blob_mode == "lazy":
-        _assert_lazy_blob(df["blob"].iloc[0], b"one")
-    elif blob_mode == "bytes":
-        assert df["blob"].tolist() == [b"one"]
-    else:
-        first = df["blob"].iloc[0]
-        assert first != b"one"
-        assert not hasattr(first, "readall")
-
-
-def test_plain_scan_query_to_pandas_blob_projection(tmp_db):
-    pytest.importorskip("lance")
-    table = tmp_db.create_table(
-        "test_query_to_pandas_blob_projection", _blob_query_data()
-    )
-
-    df = (
-        table.search()
-        .where("id >= 2")
-        .select({"id_alias": "id", "payload": "blob", "double_id": "id * 2"})
-        .limit(2)
-        .offset(1)
-        .to_pandas(blob_mode="bytes")
-    )
-
-    assert df["id_alias"].tolist() == [3, 4]
-    assert df["payload"].tolist() == [b"three", b"four"]
-    assert df["double_id"].tolist() == [6, 8]
-
-
-@pytest.mark.asyncio
-async def test_async_plain_scan_query_to_pandas_blob_projection(tmp_db_async):
-    pytest.importorskip("lance")
-    table = await tmp_db_async.create_table(
-        "test_async_query_to_pandas_blob_projection", _blob_query_data()
-    )
-
-    lazy_df = await (
-        table.query().where("id = 1").select(["id", "blob"]).to_pandas(blob_mode="lazy")
-    )
-    assert lazy_df["id"].tolist() == [1]
-    _assert_lazy_blob(lazy_df["blob"].iloc[0], b"one")
-
-    bytes_df = await (
-        table.query()
-        .where("id >= 2")
-        .select({"id_alias": "id", "payload": "blob", "double_id": "id * 2"})
-        .limit(2)
-        .offset(1)
-        .to_pandas(blob_mode="bytes")
-    )
-    assert bytes_df["id_alias"].tolist() == [3, 4]
-    assert bytes_df["payload"].tolist() == [b"three", b"four"]
-    assert bytes_df["double_id"].tolist() == [6, 8]
-
-    desc_df = await (
-        table.query()
-        .where("id = 1")
-        .select(["blob"])
-        .to_pandas(blob_mode="descriptions")
-    )
-    first = desc_df["blob"].iloc[0]
-    assert first != b"one"
-    assert not hasattr(first, "readall")
-
-
-def test_vector_query_to_pandas_blob_mode_requires_native_path(tmp_db):
-    pytest.importorskip("lance")
-    table = tmp_db.create_table("test_vector_query_blob_mode", _blob_query_data())
-
-    with pytest.raises(RuntimeError, match="Lance native pandas conversion"):
-        table.search([1.0, 0.0]).select(["blob", "vector"]).limit(1).to_pandas(
-            blob_mode="lazy"
-        )
-
-
 def test_order_by_plain_query(mem_db):
    table = mem_db.create_table(
        "test_order_by",
--- a/python/python/tests/test_remote_db.py
+++ b/python/python/tests/test_remote_db.py
@@ -1,13 +1,12 @@
 # SPDX-License-Identifier: Apache-2.0
 # SPDX-FileCopyrightText: Copyright The LanceDB Authors
+import re
 from concurrent.futures import ThreadPoolExecutor
 import contextlib
 from datetime import timedelta
 import http.server
 import json
 import multiprocessing as mp
-import pickle
-import re
 import sys
 import threading
 import time
@@ -172,155 +171,6 @@ def test_table_len_sync():
        assert len(table) == 1


-def test_remote_connection_serializes():
-    def handler(request):
-        request.send_response(200)
-        request.send_header("Content-Type", "application/json")
-        request.end_headers()
-        request.wfile.write(b'{"tables": []}')
-
-    with mock_lancedb_connection(handler) as db:
-        serialized = json.loads(db.serialize())
-        assert isinstance(serialized["client_config"], dict)
-        restored = lancedb.deserialize_conn(db.serialize())
-        assert restored.table_names() == []
-
-
-def test_remote_table_is_picklable():
-    def handler(request):
-        request.close_connection = True
-        if request.path == "/v1/table/test/describe/":
-            request.send_response(200)
-            request.send_header("Content-Type", "application/json")
-            request.end_headers()
-            payload = json.dumps(
-                {
-                    "version": 1,
-                    "schema": {
-                        "fields": [
-                            {"name": "id", "type": {"type": "int64"}, "nullable": False}
-                        ]
-                    },
-                }
-            )
-            request.wfile.write(payload.encode())
-        elif request.path == "/v1/table/test/count_rows/":
-            request.send_response(200)
-            request.send_header("Content-Type", "application/json")
-            request.end_headers()
-            request.wfile.write(b"3")
-        else:
-            request.send_response(404)
-            request.end_headers()
-
-    with mock_lancedb_connection(handler) as db:
-        table = db.open_table("test")
-        restored = pickle.loads(pickle.dumps(table))
-        assert restored.count_rows() == 3
-
-
-def test_remote_table_open_does_not_require_picklable_client_config():
-    from lancedb.remote import HeaderProvider
-
-    class LocalHeaderProvider(HeaderProvider):
-        def get_headers(self):
-            return {"X-Test-Header": "present"}
-
-    def handler(request):
-        request.close_connection = True
-        assert request.headers.get("X-Test-Header") == "present"
-        if request.path == "/v1/table/test/describe/":
-            request.send_response(200)
-            request.send_header("Content-Type", "application/json")
-            request.end_headers()
-            request.wfile.write(b'{"version": 1, "schema": {"fields": []}}')
-        elif request.path == "/v1/table/test/count_rows/":
-            request.send_response(200)
-            request.send_header("Content-Type", "application/json")
-            request.end_headers()
-            request.wfile.write(b"3")
-        else:
-            request.send_response(404)
-            request.end_headers()
-
-    with http.server.HTTPServer(
-        ("localhost", 0), make_mock_http_handler(handler)
-    ) as server:
-        port = server.server_address[1]
-        handle = threading.Thread(target=server.serve_forever)
-        handle.start()
-        try:
-            db = lancedb.connect(
-                "db://dev",
-                api_key="fake",
-                host_override=f"http://localhost:{port}",
-                client_config={
-                    "retry_config": {"retries": 0},
-                    "timeout_config": {"connect_timeout": 2, "read_timeout": 2},
-                    "header_provider": LocalHeaderProvider(),
-                },
-            )
-            table = db.open_table("test")
-            assert table.count_rows() == 3
-            with pytest.raises(ValueError, match="header_provider"):
-                pickle.dumps(table)
-        finally:
-            server.shutdown()
-            handle.join()
-
-
-def test_remote_permutation_is_picklable():
-    from lancedb.permutation import Permutation
-
-    rows = list(range(10))
-
-    def handler(request):
-        request.close_connection = True
-        if request.path == "/v1/table/test/describe/":
-            request.send_response(200)
-            request.send_header("Content-Type", "application/json")
-            request.end_headers()
-            payload = json.dumps(
-                {
-                    "version": 1,
-                    "schema": {
-                        "fields": [
-                            {"name": "a", "type": {"type": "int64"}, "nullable": False}
-                        ]
-                    },
-                }
-            )
-            request.wfile.write(payload.encode())
-        elif request.path == "/v1/table/test/count_rows/":
-            request.send_response(200)
-            request.send_header("Content-Type", "application/json")
-            request.end_headers()
-            request.wfile.write(str(len(rows)).encode())
-        elif request.path == "/v1/table/test/query/":
-            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(",")]
-            else:
-                offsets = rows
-            table = pa.table({"a": [rows[offset] for offset in offsets]})
-
-            request.send_response(200)
-            request.send_header("Content-Type", "application/vnd.apache.arrow.file")
-            request.end_headers()
-            with pa.ipc.new_file(request.wfile, schema=table.schema) as writer:
-                writer.write_table(table)
-        else:
-            request.send_response(404)
-            request.end_headers()
-
-    with mock_lancedb_connection(handler) as db:
-        permutation = Permutation.identity(db.open_table("test"))
-        restored = pickle.loads(pickle.dumps(permutation))
-        assert restored.__getitems__([0, 2, 4]) == [{"a": 0}, {"a": 2}, {"a": 4}]
-
-
 def test_create_table_exist_ok():
    def handler(request):
        if request.path == "/v1/table/test/create/?mode=exist_ok":
@@ -586,25 +436,22 @@ def test_table_create_indices():
        # This is a smoke-test.
        table = db.create_table("test", [{"id": 1}])

-        # Test create_scalar_index with custom name (legacy method)
-        with pytest.warns(DeprecationWarning, match="create_scalar_index"):
-            table.create_scalar_index(
-                "id", wait_timeout=timedelta(seconds=2), name="custom_scalar_idx"
-            )
+        # Test create_scalar_index with custom name
+        table.create_scalar_index(
+            "id", wait_timeout=timedelta(seconds=2), name="custom_scalar_idx"
+        )

-        # Test create_fts_index with custom name (legacy method)
-        with pytest.warns(DeprecationWarning, match="create_fts_index"):
-            table.create_fts_index(
-                "text", wait_timeout=timedelta(seconds=2), name="custom_fts_idx"
-            )
+        # Test create_fts_index with custom name
+        table.create_fts_index(
+            "text", wait_timeout=timedelta(seconds=2), name="custom_fts_idx"
+        )

-        # Test create_index with custom name (legacy form: vector_column_name kwarg)
-        with pytest.warns(DeprecationWarning, match="create_index"):
-            table.create_index(
-                vector_column_name="vector",
-                wait_timeout=timedelta(seconds=10),
-                name="custom_vector_idx",
-            )
+        # Test create_index with custom name
+        table.create_index(
+            vector_column_name="vector",
+            wait_timeout=timedelta(seconds=10),
+            name="custom_vector_idx",
+        )

        # Validate that the name parameter was passed correctly in requests
        assert len(received_requests) == 3
@@ -633,98 +480,6 @@ def test_table_create_indices():
        table.drop_index("custom_fts_idx")


-def test_remote_create_index_new_api():
-    received_requests = []
-
-    def handler(request):
-        if request.path == "/v1/table/test/create_index/":
-            content_len = int(request.headers.get("Content-Length", 0))
-            body = request.rfile.read(content_len) if content_len > 0 else b""
-            received_requests.append(json.loads(body) if body else {})
-            request.send_response(200)
-            request.end_headers()
-        elif request.path == "/v1/table/test/create/?mode=create":
-            request.send_response(200)
-            request.send_header("Content-Type", "application/json")
-            request.end_headers()
-            request.wfile.write(b"{}")
-        elif request.path == "/v1/table/test/describe/":
-            request.send_response(200)
-            request.send_header("Content-Type", "application/json")
-            request.end_headers()
-            request.wfile.write(
-                json.dumps(
-                    dict(
-                        version=1,
-                        schema=dict(
-                            fields=[
-                                dict(name="id", type={"type": "int64"}, nullable=False),
-                                dict(
-                                    name="category",
-                                    type={"type": "string"},
-                                    nullable=False,
-                                ),
-                                dict(
-                                    name="text", type={"type": "string"}, nullable=False
-                                ),
-                                dict(
-                                    name="vector",
-                                    type={
-                                        "type": "fixed_size_list",
-                                        "fields": [
-                                            dict(
-                                                name="item",
-                                                type={"type": "float"},
-                                                nullable=True,
-                                            )
-                                        ],
-                                        "length": 2,
-                                    },
-                                    nullable=False,
-                                ),
-                            ]
-                        ),
-                    )
-                ).encode()
-            )
-        else:
-            request.send_response(404)
-            request.end_headers()
-
-    from lancedb.index import BTree, FTS, IvfPq, IvfRq
-
-    with mock_lancedb_connection(handler) as db:
-        table = db.create_table("test", [{"id": 1}])
-
-        # New API: column-first, config= kwarg. Should NOT emit DeprecationWarning.
-        import warnings as _warnings
-
-        with _warnings.catch_warnings():
-            _warnings.simplefilter("error", DeprecationWarning)
-            table.create_index("vector", config=IvfPq(distance_type="l2"))
-            table.create_index("category", config=BTree())
-            table.create_index("text", config=FTS())
-            # IvfRq via new API
-            table.create_index("vector", config=IvfRq(distance_type="l2"))
-
-        # Legacy index_type="IVF_RQ" routes to IvfRq config under the hood.
-        with pytest.warns(DeprecationWarning, match="create_index"):
-            table.create_index(
-                vector_column_name="vector",
-                index_type="IVF_RQ",
-                num_partitions=8,
-            )
-
-        assert len(received_requests) == 5
-        assert [req["column"] for req in received_requests] == [
-            "vector",
-            "category",
-            "text",
-            "vector",
-            "vector",
-        ]
-
-
 def test_table_wait_for_index_timeout():
    def handler(request):
        index_stats = dict(
@@ -1550,10 +1305,6 @@ def _remote_fork_child(port: int, queue) -> None:
    queue.put(db.table_names())


-def _remote_table_fork_child(table, queue) -> None:
-    queue.put(table.count_rows())
-
-
@pytest.mark.skipif(
    sys.platform != "linux",
    reason=(
@@ -1616,65 +1367,3 @@ def test_remote_connection_after_fork():
    finally:
        server.shutdown()
        server_thread.join()
-
-
-@pytest.mark.skipif(
-    sys.platform != "linux",
-    reason=(
-        "fork() is unavailable on Windows and unsafe on macOS "
-        "(Apple frameworks/TLS are not fork-safe)"
-    ),
-)
-def test_inherited_remote_table_reopens_after_fork():
-    def handler(request):
-        if request.path == "/v1/table/test/describe/":
-            request.send_response(200)
-            request.send_header("Content-Type", "application/json")
-            request.end_headers()
-            request.wfile.write(b'{"version": 1, "schema": {"fields": []}}')
-        elif request.path == "/v1/table/test/count_rows/":
-            request.send_response(200)
-            request.send_header("Content-Type", "application/json")
-            request.end_headers()
-            request.wfile.write(b"7")
-        else:
-            request.send_response(404)
-            request.end_headers()
-
-    server = http.server.HTTPServer(("localhost", 0), make_mock_http_handler(handler))
-    port = server.server_address[1]
-    server_thread = threading.Thread(target=server.serve_forever)
-    server_thread.start()
-    try:
-        db = lancedb.connect(
-            "db://dev",
-            api_key="fake",
-            host_override=f"http://localhost:{port}",
-            client_config={
-                "retry_config": {"retries": 0},
-                "timeout_config": {"connect_timeout": 2, "read_timeout": 2},
-            },
-        )
-        table = db.open_table("test")
-        assert table.count_rows() == 7
-
-        ctx = mp.get_context("fork")
-        queue = ctx.Queue()
-        proc = ctx.Process(target=_remote_table_fork_child, args=(table, queue))
-        proc.start()
-        proc.join(timeout=15)
-
-        if proc.is_alive():
-            proc.terminate()
-            proc.join(timeout=5)
-            if proc.is_alive():
-                proc.kill()
-                proc.join()
-            pytest.fail("Remote table hung after fork")
-
-        assert proc.exitcode == 0, f"child exited with code {proc.exitcode}"
-        assert not queue.empty(), "child produced no result"
-        assert queue.get() == 7
-    finally:
-        server.shutdown()
-        server_thread.join()
--- a/python/python/tests/test_table.py
+++ b/python/python/tests/test_table.py
@@ -4,7 +4,6 @@

 import os
 import sys
-import warnings
 from datetime import date, datetime, timedelta
 from time import sleep
 from typing import List
@@ -12,7 +11,7 @@ from unittest.mock import patch

 import lancedb
 from lancedb.dependencies import _PANDAS_AVAILABLE
-from lancedb.index import BTree, FTS, HnswFlat, HnswPq, HnswSq, IvfPq
+from lancedb.index import HnswFlat, HnswPq, HnswSq, IvfPq
 import numpy as np
 import polars as pl
 import pyarrow as pa
@@ -26,28 +25,6 @@ from lancedb.table import LanceTable
 from pydantic import BaseModel


-def _blob_test_data():
-    return pa.table(
-        {
-            "id": pa.array([1, 2], pa.int64()),
-            "blob": pa.array([b"hello", b"world"], pa.large_binary()),
-        },
-        schema=pa.schema(
-            [
-                pa.field("id", pa.int64()),
-                pa.field(
-                    "blob", pa.large_binary(), metadata={"lance-encoding:blob": "true"}
-                ),
-            ]
-        ),
-    )
-
-
-def _assert_lazy_blob(value, expected: bytes):
-    assert hasattr(value, "readall")
-    assert value.readall() == expected
-
-
 def test_basic(mem_db: DBConnection):
    data = [
        {"vector": [3.1, 4.1], "item": "foo", "price": 10.0},
@@ -79,22 +56,27 @@ def test_table_to_pandas_default_matches_arrow(tmp_db: DBConnection):
    pd.testing.assert_frame_equal(table.to_pandas(), expected)


-@pytest.mark.parametrize("blob_mode", ["lazy", "bytes", "descriptions"])
-def test_table_to_pandas_blob_modes(tmp_db: DBConnection, blob_mode):
+def test_table_to_pandas_blob_bytes(tmp_db: DBConnection):
    pytest.importorskip("lance")
-    table = tmp_db.create_table(f"test_to_pandas_blob_{blob_mode}", _blob_test_data())
+    data = pa.table(
+        {
+            "id": pa.array([1, 2], pa.int64()),
+            "blob": pa.array([b"hello", b"world"], pa.large_binary()),
+        },
+        schema=pa.schema(
+            [
+                pa.field("id", pa.int64()),
+                pa.field(
+                    "blob", pa.large_binary(), metadata={"lance-encoding:blob": "true"}
+                ),
+            ]
+        ),
+    )
+    table = tmp_db.create_table("test_to_pandas_blob_bytes", data=data)

-    df = table.to_pandas(blob_mode=blob_mode)
+    df = table.to_pandas(blob_mode="bytes")

-    if blob_mode == "lazy":
-        _assert_lazy_blob(df["blob"].iloc[0], b"hello")
-        _assert_lazy_blob(df["blob"].iloc[1], b"world")
-    elif blob_mode == "bytes":
-        assert df["blob"].tolist() == [b"hello", b"world"]
-    else:
-        first = df["blob"].iloc[0]
-        assert first != b"hello"
-        assert not hasattr(first, "readall")
+    assert df["blob"].tolist() == [b"hello", b"world"]


 def test_table_to_pandas_kwargs(tmp_db: DBConnection):
@@ -110,8 +92,22 @@ def test_table_to_pandas_kwargs(tmp_db: DBConnection):
@pytest.mark.asyncio
 async def test_async_table_to_pandas_blob_bytes(tmp_db_async: AsyncConnection):
    pytest.importorskip("lance")
+    data = pa.table(
+        {
+            "id": pa.array([1, 2], pa.int64()),
+            "blob": pa.array([b"hello", b"world"], pa.large_binary()),
+        },
+        schema=pa.schema(
+            [
+                pa.field("id", pa.int64()),
+                pa.field(
+                    "blob", pa.large_binary(), metadata={"lance-encoding:blob": "true"}
+                ),
+            ]
+        ),
+    )
    table = await tmp_db_async.create_table(
-        "test_async_to_pandas_blob_bytes", data=_blob_test_data()
+        "test_async_to_pandas_blob_bytes", data=data
    )

    df = await table.to_pandas(blob_mode="bytes")
@@ -932,12 +928,7 @@ def test_create_index_method(mock_create_index, mem_db: DBConnection):
        num_bits=4,
    )
    mock_create_index.assert_called_with(
-        "vector",
-        replace=True,
-        config=expected_config,
-        wait_timeout=None,
-        name=None,
-        train=True,
+        "vector", replace=True, config=expected_config, name=None, train=True
    )

    # Test with target_partition_size
@@ -957,12 +948,7 @@ def test_create_index_method(mock_create_index, mem_db: DBConnection):
        target_partition_size=8192,
    )
    mock_create_index.assert_called_with(
-        "vector",
-        replace=True,
-        config=expected_config,
-        wait_timeout=None,
-        name=None,
-        train=True,
+        "vector", replace=True, config=expected_config, name=None, train=True
    )

    # target_partition_size has a default value,
@@ -981,12 +967,7 @@ def test_create_index_method(mock_create_index, mem_db: DBConnection):
        num_bits=4,
    )
    mock_create_index.assert_called_with(
-        "vector",
-        replace=True,
-        config=expected_config,
-        wait_timeout=None,
-        name=None,
-        train=True,
+        "vector", replace=True, config=expected_config, name=None, train=True
    )

    table.create_index(
@@ -997,12 +978,7 @@ def test_create_index_method(mock_create_index, mem_db: DBConnection):
    )
    expected_config = HnswPq(distance_type="dot")
    mock_create_index.assert_called_with(
-        "my_vector",
-        replace=False,
-        config=expected_config,
-        wait_timeout=None,
-        name=None,
-        train=True,
+        "my_vector", replace=False, config=expected_config, name=None, train=True
    )

    table.create_index(
@@ -1017,12 +993,7 @@ def test_create_index_method(mock_create_index, mem_db: DBConnection):
        distance_type="cosine", sample_rate=0.1, m=29, ef_construction=10
    )
    mock_create_index.assert_called_with(
-        "my_vector",
-        replace=True,
-        config=expected_config,
-        wait_timeout=None,
-        name=None,
-        train=True,
+        "my_vector", replace=True, config=expected_config, name=None, train=True
    )

    table.create_index(
@@ -1037,12 +1008,7 @@ def test_create_index_method(mock_create_index, mem_db: DBConnection):
        distance_type="cosine", sample_rate=0.1, m=29, ef_construction=10
    )
    mock_create_index.assert_called_with(
-        "my_vector",
-        replace=True,
-        config=expected_config,
-        wait_timeout=None,
-        name=None,
-        train=True,
+        "my_vector", replace=True, config=expected_config, name=None, train=True
    )


@@ -1066,7 +1032,6 @@ def test_create_index_name_and_train_parameters(
        "vector",
        replace=True,
        config=expected_config,
-        wait_timeout=None,
        name="my_custom_index",
        train=True,
    )
@@ -1074,82 +1039,13 @@ def test_create_index_name_and_train_parameters(
    # Test with train=False
    table.create_index(vector_column_name="vector", train=False)
    mock_create_index.assert_called_with(
-        "vector",
-        replace=True,
-        config=expected_config,
-        wait_timeout=None,
-        name=None,
-        train=False,
+        "vector", replace=True, config=expected_config, name=None, train=False
    )

    # Test with both name and train
    table.create_index(vector_column_name="vector", name="my_index_name", train=True)
    mock_create_index.assert_called_with(
-        "vector",
-        replace=True,
-        config=expected_config,
-        wait_timeout=None,
-        name="my_index_name",
-        train=True,
-    )
-
-
-@patch("lancedb.table.AsyncTable.create_index")
-def test_create_index_legacy_emits_deprecation_warning(
-    mock_create_index, mem_db: DBConnection
-):
-    table = mem_db.create_table(
-        "test",
-        data=[{"vector": [3.1, 4.1]}, {"vector": [5.9, 26.5]}],
-    )
-
-    with pytest.warns(DeprecationWarning, match="create_index"):
-        table.create_index(metric="l2", num_partitions=8, vector_column_name="vector")
-
-
-@patch("lancedb.table.AsyncTable.create_index")
-def test_create_index_new_api(mock_create_index, mem_db: DBConnection):
-    table = mem_db.create_table(
-        "test",
-        data=[
-            {"vector": [3.1, 4.1], "category": "a", "text": "hello world"},
-            {"vector": [5.9, 26.5], "category": "b", "text": "goodbye"},
-        ],
-    )
-
-    # Vector index via new API should not warn
-    with warnings.catch_warnings():
-        warnings.simplefilter("error", DeprecationWarning)
-        table.create_index("vector", config=IvfPq(distance_type="l2"))
-    mock_create_index.assert_called_with(
-        "vector",
-        replace=True,
-        config=IvfPq(distance_type="l2"),
-        wait_timeout=None,
-        name=None,
-        train=True,
-    )
-
-    # Scalar index via new API
-    table.create_index("category", config=BTree())
-    mock_create_index.assert_called_with(
-        "category",
-        replace=True,
-        config=BTree(),
-        wait_timeout=None,
-        name=None,
-        train=True,
-    )
-
-    # FTS index via new API
-    table.create_index("text", config=FTS(with_position=True))
-    mock_create_index.assert_called_with(
-        "text",
-        replace=True,
-        config=FTS(with_position=True),
-        wait_timeout=None,
-        name=None,
-        train=True,
+        "vector", replace=True, config=expected_config, name="my_index_name", train=True
    )


@@ -1965,9 +1861,8 @@ def test_create_scalar_index(mem_db: DBConnection):
        "my_table",
        data=test_data,
    )
-    # Test with default name; confirm DeprecationWarning fires
-    with pytest.warns(DeprecationWarning, match="create_scalar_index"):
-        table.create_scalar_index("x")
+    # Test with default name
+    table.create_scalar_index("x")
    indices = table.list_indices()
    assert len(indices) == 1
    scalar_index = indices[0]
--- a/python/python/tests/test_torch.py
+++ b/python/python/tests/test_torch.py
@@ -1,15 +1,10 @@
 # SPDX-License-Identifier: Apache-2.0
 # SPDX-FileCopyrightText: Copyright The LanceDB Authors

-import contextlib
 import functools
-import http.server
-import json
 import multiprocessing as mp
 import pickle
-import re
 import sys
-import threading

 import lancedb
 import pyarrow as pa
@@ -20,107 +15,6 @@ from lancedb.util import tbl_to_tensor
 torch = pytest.importorskip("torch")


-REMOTE_ROWS = list(range(100))
-
-
-def _make_mock_http_handler(handler):
-    class MockLanceDBHandler(http.server.BaseHTTPRequestHandler):
-        def do_GET(self):
-            handler(self)
-
-        def do_POST(self):
-            handler(self)
-
-    return MockLanceDBHandler
-
-
-def _remote_schema_payload():
-    return {
-        "version": 1,
-        "schema": {
-            "fields": [
-                {"name": "a", "type": {"type": "int64"}, "nullable": False},
-            ]
-        },
-    }
-
-
-def _offsets_from_filter(filter_sql: str | None) -> list[int]:
-    if filter_sql is None:
-        return REMOTE_ROWS
-    match = re.search(r"_rowoffset in \((.*?)\)", filter_sql)
-    if match is None:
-        return REMOTE_ROWS
-    raw_offsets = match.group(1).strip()
-    if raw_offsets == "":
-        return []
-    return [int(offset.strip()) for offset in raw_offsets.split(",")]
-
-
-def _remote_dataset_handler(request):
-    request.close_connection = True
-    if request.path == "/v1/table/test/describe/":
-        request.send_response(200)
-        request.send_header("Content-Type", "application/json")
-        request.end_headers()
-        request.wfile.write(json.dumps(_remote_schema_payload()).encode())
-    elif request.path == "/v1/table/test/count_rows/":
-        request.send_response(200)
-        request.send_header("Content-Type", "application/json")
-        request.end_headers()
-        request.wfile.write(str(len(REMOTE_ROWS)).encode())
-    elif request.path == "/v1/table/test/query/":
-        content_len = int(request.headers.get("Content-Length"))
-        body = json.loads(request.rfile.read(content_len))
-        offsets = _offsets_from_filter(body.get("filter"))
-        requested_columns = body.get("columns") or ["a"]
-        if isinstance(requested_columns, dict):
-            requested_columns = list(requested_columns)
-
-        data = {}
-        for column in requested_columns:
-            if column == "a":
-                data[column] = [REMOTE_ROWS[offset] for offset in offsets]
-            elif column == "_rowoffset":
-                data[column] = offsets
-            elif column == "_rowid":
-                data[column] = offsets
-
-        table = pa.table(data)
-        request.send_response(200)
-        request.send_header("Content-Type", "application/vnd.apache.arrow.file")
-        request.end_headers()
-        with pa.ipc.new_file(request.wfile, schema=table.schema) as writer:
-            writer.write_table(table)
-    else:
-        request.send_response(404)
-        request.end_headers()
-
-
-@contextlib.contextmanager
-def _remote_dataset_table():
-    with http.server.ThreadingHTTPServer(
-        ("localhost", 0), _make_mock_http_handler(_remote_dataset_handler)
-    ) as server:
-        port = server.server_address[1]
-        handle = threading.Thread(target=server.serve_forever)
-        handle.start()
-        try:
-            db = lancedb.connect(
-                "db://dev",
-                api_key="fake",
-                host_override=f"http://localhost:{port}",
-                client_config={
-                    "retry_config": {"retries": 0},
-                    "timeout_config": {"connect_timeout": 2, "read_timeout": 2},
-                },
-            )
-            yield db.open_table("test")
-        finally:
-            server.shutdown()
-            handle.join()
-
-
 def _open_native_table(uri: str, table_name: str):
    """Top-level connection factory used by the explicit-factory pickle test.

@@ -213,39 +107,6 @@ def test_permutation_dataloader_multiprocessing(tmp_db):
    assert seen == 1000


-def test_remote_table_dataloader_multiprocessing():
-    with _remote_dataset_table() as table:
-        dataloader = torch.utils.data.DataLoader(
-            table,
-            collate_fn=tbl_to_tensor,
-            batch_size=10,
-            num_workers=2,
-            multiprocessing_context="spawn",
-        )
-        seen = 0
-        for batch in dataloader:
-            assert batch.size(0) == 1
-            assert batch.size(1) == 10
-            seen += batch.size(1)
-        assert seen == len(REMOTE_ROWS)
-
-
-def test_remote_permutation_dataloader_multiprocessing():
-    with _remote_dataset_table() as table:
-        permutation = Permutation.identity(table)
-        dataloader = torch.utils.data.DataLoader(
-            permutation,
-            batch_size=10,
-            num_workers=2,
-            multiprocessing_context="spawn",
-        )
-        seen = 0
-        for batch in dataloader:
-            assert batch["a"].size(0) == 10
-            seen += batch["a"].size(0)
-        assert seen == len(REMOTE_ROWS)
-
-
 def test_permutation_pickle_with_connection_factory(tmp_path):
    """When the user provides a connection_factory, pickling should round-trip
    through that factory rather than introspecting the connection URI. Useful
@@ -310,35 +171,6 @@ def _multiworker_dataloader_target(db_uri: str, result_queue):
    result_queue.put(count)


-def _remote_multiworker_dataloader_target(port: int, result_queue):
-    import lancedb
-    from lancedb.permutation import Permutation
-
-    db = lancedb.connect(
-        "db://dev",
-        api_key="fake",
-        host_override=f"http://localhost:{port}",
-        client_config={
-            "retry_config": {"retries": 0},
-            "timeout_config": {"connect_timeout": 2, "read_timeout": 2},
-        },
-    )
-    table = db.open_table("test")
-    permutation = Permutation.identity(table)
-
-    dataloader = torch.utils.data.DataLoader(
-        permutation,
-        batch_size=10,
-        num_workers=2,
-        multiprocessing_context="fork",
-    )
-    count = 0
-    for batch in dataloader:
-        assert batch["a"].size(0) == 10
-        count += 1
-    result_queue.put(count)
-
-
@pytest.mark.skipif(
    sys.platform != "linux",
    reason=(
@@ -376,46 +208,3 @@ def test_permutation_dataloader_fork_workers(tmp_path):
    assert proc.exitcode == 0, f"child exited with code {proc.exitcode}"
    assert not queue.empty(), "child produced no batches"
    assert queue.get() == 100
-
-
-@pytest.mark.skipif(
-    sys.platform != "linux",
-    reason=(
-        "fork() is unavailable on Windows and unsafe on macOS "
-        "(Apple frameworks/TLS are not fork-safe)"
-    ),
-)
-def test_remote_permutation_dataloader_fork_workers():
-    with http.server.ThreadingHTTPServer(
-        ("localhost", 0), _make_mock_http_handler(_remote_dataset_handler)
-    ) as server:
-        port = server.server_address[1]
-        handle = threading.Thread(target=server.serve_forever)
-        handle.start()
-        try:
-            ctx = mp.get_context("spawn")
-            queue = ctx.Queue()
-            proc = ctx.Process(
-                target=_remote_multiworker_dataloader_target,
-                args=(port, queue),
-            )
-            proc.start()
-            proc.join(timeout=30)
-
-            if proc.is_alive():
-                proc.terminate()
-                proc.join(timeout=5)
-                if proc.is_alive():
-                    proc.kill()
-                    proc.join()
-                pytest.fail(
-                    "Remote permutation hung when iterated in a fork-based "
-                    "DataLoader worker"
-                )
-
-            assert proc.exitcode == 0, f"child exited with code {proc.exitcode}"
-            assert not queue.empty(), "child produced no batches"
-            assert queue.get() == 10
-        finally:
-            server.shutdown()
-            handle.join()
--- a/python/src/table.rs
+++ b/python/src/table.rs
@@ -143,20 +143,18 @@ pub struct MergeResult {
    pub num_inserted_rows: u64,
    pub num_deleted_rows: u64,
    pub num_attempts: u32,
-    pub num_rows: u64,
 }

 #[pymethods]
 impl MergeResult {
    pub fn __repr__(&self) -> String {
        format!(
-            "MergeResult(version={}, num_updated_rows={}, num_inserted_rows={}, num_deleted_rows={}, num_attempts={}, num_rows={})",
+            "MergeResult(version={}, num_updated_rows={}, num_inserted_rows={}, num_deleted_rows={}, num_attempts={})",
            self.version,
            self.num_updated_rows,
            self.num_inserted_rows,
            self.num_deleted_rows,
-            self.num_attempts,
-            self.num_rows
+            self.num_attempts
        )
    }
 }
@@ -169,7 +167,6 @@ impl From<lancedb::table::MergeResult> for MergeResult {
            num_inserted_rows: result.num_inserted_rows,
            num_deleted_rows: result.num_deleted_rows,
            num_attempts: result.num_attempts,
-            num_rows: result.num_rows,
        }
    }
 }
@@ -197,12 +194,6 @@ impl LsmWriteSpec {
    }

    /// Identity sharding — shard by the raw value of `column`.
-    ///
-    /// `column` must be a deterministic function of the unenforced primary
-    /// key: every row with a given primary key must always produce the same
-    /// `column` value, or upserts of that key can land in different shards
-    /// and a stale version can win. Typically `column` is the primary key
-    /// itself or a stable attribute of it.
    #[staticmethod]
    pub fn identity(column: String) -> Self {
        Self {
@@ -942,12 +933,6 @@ impl Table {
        if let Some(use_index) = parameters.use_index {
            builder.use_index(use_index);
        }
-        if let Some(use_lsm_write) = parameters.use_lsm_write {
-            builder.use_lsm_write(use_lsm_write);
-        }
-        if let Some(validate_single_shard) = parameters.validate_single_shard {
-            builder.validate_single_shard(validate_single_shard);
-        }

        future_into_py(self_.py(), async move {
            let res = builder.execute(Box::new(batches)).await.infer_error()?;
@@ -986,13 +971,6 @@ impl Table {
        })
    }

-    pub fn close_lsm_writers(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
-        let inner = self_.inner_ref()?.clone();
-        future_into_py(self_.py(), async move {
-            inner.close_lsm_writers().await.infer_error()
-        })
-    }
-
    pub fn uses_v2_manifest_paths(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
        let inner = self_.inner_ref()?.clone();
        future_into_py(self_.py(), async move {
@@ -1146,8 +1124,6 @@ pub struct MergeInsertParams {
    when_not_matched_by_source_condition: Option<String>,
    timeout: Option<std::time::Duration>,
    use_index: Option<bool>,
-    use_lsm_write: Option<bool>,
-    validate_single_shard: Option<bool>,
 }

 #[pyclass]
--- a/rust-toolchain.toml
+++ b/rust-toolchain.toml
@@ -1,2 +1,2 @@
 [toolchain]
-channel = "1.95.0"
+channel = "1.94.0"
--- a/rust/lancedb/Cargo.toml
+++ b/rust/lancedb/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "lancedb"
-version = "0.30.1-beta.0"
+version = "0.30.0-beta.1"
 edition.workspace = true
 description = "LanceDB: A serverless, low-latency vector database for AI applications"
 license.workspace = true
@@ -75,7 +75,7 @@ reqwest = { version = "0.12.0", default-features = false, features = [
    "stream",
 ], optional = true }
 http = { version = "1", optional = true } # Matching what is in reqwest
-uuid = { version = "1.7.0", features = ["v4", "v5"] }
+uuid = { version = "1.7.0", features = ["v4"] }
 polars-arrow = { version = ">=0.37,<0.40.0", optional = true }
 polars = { version = ">=0.37,<0.40.0", optional = true }
 hf-hub = { version = "0.4.1", optional = true, default-features = false, features = [
--- a/rust/lancedb/src/dataloader/permutation/shuffle.rs
+++ b/rust/lancedb/src/dataloader/permutation/shuffle.rs
@@ -464,9 +464,11 @@ mod tests {
        let mut iter = ids.into_iter().map(|o| o.unwrap());
        while let Some(first) = iter.next() {
            let rows_left_in_clump = if first == 4470 { 19 } else { 29 };
-            for expected_next in (first + 1)..=(first + rows_left_in_clump) {
+            let mut expected_next = first + 1;
+            for _ in 0..rows_left_in_clump {
                let next = iter.next().unwrap();
                assert_eq!(next, expected_next);
+                expected_next += 1;
            }
        }
    }
--- a/rust/lancedb/src/remote/client.rs
+++ b/rust/lancedb/src/remote/client.rs
@@ -908,15 +908,6 @@ mod tests {
    use serial_test::serial;
    use std::time::Duration;

-    // Serializes the env-var-mutating tests below: cargo test runs tests in
-    // parallel, but several of these tests read and write the same process-
-    // global env vars (`LANCEDB_USER_ID*`), so they would race without this.
-    static ENV_MUTEX: std::sync::Mutex<()> = std::sync::Mutex::new(());
-
-    fn lock_env() -> std::sync::MutexGuard<'static, ()> {
-        ENV_MUTEX.lock().unwrap_or_else(|e| e.into_inner())
-    }
-
    #[test]
    fn test_timeout_config_default() {
        let config = TimeoutConfig::default();
@@ -1175,7 +1166,6 @@ mod tests {
    #[test]
    #[serial(user_id_env)]
    fn test_resolve_user_id_none() {
-        let _guard = lock_env();
        let config = ClientConfig::default();
        // Clear env vars that might be set from other tests
        // SAFETY: This is only called in tests
@@ -1189,7 +1179,6 @@ mod tests {
    #[test]
    #[serial(user_id_env)]
    fn test_resolve_user_id_from_env() {
-        let _guard = lock_env();
        // SAFETY: This is only called in tests
        unsafe {
            std::env::set_var("LANCEDB_USER_ID", "env-user-id");
@@ -1205,7 +1194,6 @@ mod tests {
    #[test]
    #[serial(user_id_env)]
    fn test_resolve_user_id_from_env_key() {
-        let _guard = lock_env();
        // SAFETY: This is only called in tests
        unsafe {
            std::env::remove_var("LANCEDB_USER_ID");
@@ -1227,7 +1215,6 @@ mod tests {
    #[test]
    #[serial(user_id_env)]
    fn test_resolve_user_id_direct_takes_precedence() {
-        let _guard = lock_env();
        // SAFETY: This is only called in tests
        unsafe {
            std::env::set_var("LANCEDB_USER_ID", "env-user-id");
@@ -1246,7 +1233,6 @@ mod tests {
    #[test]
    #[serial(user_id_env)]
    fn test_resolve_user_id_empty_env_ignored() {
-        let _guard = lock_env();
        // SAFETY: This is only called in tests
        unsafe {
            std::env::set_var("LANCEDB_USER_ID", "");
--- a/rust/lancedb/src/remote/table.rs
+++ b/rust/lancedb/src/remote/table.rs
@@ -66,47 +66,36 @@ use std::time::{Duration, SystemTime};
 use tokio::sync::RwLock;

 const REQUEST_TIMEOUT_HEADER: HeaderName = HeaderName::from_static("x-request-timeout-ms");
-const MIN_VERSION_HEADER: HeaderName = HeaderName::from_static("x-lancedb-min-version");
 const MIN_TIMESTAMP_HEADER: HeaderName = HeaderName::from_static("x-lancedb-min-timestamp");
 const METRIC_TYPE_KEY: &str = "metric_type";
 const INDEX_TYPE_KEY: &str = "index_type";
 const SCHEMA_CACHE_TTL: Duration = Duration::from_secs(30);
 const SCHEMA_CACHE_REFRESH_WINDOW: Duration = Duration::from_secs(5);

-/// Per-table state driving the freshness headers (`x-lancedb-min-version` and
-/// `x-lancedb-min-timestamp`) sent on read requests.
+/// Per-table state driving the `x-lancedb-min-timestamp` freshness header
+/// sent on read requests.
 #[derive(Debug, Default, Clone, Copy)]
 struct FreshnessState {
-    /// Provides read-your-write within a single handle: writes that return a
-    /// version update this, and reads send it as `x-lancedb-min-version`.
-    min_version: Option<u64>,
-    /// Wall-clock time captured at the last [`BaseTable::checkout_latest`]
-    /// call. Subsequent reads send
-    /// `max(baseline, now - read_consistency_interval)` as
-    /// `x-lancedb-min-timestamp`.
+    /// Wall-clock floor for read freshness, bumped to `now` whenever this
+    /// handle performs a write or an explicit [`BaseTable::checkout_latest`].
+    /// Subsequent reads send `max(baseline, now - read_consistency_interval)`
+    /// as `x-lancedb-min-timestamp`.
    ///
-    /// Without this, `checkout_latest()` would have no effect on subsequent
-    /// reads when `read_consistency_interval` is unset (the default): a
-    /// server-side cache could still serve a snapshot older than the moment
-    /// the user explicitly asked for "latest". The baseline forces the
-    /// server to skip any cache entry older than the checkout time, so the
-    /// `checkout_latest()` signal is preserved across reads on the same
-    /// handle regardless of the configured consistency interval.
-    checkout_baseline: Option<SystemTime>,
+    /// This is what provides read-your-write on a single handle: after a write
+    /// the next read forces the server cache past the write time. It also
+    /// preserves the `checkout_latest()` signal when `read_consistency_interval`
+    /// is unset (the default), where there is no interval-based floor.
+    freshness_baseline: Option<SystemTime>,
 }

 /// Snapshot of the headers that should be attached to a single read request.
 #[derive(Debug, Default, Clone, Copy)]
 struct FreshnessHeaders {
-    min_version: Option<u64>,
    min_timestamp: Option<SystemTime>,
 }

 impl FreshnessHeaders {
    fn apply(self, mut request: RequestBuilder) -> RequestBuilder {
-        if let Some(v) = self.min_version {
-            request = request.header(MIN_VERSION_HEADER, v.to_string());
-        }
        if let Some(ts) = self.min_timestamp {
            let dt: chrono::DateTime<chrono::Utc> = ts.into();
            request = request.header(MIN_TIMESTAMP_HEADER, dt.to_rfc3339());
@@ -115,6 +104,14 @@ impl FreshnessHeaders {
    }
 }

+/// Monotonic floor for the freshness baseline. `SystemTime` is not monotonic
+/// (NTP steps, hibernate/resume can move it backward), so a write must never
+/// lower the baseline below a prior write's — that would let the next read send
+/// a `min_timestamp` earlier than an earlier write and break read-your-write.
+fn next_freshness_baseline(prev: Option<SystemTime>, now: SystemTime) -> SystemTime {
+    prev.map_or(now, |prev| prev.max(now))
+}
+
 fn compute_min_timestamp(
    state: &FreshnessState,
    interval: Option<Duration>,
@@ -125,7 +122,7 @@ fn compute_min_timestamp(
        Some(d) if d.is_zero() => Some(now),
        Some(d) => Some(now.checked_sub(d).unwrap_or(now)),
    };
-    match (interval_based, state.checkout_baseline) {
+    match (interval_based, state.freshness_baseline) {
        (None, None) => None,
        (Some(t), None) | (None, Some(t)) => Some(t),
        (Some(a), Some(b)) => Some(a.max(b)),
@@ -787,7 +784,6 @@ impl<S: HttpSend> RemoteTable<S> {
    fn snapshot_freshness_headers(&self) -> FreshnessHeaders {
        let state = *self.freshness.lock().unwrap();
        FreshnessHeaders {
-            min_version: state.min_version,
            min_timestamp: compute_min_timestamp(
                &state,
                self.client.read_consistency_interval,
@@ -796,22 +792,20 @@ impl<S: HttpSend> RemoteTable<S> {
        }
    }

-    /// Build a POST request and attach the read-freshness headers
-    /// (`x-lancedb-min-version`, `x-lancedb-min-timestamp`).
+    /// Build a POST request and attach the `x-lancedb-min-timestamp` freshness
+    /// header.
    fn post_read(&self, uri: &str) -> RequestBuilder {
        self.snapshot_freshness_headers()
            .apply(self.client.post(uri))
    }

-    /// Record a version returned by a write so subsequent reads can request at
-    /// least that version via `x-lancedb-min-version`. A returned `0` from a
-    /// backward-compatible old server is ignored.
-    fn track_write_version(&self, version: u64) {
-        if version == 0 {
-            return;
-        }
+    /// Record that this handle just performed a write, so the next read forces
+    /// the server cache past the write time (read-your-write on a single
+    /// handle).
+    fn bump_freshness_baseline(&self) {
+        let now = SystemTime::now();
        let mut state = self.freshness.lock().unwrap();
-        state.min_version = Some(state.min_version.map_or(version, |v| v.max(version)));
+        state.freshness_baseline = Some(next_freshness_baseline(state.freshness_baseline, now));
    }

    async fn execute_query(
@@ -1115,7 +1109,7 @@ impl<S: HttpSend + 'static> RemoteTable<S> {
                    if output.overwrite {
                        self.invalidate_schema_cache();
                    }
-                    self.track_write_version(add_result.version);
+                    self.bump_freshness_baseline();

                    return Ok(add_result);
                }
@@ -1153,7 +1147,7 @@ impl<S: HttpSend + 'static> RemoteTable<S> {
                        if output.overwrite {
                            self.invalidate_schema_cache();
                        }
-                        self.track_write_version(result.version);
+                        self.bump_freshness_baseline();
                        return Ok(result);
                    }
                    Err(e) => {
@@ -1271,7 +1265,7 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
    }
    async fn checkout(&self, version: u64) -> Result<()> {
        // Validate the version exists. The describe is sent without freshness
-        // headers so a stale `min_version` from a previous write doesn't ride
+        // headers so a stale baseline from a previous write doesn't ride
        // along on an explicit time-travel request.
        let request = self
            .client
@@ -1306,11 +1300,10 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
        *write_guard = None;
        drop(write_guard);

-        // Drop any per-handle write tracking; subsequent reads use the
-        // baseline timestamp captured now to guarantee freshness.
+        // Reset the freshness baseline to now so subsequent reads see at least
+        // the state as of this explicit `checkout_latest()`.
        *self.freshness.lock().unwrap() = FreshnessState {
-            min_version: None,
-            checkout_baseline: Some(SystemTime::now()),
+            freshness_baseline: Some(SystemTime::now()),
        };

        // Invalidate schema cache since we're switching versions
@@ -1627,7 +1620,7 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
                status_code: None,
            })?;

-        self.track_write_version(update_response.version);
+        self.bump_freshness_baseline();
        Ok(update_response)
    }

@@ -1658,7 +1651,7 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
                request_id,
                status_code: None,
            })?;
-        self.track_write_version(delete_response.version);
+        self.bump_freshness_baseline();
        Ok(delete_response)
    }

@@ -1805,7 +1798,6 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
                num_inserted_rows: 0,
                num_updated_rows: 0,
                num_attempts: 0,
-                num_rows: 0,
            });
        }

@@ -1816,7 +1808,7 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
                status_code: None,
            })?;

-        self.track_write_version(merge_insert_response.version);
+        self.bump_freshness_baseline();
        Ok(merge_insert_response)
    }

@@ -1843,7 +1835,7 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
    }
    async fn checkout_tag(&self, tag: &str) -> Result<()> {
        // Resolve the tag without attaching freshness headers; a stale
-        // `min_version` from a previous write should not ride along on an
+        // baseline from a previous write should not ride along on an
        // explicit time-travel request.
        let request = self
            .client
@@ -1908,7 +1900,7 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
                    })?;

                self.invalidate_schema_cache();
-                self.track_write_version(result.version);
+                self.bump_freshness_baseline();

                Ok(result)
            }
@@ -1963,7 +1955,7 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
        })?;

        self.invalidate_schema_cache();
-        self.track_write_version(result.version);
+        self.bump_freshness_baseline();

        Ok(result)
    }
@@ -1991,7 +1983,7 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
        })?;

        self.invalidate_schema_cache();
-        self.track_write_version(result.version);
+        self.bump_freshness_baseline();

        Ok(result)
    }
@@ -6168,6 +6160,21 @@ mod tests {

    // ---- Read freshness header tests ------------------------------------

+    #[test]
+    fn test_next_freshness_baseline_is_monotonic() {
+        let t100 = SystemTime::UNIX_EPOCH + Duration::from_secs(100);
+        let t101 = SystemTime::UNIX_EPOCH + Duration::from_secs(101);
+        let t99 = SystemTime::UNIX_EPOCH + Duration::from_secs(99);
+
+        // No prior baseline -> take the current time.
+        assert_eq!(next_freshness_baseline(None, t100), t100);
+        // Clock moved forward -> advance to it.
+        assert_eq!(next_freshness_baseline(Some(t100), t101), t101);
+        // Clock moved backward (NTP step / resume) -> keep the higher prior
+        // baseline so read-your-write isn't lowered below an earlier write.
+        assert_eq!(next_freshness_baseline(Some(t100), t99), t100);
+    }
+
    #[test]
    fn test_compute_min_timestamp_combines_baseline_and_interval() {
        let now = SystemTime::now();
@@ -6181,8 +6188,7 @@ mod tests {

        // Baseline only -> baseline.
        let state = FreshnessState {
-            min_version: None,
-            checkout_baseline: Some(baseline),
+            freshness_baseline: Some(baseline),
        };
        assert_eq!(compute_min_timestamp(&state, None, now), Some(baseline));

@@ -6205,8 +6211,7 @@ mod tests {
        // Both: pick the more-recent (i.e. tighter) constraint.
        // baseline = now-60, now-interval = now-10. now-10 is newer.
        let state = FreshnessState {
-            min_version: None,
-            checkout_baseline: Some(baseline),
+            freshness_baseline: Some(baseline),
        };
        assert_eq!(
            compute_min_timestamp(&state, Some(Duration::from_secs(10)), now),
@@ -6216,8 +6221,7 @@ mod tests {
        // Both, baseline newer: pick baseline.
        let recent_baseline = now - Duration::from_secs(5);
        let state = FreshnessState {
-            min_version: None,
-            checkout_baseline: Some(recent_baseline),
+            freshness_baseline: Some(recent_baseline),
        };
        assert_eq!(
            compute_min_timestamp(&state, Some(Duration::from_secs(60)), now),
@@ -6272,7 +6276,6 @@ mod tests {

        let headers = captured.lock().unwrap().clone().unwrap();
        assert!(!headers.contains_key("x-lancedb-min-timestamp"));
-        assert!(!headers.contains_key("x-lancedb-min-version"));
    }

    #[tokio::test]
@@ -6291,7 +6294,6 @@ mod tests {
            sent >= before - FRESHNESS_TOLERANCE && sent <= after + FRESHNESS_TOLERANCE,
            "expected timestamp roughly equal to wall clock"
        );
-        assert!(!headers.contains_key("x-lancedb-min-version"));
    }

    #[tokio::test]
@@ -6335,11 +6337,10 @@ mod tests {
                && sent <= after_checkout + FRESHNESS_TOLERANCE,
            "expected timestamp captured at checkout_latest() time"
        );
-        assert!(!headers.contains_key("x-lancedb-min-version"));
    }

    #[tokio::test]
-    async fn test_freshness_min_version_tracked_after_write() {
+    async fn test_freshness_baseline_bumped_after_write() {
        let (handler, captured) = capturing_handler(|path| match path {
            "/v1/table/my_table/update/" => r#"{"rows_updated":1,"version":7}"#.to_string(),
            "/v1/table/my_table/count_rows/" => "42".to_string(),
@@ -6347,19 +6348,17 @@ mod tests {
        });
        let table = Table::new_with_handler("my_table", handler);

+        let before = SystemTime::now();
        let _ = table.update().column("a", "a + 1").execute().await.unwrap();
-        // Update headers also pass through captured; reset by reading after.
        table.count_rows(None).await.unwrap();
+        let after = SystemTime::now();

+        // The write bumped the freshness baseline, so the next read carries a
+        // min-timestamp at or after the write time (read-your-write).
        let headers = captured.lock().unwrap().clone().unwrap();
-        assert_eq!(
-            headers
-                .get("x-lancedb-min-version")
-                .unwrap()
-                .to_str()
-                .unwrap(),
-            "7"
-        );
+        let ts = parse_min_timestamp(&headers);
+        assert!(ts >= before - FRESHNESS_TOLERANCE);
+        assert!(ts <= after + FRESHNESS_TOLERANCE);
    }

    /// Like `capturing_handler`, but keeps a per-path snapshot of the headers
@@ -6392,9 +6391,10 @@ mod tests {
    }

    #[tokio::test]
-    async fn test_freshness_checkout_validation_sends_no_min_version() {
-        // After a write bumps min_version, calling checkout(v) must not let
-        // that stale header ride along on the validating /describe/ request.
+    async fn test_freshness_checkout_validation_sends_no_freshness_headers() {
+        // After a write bumps the baseline, calling checkout(v) must not let
+        // that stale freshness header ride along on the validating /describe/
+        // request.
        let (handler, captured) = path_capturing_handler(|path| match path {
            "/v1/table/my_table/update/" => r#"{"rows_updated":1,"version":7}"#.to_string(),
            "/v1/table/my_table/describe/" => r#"{"version":5,"schema":{"fields":[]}}"#.to_string(),
@@ -6410,16 +6410,15 @@ mod tests {
            .get("/v1/table/my_table/describe/")
            .expect("describe should have been called by checkout(v)");
        assert!(
-            !describe_headers.contains_key("x-lancedb-min-version"),
-            "checkout(v) describe must not carry stale min_version",
+            !describe_headers.contains_key("x-lancedb-min-timestamp"),
+            "checkout(v) describe must not carry a stale freshness baseline",
        );
-        assert!(!describe_headers.contains_key("x-lancedb-min-timestamp"));
    }

    #[tokio::test]
-    async fn test_freshness_checkout_tag_resolve_sends_no_min_version() {
+    async fn test_freshness_checkout_tag_resolve_sends_no_freshness_headers() {
        // Same invariant for checkout_tag: the tag-resolve request must not
-        // pick up a stale min_version from a prior write.
+        // pick up a stale freshness baseline from a prior write.
        let (handler, captured) = path_capturing_handler(|path| match path {
            "/v1/table/my_table/update/" => r#"{"rows_updated":1,"version":7}"#.to_string(),
            "/v1/table/my_table/tags/version/" => r#"{"version":5}"#.to_string(),
@@ -6435,14 +6434,13 @@ mod tests {
            .get("/v1/table/my_table/tags/version/")
            .expect("tags/version should have been called by checkout_tag");
        assert!(
-            !resolve_headers.contains_key("x-lancedb-min-version"),
-            "checkout_tag resolve must not carry stale min_version",
+            !resolve_headers.contains_key("x-lancedb-min-timestamp"),
+            "checkout_tag resolve must not carry a stale freshness baseline",
        );
-        assert!(!resolve_headers.contains_key("x-lancedb-min-timestamp"));
    }

    #[tokio::test]
-    async fn test_freshness_checkout_clears_min_version() {
+    async fn test_freshness_checkout_clears_baseline() {
        let (handler, captured) = capturing_handler(|path| match path {
            "/v1/table/my_table/update/" => r#"{"rows_updated":1,"version":7}"#.to_string(),
            // checkout(5) needs to describe version 5 first
@@ -6457,7 +6455,6 @@ mod tests {
        table.count_rows(None).await.unwrap();

        let headers = captured.lock().unwrap().clone().unwrap();
-        assert!(!headers.contains_key("x-lancedb-min-version"));
        assert!(!headers.contains_key("x-lancedb-min-timestamp"));
    }
 }
--- a/rust/lancedb/src/table.rs
+++ b/rust/lancedb/src/table.rs
@@ -89,6 +89,7 @@ use futures::future::join_all;
 pub use lance::dataset::refs::{TagContents, Tags as LanceTags};
 pub use lance::dataset::scanner::DatasetRecordBatchStream;
 use lance::dataset::statistics::DatasetStatisticsExt;
+use lance_index::frag_reuse::FRAG_REUSE_INDEX_NAME;
 pub use lance_index::optimize::OptimizeOptions;
 pub use optimize::{CompactionOptions, OptimizeAction, OptimizeStats};
 pub use schema_evolution::{AddColumnsResult, AlterColumnsResult, DropColumnsResult};
@@ -366,14 +367,6 @@ impl LsmWriteSpec {

    /// Construct an identity-sharding spec (shard by the raw value of
    /// `column`) with no maintained indexes.
-    ///
-    /// `column` must be a deterministic function of the unenforced primary
-    /// key: every row with a given primary key must always produce the same
-    /// `column` value. MemWAL dedups upserts by primary key but tracks
-    /// generations per shard, so if the same key is written with two
-    /// different `column` values its versions land in different shards and a
-    /// stale value can win. Typically `column` is the primary key itself, or
-    /// a stable attribute of it (e.g. a tenant id).
    pub fn identity(column: impl Into<String>) -> Self {
        Self::Identity {
            column: column.into(),
@@ -588,13 +581,6 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
            message: "unset_lsm_write_spec is not supported on this table type".into(),
        })
    }
-    /// Drain and close any cached MemWAL shard writers for this table.
-    ///
-    /// The default implementation is a no-op; table types that maintain
-    /// MemWAL shard writers override it.
-    async fn close_lsm_writers(&self) -> Result<()> {
-        Ok(())
-    }
    /// Gets the table tag manager.
    async fn tags(&self) -> Result<Box<dyn Tags + '_>>;
    /// Optimize the dataset.
@@ -1401,16 +1387,6 @@ impl Table {
        self.inner.unset_lsm_write_spec().await
    }

-    /// Drain and close any cached MemWAL shard writers held for this table.
-    ///
-    /// When an [`LsmWriteSpec`] is installed, `merge_insert` opens MemWAL shard
-    /// writers and caches them for reuse across calls. This closes them,
-    /// flushing pending data; writers reopen lazily on the next `merge_insert`.
-    /// It is a no-op when no writers are cached.
-    pub async fn close_lsm_writers(&self) -> Result<()> {
-        self.inner.close_lsm_writers().await
-    }
-
    /// Retrieve the version of the table
    ///
    /// LanceDb supports versioning.  Every operation that modifies the table increases
@@ -2854,10 +2830,6 @@ impl BaseTable for NativeTable {
        merge::lsm::unset_lsm_write_spec(self).await
    }

-    async fn close_lsm_writers(&self) -> Result<()> {
-        merge::lsm::close_lsm_writers(self).await
-    }
-
    /// Delete rows from the table
    async fn delete(&self, predicate: Predicate<'_>) -> Result<DeleteResult> {
        delete::execute_delete(self, predicate).await
@@ -2892,32 +2864,71 @@ impl BaseTable for NativeTable {

    async fn list_indices(&self) -> Result<Vec<IndexConfig>> {
        let dataset = self.dataset.get().await?;
-        let indices = dataset
-            .describe_indices(None)
-            .await?
-            .into_iter()
-            .filter_map(|idx_desc| {
-                let index_type: crate::index::IndexType = match idx_desc.index_type().parse() {
-                    Ok(index_type) => index_type,
+        let indices = dataset.load_indices().await?;
+        let results = futures::stream::iter(indices.as_slice())
+            .then(|idx| async {
+                // skip Lance internal indexes
+                if idx.name == FRAG_REUSE_INDEX_NAME {
+                    return None;
+                }
+
+                let stats = match dataset.index_statistics(idx.name.as_str()).await {
+                    Ok(stats) => stats,
                    Err(e) => {
                        log::warn!(
-                            "Failed to parse index type for index {}: {}",
-                            idx_desc.name(),
+                            "Failed to get statistics for index {} ({}): {}",
+                            idx.name,
+                            idx.uuid,
                            e
                        );
                        return None;
                    }
                };

-                let field_ids = idx_desc.field_ids();
-                let mut columns = Vec::with_capacity(field_ids.len());
-                for field_id in field_ids {
-                    let field_path = match dataset.schema().field_path(*field_id as i32) {
+                let stats: serde_json::Value = match serde_json::from_str(&stats) {
+                    Ok(stats) => stats,
+                    Err(e) => {
+                        log::warn!(
+                            "Failed to deserialize index statistics for index {} ({}): {}",
+                            idx.name,
+                            idx.uuid,
+                            e
+                        );
+                        return None;
+                    }
+                };
+
+                let Some(index_type) = stats.get("index_type").and_then(|v| v.as_str()) else {
+                    log::warn!(
+                        "Index statistics was missing 'index_type' field for index {} ({})",
+                        idx.name,
+                        idx.uuid
+                    );
+                    return None;
+                };
+
+                let index_type: crate::index::IndexType = match index_type.parse() {
+                    Ok(index_type) => index_type,
+                    Err(e) => {
+                        log::warn!(
+                            "Failed to parse index type for index {} ({}): {}",
+                            idx.name,
+                            idx.uuid,
+                            e
+                        );
+                        return None;
+                    }
+                };
+
+                let mut columns = Vec::with_capacity(idx.fields.len());
+                for field_id in &idx.fields {
+                    let field_path = match dataset.schema().field_path(*field_id) {
                        Ok(field_path) => field_path,
                        Err(e) => {
                            log::warn!(
-                                "Failed to resolve field path for index {} field id {}: {}",
-                                idx_desc.name(),
+                                "Failed to resolve field path for index {} ({}) field id {}: {}",
+                                idx.name,
+                                idx.uuid,
                                field_id,
                                e
                            );
@@ -2927,14 +2938,17 @@ impl BaseTable for NativeTable {
                    columns.push(field_path);
                }

+                let name = idx.name.clone();
                Some(IndexConfig {
-                    name: idx_desc.name().to_string(),
                    index_type,
                    columns,
+                    name,
                })
            })
-            .collect();
-        Ok(indices)
+            .collect::<Vec<_>>()
+            .await;
+
+        Ok(results.into_iter().flatten().collect())
    }

    async fn uri(&self) -> Result<String> {
@@ -3044,12 +3058,11 @@ impl BaseTable for NativeTable {
        let p99 = *sorted_sizes.get(num_fragments * 99 / 100).unwrap_or(&0);
        let min = sorted_sizes.first().copied().unwrap_or(0);
        let max = sorted_sizes.last().copied().unwrap_or(0);
-        let mean = sorted_sizes
-            .iter()
-            .copied()
-            .sum::<usize>()
-            .checked_div(num_fragments)
-            .unwrap_or(0);
+        let mean = if num_fragments == 0 {
+            0
+        } else {
+            sorted_sizes.iter().copied().sum::<usize>() / num_fragments
+        };

        let frag_stats = FragmentStatistics {
            num_fragments,
@@ -4049,27 +4062,26 @@ mod tests {
        let index_configs = table.list_indices().await.unwrap();
        assert_eq!(index_configs.len(), 5);

-        // list_indices returns indices in alphabetical order by name
        let mut configs_iter = index_configs.into_iter();
        let index = configs_iter.next().unwrap();
        assert_eq!(index.index_type, crate::index::IndexType::Bitmap);
        assert_eq!(index.columns, vec!["category".to_string()]);

-        let index = configs_iter.next().unwrap();
-        assert_eq!(index.index_type, crate::index::IndexType::Bitmap);
-        assert_eq!(index.columns, vec!["data".to_string()]);
-
        let index = configs_iter.next().unwrap();
        assert_eq!(index.index_type, crate::index::IndexType::Bitmap);
        assert_eq!(index.columns, vec!["is_active".to_string()]);

        let index = configs_iter.next().unwrap();
        assert_eq!(index.index_type, crate::index::IndexType::Bitmap);
-        assert_eq!(index.columns, vec!["large_category".to_string()]);
+        assert_eq!(index.columns, vec!["data".to_string()]);

        let index = configs_iter.next().unwrap();
        assert_eq!(index.index_type, crate::index::IndexType::Bitmap);
        assert_eq!(index.columns, vec!["large_data".to_string()]);
+
+        let index = configs_iter.next().unwrap();
+        assert_eq!(index.index_type, crate::index::IndexType::Bitmap);
+        assert_eq!(index.columns, vec!["large_category".to_string()]);
    }

    #[tokio::test]
--- a/rust/lancedb/src/table/dataset.rs
+++ b/rust/lancedb/src/table/dataset.rs
@@ -8,7 +8,6 @@ use std::{

 use lance::{Dataset, dataset::refs};

-use crate::table::merge::lsm::ShardWriterCache;
 use crate::{Error, error::Result, utils::background_cache::BackgroundCache};

 /// A wrapper around a [Dataset] that provides consistency checks.
@@ -19,10 +18,6 @@ use crate::{Error, error::Result, utils::background_cache::BackgroundCache};
 pub struct DatasetConsistencyWrapper {
    state: Arc<Mutex<DatasetState>>,
    consistency: ConsistencyMode,
-    /// The single MemWAL `ShardWriter` for this dataset, co-located so it is
-    /// cached for the session and shares the dataset's lifecycle. A dataset
-    /// writes to one shard at a time. Shared by `Arc` across clones.
-    shard_writer: Arc<ShardWriterCache>,
 }

 /// The current dataset and whether it is pinned to a specific version.
@@ -72,15 +67,9 @@ impl DatasetConsistencyWrapper {
                pinned_version: None,
            })),
            consistency,
-            shard_writer: Arc::new(ShardWriterCache::default()),
        }
    }

-    /// The MemWAL `ShardWriter` cache co-located with this dataset.
-    pub(crate) fn shard_writer(&self) -> &Arc<ShardWriterCache> {
-        &self.shard_writer
-    }
-
    /// Get the current dataset.
    ///
    /// Behavior depends on the consistency mode:
--- a/rust/lancedb/src/table/merge.rs
+++ b/rust/lancedb/src/table/merge.rs
@@ -41,16 +41,6 @@ pub struct MergeResult {
    /// A value of 1 means the operation succeeded on the first try.
    #[serde(default)]
    pub num_attempts: u32,
-    /// Total number of rows written.
-    ///
-    /// On the standard `merge_insert` path this equals
-    /// `num_inserted_rows + num_updated_rows`. On the MemWAL LSM write path the
-    /// insert/update breakdown is not known until compaction; in that mode
-    /// `num_inserted_rows`, `num_updated_rows`, `num_deleted_rows`, `version`
-    /// and `num_attempts` are all `0` and this field holds the total number of
-    /// rows written through the shard writer.
-    #[serde(default)]
-    pub num_rows: u64,
 }

 /// A builder used to create and run a merge insert operation
@@ -67,8 +57,6 @@ pub struct MergeInsertBuilder {
    pub(crate) when_not_matched_by_source_delete_filt: Option<String>,
    pub(crate) timeout: Option<Duration>,
    pub(crate) use_index: bool,
-    pub(crate) use_lsm_write: Option<bool>,
-    pub(crate) validate_single_shard: bool,
 }

 impl MergeInsertBuilder {
@@ -83,8 +71,6 @@ impl MergeInsertBuilder {
            when_not_matched_by_source_delete_filt: None,
            timeout: None,
            use_index: true,
-            use_lsm_write: None,
-            validate_single_shard: true,
        }
    }

@@ -164,34 +150,6 @@ impl MergeInsertBuilder {
        self
    }

-    /// Controls whether `merge_insert` uses the MemWAL LSM write path.
-    ///
-    /// By default (unset), a `merge_insert` on a table with an
-    /// [`LsmWriteSpec`](super::LsmWriteSpec) installed is routed through
-    /// Lance's MemWAL shard writer, and a table without one uses the standard
-    /// path. Calling this with `false` forces the standard path even when a
-    /// spec is set. Calling it with `true` requires a spec — `merge_insert`
-    /// errors if none is installed.
-    pub fn use_lsm_write(&mut self, use_lsm_write: bool) -> &mut Self {
-        self.use_lsm_write = Some(use_lsm_write);
-        self
-    }
-
-    /// Controls how an LSM `merge_insert` checks that its input targets a
-    /// single shard.
-    ///
-    /// When a table has an LSM write spec, every row in a `merge_insert` call
-    /// must route to the same shard. When `true` (the default), every row is
-    /// inspected to verify this. When `false`, only the first row is inspected
-    /// and the shard it routes to is used for the whole input — a faster path
-    /// for callers that have already pre-sharded their input.
-    ///
-    /// Has no effect on tables without an LSM write spec.
-    pub fn validate_single_shard(&mut self, validate_single_shard: bool) -> &mut Self {
-        self.validate_single_shard = validate_single_shard;
-        self
-    }
-
    /// Executes the merge insert operation
    ///
    /// Returns version and statistics about the merge operation including the number of rows
@@ -209,23 +167,6 @@ pub(crate) async fn execute_merge_insert(
    params: MergeInsertBuilder,
    new_data: Box<dyn RecordBatchReader + Send>,
 ) -> Result<MergeResult> {
-    match lsm::lsm_dispatch_decision(table, &params).await? {
-        lsm::LsmDispatch::Lsm(plan) => {
-            let future =
-                lsm::execute_lsm_merge_insert(table, plan, params.validate_single_shard, new_data);
-            return match params.timeout {
-                Some(timeout) => match tokio::time::timeout(timeout, future).await {
-                    Ok(result) => result,
-                    Err(_) => Err(Error::Runtime {
-                        message: "merge insert timed out".to_string(),
-                    }),
-                },
-                None => future.await,
-            };
-        }
-        lsm::LsmDispatch::Standard => {}
-    }
-
    let dataset = table.dataset.get().await?;
    let mut builder = LanceMergeInsertBuilder::try_new(dataset.clone(), params.on)?;
    match (
@@ -278,7 +219,6 @@ pub(crate) async fn execute_merge_insert(
        num_inserted_rows: stats.num_inserted_rows,
        num_deleted_rows: stats.num_deleted_rows,
        num_attempts: stats.num_attempts,
-        num_rows: stats.num_inserted_rows + stats.num_updated_rows,
    })
 }

@@ -387,366 +327,3 @@ mod tests {
        assert_eq!(table.count_rows(None).await.unwrap(), 25);
    }
 }
-
-#[cfg(test)]
-mod lsm_tests {
-    use std::sync::Arc;
-
-    use arrow_array::{
-        Int64Array, RecordBatch, RecordBatchIterator, RecordBatchReader, StringArray,
-    };
-    use arrow_schema::{DataType, Field, Schema};
-    use tempfile::{TempDir, tempdir};
-
-    use crate::connect;
-    use crate::error::Error;
-    use crate::table::{LsmWriteSpec, Table};
-
-    /// A reader of `[id: Int64, value: Int64]` rows; `value` is `0..n`.
-    fn id_value_reader(ids: Vec<i64>) -> Box<dyn RecordBatchReader + Send> {
-        let schema = Arc::new(Schema::new(vec![
-            Field::new("id", DataType::Int64, false),
-            Field::new("value", DataType::Int64, false),
-        ]));
-        let n = ids.len() as i64;
-        let batch = RecordBatch::try_new(
-            schema.clone(),
-            vec![
-                Arc::new(Int64Array::from(ids)),
-                Arc::new(Int64Array::from_iter_values(0..n)),
-            ],
-        )
-        .unwrap();
-        Box::new(RecordBatchIterator::new(vec![Ok(batch)], schema))
-    }
-
-    /// A reader of `[id: Int64, region: Utf8]` rows.
-    fn id_region_reader(rows: Vec<(i64, &str)>) -> Box<dyn RecordBatchReader + Send> {
-        let schema = Arc::new(Schema::new(vec![
-            Field::new("id", DataType::Int64, false),
-            Field::new("region", DataType::Utf8, false),
-        ]));
-        let ids: Vec<i64> = rows.iter().map(|(id, _)| *id).collect();
-        let regions: Vec<&str> = rows.iter().map(|(_, region)| *region).collect();
-        let batch = RecordBatch::try_new(
-            schema.clone(),
-            vec![
-                Arc::new(Int64Array::from(ids)),
-                Arc::new(StringArray::from(regions)),
-            ],
-        )
-        .unwrap();
-        Box::new(RecordBatchIterator::new(vec![Ok(batch)], schema))
-    }
-
-    /// A multi-batch reader of `[id: Int64, region: Utf8]` rows.
-    fn id_region_multi_reader(batches: Vec<Vec<(i64, &str)>>) -> Box<dyn RecordBatchReader + Send> {
-        let schema = Arc::new(Schema::new(vec![
-            Field::new("id", DataType::Int64, false),
-            Field::new("region", DataType::Utf8, false),
-        ]));
-        let records: Vec<_> = batches
-            .into_iter()
-            .map(|rows| {
-                let ids: Vec<i64> = rows.iter().map(|(id, _)| *id).collect();
-                let regions: Vec<&str> = rows.iter().map(|(_, region)| *region).collect();
-                Ok(RecordBatch::try_new(
-                    schema.clone(),
-                    vec![
-                        Arc::new(Int64Array::from(ids)),
-                        Arc::new(StringArray::from(regions)),
-                    ],
-                )
-                .unwrap())
-            })
-            .collect();
-        Box::new(RecordBatchIterator::new(records, schema))
-    }
-
-    /// Create an `[id, value]` table with `id` as the unenforced primary key.
-    async fn id_value_table(dir: &TempDir) -> Table {
-        let conn = connect(dir.path().to_str().unwrap())
-            .execute()
-            .await
-            .unwrap();
-        let table = conn
-            .create_table("t", id_value_reader(vec![1, 2, 3]))
-            .execute()
-            .await
-            .unwrap();
-        table.set_unenforced_primary_key(["id"]).await.unwrap();
-        table
-    }
-
-    #[tokio::test]
-    async fn lsm_merge_insert_bucket() {
-        let dir = tempdir().unwrap();
-        let table = id_value_table(&dir).await;
-        // num_buckets = 1: every row routes to the single bucket.
-        table
-            .set_lsm_write_spec(LsmWriteSpec::bucket("id", 1))
-            .await
-            .unwrap();
-
-        // Empty `on` defaults to the primary key.
-        let mut builder = table.merge_insert(&[]);
-        builder
-            .when_matched_update_all(None)
-            .when_not_matched_insert_all();
-        let result = builder
-            .execute(id_value_reader(vec![3, 4, 5]))
-            .await
-            .unwrap();
-
-        // LSM path: rows go to the MemWAL, the breakdown is unknown until
-        // compaction, so only `num_rows` is populated.
-        assert_eq!(result.num_rows, 3);
-        assert_eq!(result.version, 0);
-        assert_eq!(result.num_inserted_rows, 0);
-        assert_eq!(result.num_updated_rows, 0);
-    }
-
-    #[tokio::test]
-    async fn lsm_merge_insert_unsharded() {
-        let dir = tempdir().unwrap();
-        let table = id_value_table(&dir).await;
-        table
-            .set_lsm_write_spec(LsmWriteSpec::unsharded())
-            .await
-            .unwrap();
-
-        let mut builder = table.merge_insert(&["id"]);
-        builder
-            .when_matched_update_all(None)
-            .when_not_matched_insert_all();
-        let result = builder
-            .execute(id_value_reader(vec![10, 11, 12, 13]))
-            .await
-            .unwrap();
-        assert_eq!(result.num_rows, 4);
-    }
-
-    #[tokio::test]
-    async fn lsm_merge_insert_identity() {
-        let dir = tempdir().unwrap();
-        let conn = connect(dir.path().to_str().unwrap())
-            .execute()
-            .await
-            .unwrap();
-        let table = conn
-            .create_table("t", id_region_reader(vec![(1, "us"), (2, "us")]))
-            .execute()
-            .await
-            .unwrap();
-        table.set_unenforced_primary_key(["id"]).await.unwrap();
-        table
-            .set_lsm_write_spec(LsmWriteSpec::identity("region"))
-            .await
-            .unwrap();
-
-        // All rows share one identity value, so they route to one shard.
-        let mut builder = table.merge_insert(&[]);
-        builder
-            .when_matched_update_all(None)
-            .when_not_matched_insert_all();
-        let result = builder
-            .execute(id_region_reader(vec![(3, "us"), (4, "us")]))
-            .await
-            .unwrap();
-        assert_eq!(result.num_rows, 2);
-    }
-
-    #[tokio::test]
-    async fn lsm_merge_insert_use_lsm_write_false_falls_back() {
-        let dir = tempdir().unwrap();
-        let table = id_value_table(&dir).await;
-        table
-            .set_lsm_write_spec(LsmWriteSpec::bucket("id", 1))
-            .await
-            .unwrap();
-
-        // use_lsm_write(false) opts out: the standard path runs and commits.
-        let mut builder = table.merge_insert(&["id"]);
-        builder.when_not_matched_insert_all().use_lsm_write(false);
-        let result = builder
-            .execute(id_value_reader(vec![3, 4, 5]))
-            .await
-            .unwrap();
-
-        assert_eq!(result.num_inserted_rows, 2);
-        assert_eq!(table.count_rows(None).await.unwrap(), 5);
-    }
-
-    #[tokio::test]
-    async fn lsm_merge_insert_rejects_on_not_primary_key() {
-        let dir = tempdir().unwrap();
-        let table = id_value_table(&dir).await;
-        table
-            .set_lsm_write_spec(LsmWriteSpec::bucket("id", 1))
-            .await
-            .unwrap();
-
-        let mut builder = table.merge_insert(&["value"]);
-        builder
-            .when_matched_update_all(None)
-            .when_not_matched_insert_all();
-        let err = builder.execute(id_value_reader(vec![1])).await.unwrap_err();
-        assert!(matches!(err, Error::InvalidInput { .. }), "got {err:?}");
-    }
-
-    #[tokio::test]
-    async fn lsm_merge_insert_rejects_non_upsert() {
-        let dir = tempdir().unwrap();
-        let table = id_value_table(&dir).await;
-        table
-            .set_lsm_write_spec(LsmWriteSpec::bucket("id", 1))
-            .await
-            .unwrap();
-
-        // Insert-only (no when_matched_update_all) is not the upsert shape.
-        let mut builder = table.merge_insert(&[]);
-        builder.when_not_matched_insert_all();
-        let err = builder.execute(id_value_reader(vec![4])).await.unwrap_err();
-        assert!(matches!(err, Error::InvalidInput { .. }), "got {err:?}");
-    }
-
-    #[tokio::test]
-    async fn lsm_close_writers_then_reopen() {
-        let dir = tempdir().unwrap();
-        let table = id_value_table(&dir).await;
-        table
-            .set_lsm_write_spec(LsmWriteSpec::bucket("id", 1))
-            .await
-            .unwrap();
-
-        let mut builder = table.merge_insert(&[]);
-        builder
-            .when_matched_update_all(None)
-            .when_not_matched_insert_all();
-        builder.execute(id_value_reader(vec![7, 8])).await.unwrap();
-
-        table.close_lsm_writers().await.unwrap();
-
-        // The writer reopens lazily on the next merge_insert.
-        let mut builder = table.merge_insert(&[]);
-        builder
-            .when_matched_update_all(None)
-            .when_not_matched_insert_all();
-        let result = builder.execute(id_value_reader(vec![9])).await.unwrap();
-        assert_eq!(result.num_rows, 1);
-    }
-
-    #[tokio::test]
-    async fn lsm_merge_insert_multi_batch() {
-        let dir = tempdir().unwrap();
-        let conn = connect(dir.path().to_str().unwrap())
-            .execute()
-            .await
-            .unwrap();
-        let table = conn
-            .create_table("t", id_region_reader(vec![(1, "us")]))
-            .execute()
-            .await
-            .unwrap();
-        table.set_unenforced_primary_key(["id"]).await.unwrap();
-        table
-            .set_lsm_write_spec(LsmWriteSpec::identity("region"))
-            .await
-            .unwrap();
-
-        // Multiple batches that all route to one shard are written together.
-        let mut builder = table.merge_insert(&[]);
-        builder
-            .when_matched_update_all(None)
-            .when_not_matched_insert_all();
-        let result = builder
-            .execute(id_region_multi_reader(vec![
-                vec![(2, "us"), (3, "us")],
-                vec![(4, "us")],
-            ]))
-            .await
-            .unwrap();
-        assert_eq!(result.num_rows, 3);
-
-        // Batches that route to different shards are rejected; the validation
-        // runs before any write, so no partial write is left behind.
-        let mut builder = table.merge_insert(&[]);
-        builder
-            .when_matched_update_all(None)
-            .when_not_matched_insert_all();
-        let err = builder
-            .execute(id_region_multi_reader(vec![
-                vec![(5, "us")],
-                vec![(6, "eu")],
-            ]))
-            .await
-            .unwrap_err();
-        assert!(matches!(err, Error::InvalidInput { .. }), "got {err:?}");
-    }
-
-    #[tokio::test]
-    async fn lsm_merge_insert_use_lsm_write_true_requires_spec() {
-        let dir = tempdir().unwrap();
-        // id_value_table sets a primary key but no LSM write spec.
-        let table = id_value_table(&dir).await;
-
-        let mut builder = table.merge_insert(&["id"]);
-        builder
-            .when_matched_update_all(None)
-            .when_not_matched_insert_all()
-            .use_lsm_write(true);
-        let err = builder.execute(id_value_reader(vec![4])).await.unwrap_err();
-        assert!(matches!(err, Error::InvalidInput { .. }), "got {err:?}");
-    }
-
-    #[tokio::test]
-    async fn lsm_merge_insert_rejects_second_shard() {
-        let dir = tempdir().unwrap();
-        let conn = connect(dir.path().to_str().unwrap())
-            .execute()
-            .await
-            .unwrap();
-        let table = conn
-            .create_table("t", id_region_reader(vec![(1, "us")]))
-            .execute()
-            .await
-            .unwrap();
-        table.set_unenforced_primary_key(["id"]).await.unwrap();
-        table
-            .set_lsm_write_spec(LsmWriteSpec::identity("region"))
-            .await
-            .unwrap();
-
-        // The first merge_insert opens the single writer for shard "us".
-        let mut builder = table.merge_insert(&[]);
-        builder
-            .when_matched_update_all(None)
-            .when_not_matched_insert_all();
-        builder
-            .execute(id_region_reader(vec![(2, "us")]))
-            .await
-            .unwrap();
-
-        // A merge_insert routing to a different shard is rejected.
-        let mut builder = table.merge_insert(&[]);
-        builder
-            .when_matched_update_all(None)
-            .when_not_matched_insert_all();
-        let err = builder
-            .execute(id_region_reader(vec![(3, "eu")]))
-            .await
-            .unwrap_err();
-        assert!(matches!(err, Error::InvalidInput { .. }), "got {err:?}");
-
-        // After closing the writer, a different shard can be written.
-        table.close_lsm_writers().await.unwrap();
-        let mut builder = table.merge_insert(&[]);
-        builder
-            .when_matched_update_all(None)
-            .when_not_matched_insert_all();
-        builder
-            .execute(id_region_reader(vec![(4, "eu")]))
-            .await
-            .unwrap();
-    }
-}
--- a/rust/lancedb/src/table/merge/lsm.rs
+++ b/rust/lancedb/src/table/merge/lsm.rs