feat: pause and resume jobs from the Python SDK

Cancellation is the only job control the SDK exposes, and it is terminal, so a long-running server-side job cannot be parked and picked up again. This adds pause_job and resume_job to the Rust core connection and the Python bindings (sync and async), posting to the server's /v1/jobs/pause and /v1/jobs/resume endpoints. A pause parks the job until it is resumed: its workers drain and stop. The outcome strings mirror the server's answers -- a job finalizing its results reports "committing" and cannot be parked, and a resume before the drain is confirmed reports "still_pausing"; both are retried rather than failed. Resuming re-queues the job and its workers pick their work back up from checkpoints. Local connections report the operations as unsupported, like the rest of the jobs API.
2026-09-11 15:52:17 +00:00 · 2026-09-01 13:23:13 +00:00
55 changed files with 956 additions and 2204 deletions
@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "0.39.0-beta.1"
+current_version = "0.38.0"
 parse = """(?x)
    (?P<major>0|[1-9]\\d*)\\.
    (?P<minor>0|[1-9]\\d*)\\.
@@ -3455,8 +3455,8 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"

 [[package]]
 name = "fsst"
-version = "12.0.0-beta.11"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
+version = "12.0.0-beta.5"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
 dependencies = [
 "arrow-array",
 "rand 0.9.5",
@@ -4815,8 +4815,8 @@ checksum = "e037a2e1d8d5fdbd49b16a4ea09d5d6401c1f29eca5ff29d03d3824dba16256a"

 [[package]]
 name = "lance"
-version = "12.0.0-beta.11"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
+version = "12.0.0-beta.5"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
 dependencies = [
 "arc-swap",
 "arrow",
@@ -4888,8 +4888,8 @@ dependencies = [

 [[package]]
 name = "lance-arrow"
-version = "12.0.0-beta.11"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
+version = "12.0.0-beta.5"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4911,7 +4911,7 @@ dependencies = [
 [[package]]
 name = "lance-arrow-scalar"
 version = "58.0.0"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4925,7 +4925,7 @@ dependencies = [
 [[package]]
 name = "lance-arrow-stats"
 version = "58.0.0"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -4934,8 +4934,8 @@ dependencies = [

 [[package]]
 name = "lance-bitpacking"
-version = "12.0.0-beta.11"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
+version = "12.0.0-beta.5"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
 dependencies = [
 "arrayref",
 "crunchy",
@@ -4945,8 +4945,8 @@ dependencies = [

 [[package]]
 name = "lance-core"
-version = "12.0.0-beta.11"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
+version = "12.0.0-beta.5"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4983,8 +4983,8 @@ dependencies = [

 [[package]]
 name = "lance-datafusion"
-version = "12.0.0-beta.11"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
+version = "12.0.0-beta.5"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5000,7 +5000,6 @@ dependencies = [
 "datafusion-functions",
 "datafusion-physical-expr",
 "futures",
- "half",
 "jsonb",
 "lance-arrow",
 "lance-core",
@@ -5014,8 +5013,8 @@ dependencies = [

 [[package]]
 name = "lance-datagen"
-version = "12.0.0-beta.11"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
+version = "12.0.0-beta.5"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5032,8 +5031,8 @@ dependencies = [

 [[package]]
 name = "lance-derive"
-version = "12.0.0-beta.11"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
+version = "12.0.0-beta.5"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
 dependencies = [
 "proc-macro2",
 "quote",
@@ -5042,8 +5041,8 @@ dependencies = [

 [[package]]
 name = "lance-encoding"
-version = "12.0.0-beta.11"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
+version = "12.0.0-beta.5"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
 dependencies = [
 "arrow-arith",
 "arrow-array",
@@ -5076,8 +5075,8 @@ dependencies = [

 [[package]]
 name = "lance-file"
-version = "12.0.0-beta.11"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
+version = "12.0.0-beta.5"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
 dependencies = [
 "arrow-arith",
 "arrow-array",
@@ -5108,8 +5107,8 @@ dependencies = [

 [[package]]
 name = "lance-index"
-version = "12.0.0-beta.11"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
+version = "12.0.0-beta.5"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
 dependencies = [
 "arc-swap",
 "arrow",
@@ -5173,8 +5172,8 @@ dependencies = [

 [[package]]
 name = "lance-index-core"
-version = "12.0.0-beta.11"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
+version = "12.0.0-beta.5"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -5196,8 +5195,8 @@ dependencies = [

 [[package]]
 name = "lance-io"
-version = "12.0.0-beta.11"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
+version = "12.0.0-beta.5"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5237,8 +5236,8 @@ dependencies = [

 [[package]]
 name = "lance-linalg"
-version = "12.0.0-beta.11"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
+version = "12.0.0-beta.5"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -5252,8 +5251,8 @@ dependencies = [

 [[package]]
 name = "lance-namespace"
-version = "12.0.0-beta.11"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
+version = "12.0.0-beta.5"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
 dependencies = [
 "arrow",
 "async-trait",
@@ -5265,8 +5264,8 @@ dependencies = [

 [[package]]
 name = "lance-namespace-impls"
-version = "12.0.0-beta.11"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
+version = "12.0.0-beta.5"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
 dependencies = [
 "arrow",
 "arrow-ipc",
@@ -5305,9 +5304,9 @@ dependencies = [

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

 [[package]]
 name = "lance-select"
-version = "12.0.0-beta.11"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
+version = "12.0.0-beta.5"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -5334,8 +5333,8 @@ dependencies = [

 [[package]]
 name = "lance-table"
-version = "12.0.0-beta.11"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
+version = "12.0.0-beta.5"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5375,8 +5374,8 @@ dependencies = [

 [[package]]
 name = "lance-testing"
-version = "12.0.0-beta.11"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
+version = "12.0.0-beta.5"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -5389,8 +5388,8 @@ dependencies = [

 [[package]]
 name = "lance-tokenizer"
-version = "12.0.0-beta.11"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
+version = "12.0.0-beta.5"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
 dependencies = [
 "frostem",
 "icu_segmenter",
@@ -5403,7 +5402,7 @@ dependencies = [

 [[package]]
 name = "lancedb"
-version = "0.39.0-beta.1"
+version = "0.38.0"
 dependencies = [
 "ahash",
 "anyhow",
@@ -5491,7 +5490,7 @@ dependencies = [

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

 [[package]]
 name = "lancedb-python"
-version = "0.39.0-beta.1"
+version = "0.38.0"
 dependencies = [
 "arrow",
 "async-trait",
@@ -13,20 +13,20 @@ categories = ["database-implementations"]
 rust-version = "1.91.0"

 [workspace.dependencies]
-lance = { "version" = "=12.0.0-beta.11", default-features = false, "tag" = "v12.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
-lance-core = { "version" = "=12.0.0-beta.11", "tag" = "v12.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
-lance-datagen = { "version" = "=12.0.0-beta.11", "tag" = "v12.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
-lance-file = { "version" = "=12.0.0-beta.11", "tag" = "v12.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
-lance-io = { "version" = "=12.0.0-beta.11", default-features = false, "tag" = "v12.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
-lance-index = { "version" = "=12.0.0-beta.11", "tag" = "v12.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
-lance-linalg = { "version" = "=12.0.0-beta.11", "tag" = "v12.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace = { "version" = "=12.0.0-beta.11", "tag" = "v12.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace-impls = { "version" = "=12.0.0-beta.11", default-features = false, "tag" = "v12.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
-lance-table = { "version" = "=12.0.0-beta.11", "tag" = "v12.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
-lance-testing = { "version" = "=12.0.0-beta.11", "tag" = "v12.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
-lance-datafusion = { "version" = "=12.0.0-beta.11", "tag" = "v12.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
-lance-encoding = { "version" = "=12.0.0-beta.11", "tag" = "v12.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
-lance-arrow = { "version" = "=12.0.0-beta.11", "tag" = "v12.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
+lance = { "version" = "=12.0.0-beta.5", default-features = false, "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
+lance-core = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
+lance-datagen = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
+lance-file = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
+lance-io = { "version" = "=12.0.0-beta.5", default-features = false, "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
+lance-index = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
+lance-linalg = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace-impls = { "version" = "=12.0.0-beta.5", default-features = false, "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
+lance-table = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
+lance-testing = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
+lance-datafusion = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
+lance-encoding = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
+lance-arrow = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
 lancedb = { path = "rust/lancedb", default-features = false }
 ahash = "0.8"
 # Note that this one does not include pyarrow
@@ -446,15 +446,6 @@ paths:
              properties:
                column:
                  type: string
-                name:
-                  type: string
-                  description: Optional name for the created index.
-                replace:
-                  type: boolean
-                  default: true
-                  description: |
-                    Whether to replace an existing index with the same resolved
-                    name. Defaults to true.
                metric_type:
                  type: string
                  nullable: false
@@ -14,7 +14,7 @@ Add the following dependency to your `pom.xml`:
 <dependency>
    <groupId>com.lancedb</groupId>
    <artifactId>lancedb-core</artifactId>
-    <version>0.39.0-beta.1</version>
+    <version>0.38.0</version>
 </dependency>
 ```

@@ -676,17 +676,9 @@ List all the versions of the table
 abstract mergeInsert(on): MergeInsertBuilder
 ```

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

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

 #### Returns

@@ -50,7 +50,7 @@ If this is false, and another index already exists on the same columns
 and the same name, then an error will be returned.  This is true even if
 that index is out of date.

-The default is false
+The default is true

 ***

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

 ***

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

 ```ts
@@ -8,7 +8,7 @@
    <parent>
      <groupId>com.lancedb</groupId>
      <artifactId>lancedb-parent</artifactId>
-      <version>0.39.0-beta.1</version>
+      <version>0.38.0-final.0</version>
      <relativePath>../pom.xml</relativePath>
    </parent>

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

    <groupId>com.lancedb</groupId>
    <artifactId>lancedb-parent</artifactId>
-    <version>0.39.0-beta.1</version>
+    <version>0.38.0-final.0</version>
    <packaging>pom</packaging>
    <name>${project.artifactId}</name>
    <description>LanceDB Java SDK Parent POM</description>
@@ -28,7 +28,7 @@
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <arrow.version>15.0.0</arrow.version>
-        <lance-core.version>12.0.0-beta.11</lance-core.version>
+        <lance-core.version>12.0.0-beta.5</lance-core.version>
        <spotless.skip>false</spotless.skip>
        <spotless.version>2.30.0</spotless.version>
        <spotless.java.googlejavaformat.version>1.7</spotless.java.googlejavaformat.version>
@@ -1,7 +1,7 @@
 [package]
 name = "lancedb-nodejs"
 edition.workspace = true
-version = "0.39.0-beta.1"
+version = "0.38.0"
 publish = false
 license.workspace = true
 description.workspace = true
@@ -48,28 +48,6 @@ describe("materialized views", () => {
    expect(definitionFromMetadata(safe, "v").limit).toBe(42);
  });

-  it("reads the namespaced select kind and refuses unknown kinds", () => {
-    // "namespaced_select" is the namespaced form of "select": same shape, a
-    // separate kind so readers that predate it refuse instead of resolving
-    // the source at the root.
-    const namespaced = new Map([
-      [
-        DEFINITION_META_KEY,
-        '{"kind":"namespaced_select","source_table":"people","source_namespace":["ns"]}',
-      ],
-    ]);
-    const definition = definitionFromMetadata(namespaced, "v");
-    expect(definition.sourceTable).toBe("people");
-    expect(definition.sourceNamespace).toEqual(["ns"]);
-
-    const unknown = new Map([
-      [DEFINITION_META_KEY, '{"kind":"select_v3","source_table":"people"}'],
-    ]);
-    expect(() => definitionFromMetadata(unknown, "v")).toThrow(
-      /cannot refresh/,
-    );
-  });
-
  it("creates, refreshes and queries a view", async () => {
    const view = await db.createMaterializedView("adults", "people", {
      select: ["name", ["shout", "upper(name)"]],
@@ -737,12 +737,11 @@ it("should query documents with LangChain PDF metadata", async () => {

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

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

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

    expect(result.map((row) => ({ ...row }))).toEqual(expected);
  });
-  test("upsert on a composite key", async () => {
-    const composite = await conn.createTable("composite", [
-      { shard: "a", id: 1, val: "x" },
-      { shard: "a", id: 2, val: "y" },
-      { shard: "b", id: 1, val: "z" },
-    ]);
-
-    // ("a", 1) matches an existing row and updates it. ("b", 2) agrees with an
-    // existing row on each key column separately but on neither pair, so it is
-    // an insert.
-    const mergeInsertRes = await composite
-      .mergeInsert(["shard", "id"])
-      .whenMatchedUpdateAll()
-      .whenNotMatchedInsertAll()
-      .execute([
-        { shard: "a", id: 1, val: "X" },
-        { shard: "b", id: 2, val: "W" },
-      ]);
-    expect(mergeInsertRes.numUpdatedRows).toBe(1);
-    expect(mergeInsertRes.numInsertedRows).toBe(1);
-
-    const result = (await composite.toArrow())
-      .toArray()
-      .sort((a, b) => a.shard.localeCompare(b.shard) || a.id - b.id);
-
-    expect(result.map((row) => ({ ...row }))).toEqual([
-      { shard: "a", id: 1, val: "X" },
-      { shard: "a", id: 2, val: "y" },
-      { shard: "b", id: 1, val: "z" },
-      { shard: "b", id: 2, val: "W" },
-    ]);
-  });
  test("conditional update", async () => {
    const newData = [
      { a: 2, b: "x" },
@@ -1526,7 +1493,11 @@ describe("When creating an index", () => {

  it("should allow me to replace (or not) an existing index", async () => {
    await tbl.createIndex("id");
-    await expect(tbl.createIndex("id")).rejects.toThrow("already exists");
+    // Default is replace=true
+    await tbl.createIndex("id");
+    await expect(tbl.createIndex("id", { replace: false })).rejects.toThrow(
+      "already exists",
+    );
    await tbl.createIndex("id", { replace: true });
  });

@@ -841,7 +841,7 @@ export interface IndexOptions {
   * and the same name, then an error will be returned.  This is true even if
   * that index is out of date.
   *
-   * The default is false
+   * The default is true
   */
  replace?: boolean;

@@ -19,8 +19,6 @@ export interface MaterializedViewDefinition {
  limit?: number;
  /** Source columns the projections and filter read. */
  inputs: string[];
-  /** Namespace holding the source table; empty is the root namespace. */
-  sourceNamespace: string[];
 }

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

@@ -919,16 +919,6 @@ export abstract class Table {
  /** Return the table as an arrow table */
  abstract toArrow(): Promise<ArrowTable>;

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

  /** List all the stats of a specified index
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-darwin-arm64",
-	"version": "0.39.0-beta.1",
+	"version": "0.38.0",
 	"os": ["darwin"],
 	"cpu": ["arm64"],
 	"main": "lancedb.darwin-arm64.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-gnu",
-	"version": "0.39.0-beta.1",
+	"version": "0.38.0",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-gnu.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-musl",
-	"version": "0.39.0-beta.1",
+	"version": "0.38.0",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-musl.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-gnu",
-	"version": "0.39.0-beta.1",
+	"version": "0.38.0",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-gnu.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-musl",
-	"version": "0.39.0-beta.1",
+	"version": "0.38.0",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-musl.node",
@@ -1,6 +1,6 @@
 {
  "name": "@lancedb/lancedb-win32-arm64-msvc",
-  "version": "0.39.0-beta.1",
+  "version": "0.38.0",
  "os": [
    "win32"
  ],
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-win32-x64-msvc",
-	"version": "0.39.0-beta.1",
+	"version": "0.38.0",
 	"os": ["win32"],
 	"cpu": ["x64"],
 	"main": "lancedb.win32-x64-msvc.node",
@@ -11,7 +11,7 @@
    "ann"
  ],
  "private": false,
-  "version": "0.39.0-beta.1",
+  "version": "0.38.0",
  "main": "dist/index.js",
  "exports": {
    ".": "./dist/index.js",
@@ -44,7 +44,7 @@
    "@biomejs/biome": "^1.7.3",
    "@jest/globals": "^29.7.0",
    "@napi-rs/cli": "3.7.0",
-    "@opentelemetry/sdk-metrics": "^2.10.0",
+    "@opentelemetry/sdk-metrics": "^1.30.0",
    "@types/axios": "^0.14.0",
    "@types/jest": "^29.1.2",
    "@types/node": "22.7.4",
@@ -56,7 +56,7 @@
    "eslint": "^8.57.0",
    "jest": "^29.7.0",
    "shx": "^0.3.4",
-    "tmp": "^0.2.7",
+    "tmp": "^0.2.3",
    "ts-jest": "^29.1.2",
    "typedoc": "0.26.4",
    "typedoc-plugin-markdown": "4.2.1",
@@ -16,41 +16,3 @@ allowBuilds:
  onnxruntime-node: true
  protobufjs: true
  sharp: true
-
-minimumReleaseAgeExclude:
-  - protobufjs@7.5.8
-  - tmp@0.2.6
-  - form-data@4.0.6
-  - tar@7.5.16
-  - markdown-it@14.1.2
-  - linkify-it@5.0.1
-  - js-yaml@3.15.0
-  - js-yaml@4.1.2
-  - protobufjs@7.6.1
-  - protobufjs@7.6.3
-  - '@babel/core@7.29.1'
-  - axios@1.18.0
-  - brace-expansion@2.1.2
-  - brace-expansion@1.1.16
-  - js-yaml@4.3.0
-  - tar@7.5.18
-  - tar@7.5.19
-  - tar@7.5.17
-  - protobufjs@7.6.5
-  - linkify-it@5.0.2
-  - sharp@0.35.0
-  - brace-expansion@1.1.17
-  - brace-expansion@2.1.3
-  - brace-expansion@2.1.4
-  - brace-expansion@1.1.18
-  - js-yaml@3.15.1
-  - js-yaml@4.3.1
-  - tar@7.5.21
-  - '@opentelemetry/core@2.8.0'
-
-# @huggingface/transformers pins sharp ^0.33.5 and no released version has moved
-# past ^0.34.5, all of which inherit the libvips CVEs in GHSA-f88m-g3jw-g9cj.
-# Force the patched line. sharp is only reached by transformers' image pipeline,
-# which LanceDB's text embedding function never uses.
-overrides:
-  sharp: ^0.35.4
@@ -1,6 +1,6 @@
 [package]
 name = "lancedb-python"
-version = "0.39.0-beta.1"
+version = "0.38.0"
 publish = false
 edition.workspace = true
 description = "Python bindings for LanceDB"
@@ -150,11 +150,12 @@ class Connection(object):
    def job(self, job_id: str) -> Job: ...
    async def create_function_async(self, request_json: str) -> Job: ...
    async def get_function(self, name: str, version: str) -> str: ...
-    async def list_functions(self) -> List[str]: ...
    async def drop_function(self, name: str, version: str) -> bool: ...
    async def list_jobs(self) -> List[JobInfo]: ...
    async def get_job(self, job_id: str) -> Optional[JobDescription]: ...
    async def cancel_job(self, job_id: str) -> bool: ...
+    async def pause_job(self, job_id: str) -> str: ...
+    async def resume_job(self, job_id: str) -> str: ...
    async def job_history(
        self, job_id: Optional[str] = None
    ) -> List[pa.RecordBatch]: ...
@@ -712,24 +712,6 @@ class DBConnection(EnforceOverrides):
            "Function catalog operations are not supported for this connection type"
        )

-    def list_functions(self) -> List[FunctionVersion]:
-        """List every published immutable Function version.
-
-        Results are ordered by Function name then version. Local connections
-        raise ``NotImplementedError``.
-
-        Examples
-        --------
-        List the identities available to use in Function-backed columns:
-
-        ```python
-        [(function.name, function.version) for function in db.list_functions()]
-        ```
-        """
-        raise NotImplementedError(
-            "Function catalog operations are not supported for this connection type"
-        )
-
    def drop_function(self, name: str, *, version: str) -> bool:
        """Drop one exact immutable Function version from the remote catalog.

@@ -771,6 +753,26 @@ class DBConnection(EnforceOverrides):
            "cancel_job is not supported for this connection type"
        )

+    def pause_job(self, job_id: str) -> str:
+        """Pause a server-side job by id.
+
+        The job's workers drain and it stays parked until resumed. Returns
+        "pausing", "already_paused", or "committing" -- a job finalizing its
+        results cannot be parked; retry shortly.
+        """
+        raise NotImplementedError("pause_job is not supported for this connection type")
+
+    def resume_job(self, job_id: str) -> str:
+        """Resume a paused server-side job by id.
+
+        Its workers pick their work back up from checkpoints. Returns
+        "resumed", "still_pausing" -- the pause's worker drain is not
+        confirmed yet; retry shortly -- or "not_paused".
+        """
+        raise NotImplementedError(
+            "resume_job is not supported for this connection type"
+        )
+
    def job_history(self, job_id: Optional[str] = None) -> List[pa.RecordBatch]:
        """The lifecycle event history of a server-side job, as Arrow batches.

@@ -1441,10 +1443,6 @@ class LanceDBConnection(DBConnection):
    def get_function(self, name: str, *, version: str) -> FunctionVersion:
        return LOOP.run(self._conn.get_function(name, version=version))

-    @override
-    def list_functions(self) -> List[FunctionVersion]:
-        return LOOP.run(self._conn.list_functions())
-
    @override
    def drop_function(self, name: str, *, version: str) -> bool:
        return LOOP.run(self._conn.drop_function(name, version=version))
@@ -1472,6 +1470,22 @@ class LanceDBConnection(DBConnection):
        """
        return LOOP.run(self._conn.cancel_job(job_id))

+    @override
+    def pause_job(self, job_id: str) -> str:
+        """Pause a server-side job by id.
+
+        Returns "pausing", "already_paused", or "committing".
+        """
+        return LOOP.run(self._conn.pause_job(job_id))
+
+    @override
+    def resume_job(self, job_id: str) -> str:
+        """Resume a paused server-side job by id.
+
+        Returns "resumed", "still_pausing", or "not_paused".
+        """
+        return LOOP.run(self._conn.resume_job(job_id))
+
    @override
    def job_history(self, job_id: Optional[str] = None) -> List[pa.RecordBatch]:
        """The lifecycle event history of a server-side job, as Arrow batches.
@@ -2279,17 +2293,6 @@ class AsyncConnection(object):
        """Open one exact immutable Function version from the remote catalog."""
        return FunctionVersion.from_json(await self._inner.get_function(name, version))

-    async def list_functions(self) -> List[FunctionVersion]:
-        """List every published immutable Function version.
-
-        Results are ordered by Function name then version. Local connections
-        raise ``NotImplementedError``.
-        """
-        return [
-            FunctionVersion.from_json(value)
-            for value in await self._inner.list_functions()
-        ]
-
    async def drop_function(self, name: str, *, version: str) -> bool:
        """Drop one exact immutable Function version from the remote catalog."""
        return await self._inner.drop_function(name, version)
@@ -2314,6 +2317,23 @@ class AsyncConnection(object):
        """
        return await self._inner.cancel_job(job_id)

+    async def pause_job(self, job_id: str) -> str:
+        """Pause a server-side job by id.
+
+        The job's workers drain and it stays parked until resumed. Returns
+        "pausing", "already_paused", or "committing" -- a job finalizing its
+        results cannot be parked; retry shortly.
+        """
+        return await self._inner.pause_job(job_id)
+
+    async def resume_job(self, job_id: str) -> str:
+        """Resume a paused server-side job by id.
+
+        Its workers pick their work back up from checkpoints. Returns
+        "resumed", "still_pausing" -- retry shortly -- or "not_paused".
+        """
+        return await self._inner.resume_job(job_id)
+
    async def job_history(self, job_id: Optional[str] = None) -> List[pa.RecordBatch]:
        """The lifecycle event history of a server-side job, as Arrow batches.

@@ -521,12 +521,6 @@ class RefreshColumnResult(_RemoteValue):

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


 _GRAMMAR_PRIMITIVES = (
@@ -597,19 +591,6 @@ def _validate_exact_arrow_field(field: pa.Field) -> None:
                "unsupported Arrow type for Function signature: lance.blob.v2 "
                f"requires a supported Blob storage layout, got {field}"
            )
-        metadata = {
-            (key.decode() if isinstance(key, bytes) else key): (
-                value.decode() if isinstance(value, bytes) else value
-            )
-            for key, value in (field.metadata or {}).items()
-        }
-        if metadata and metadata != {
-            _ARROW_EXTENSION_NAME_KEY: _BLOB_V2_EXTENSION_NAME
-        }:
-            raise TypeError(
-                "unsupported Arrow type for Function signature: lance.blob.v2 "
-                "field metadata must contain only its canonical extension marker"
-            )
    elif field.metadata:
        raise TypeError(
            "unsupported Arrow type for Function signature: field metadata "
@@ -674,84 +655,23 @@ def _canonical_arrow_field(field: pa.Field) -> str:
    return _canonical_arrow_type(field.type)


-def _blob_storage_type(field: pa.Field) -> pa.DataType:
-    data_type = field.type
-    if isinstance(data_type, pa.ExtensionType):
-        return data_type.storage_type
-    return data_type
-
-
-def _exact_blob_storage_type(field: pa.Field) -> dict[str, Any]:
-    storage = _blob_storage_type(field)
-    if not pa.types.is_struct(storage):
-        raise TypeError(
-            "unsupported Arrow type for Function signature: lance.blob.v2 "
-            "requires struct storage"
-        )
-    return {
-        "type": "struct",
-        "fields": [
-            {
-                "name": child.name,
-                "nullable": child.nullable,
-                "type": (
-                    {"type": "large_binary"}
-                    if pa.types.is_large_binary(child.type)
-                    else _exact_arrow_type(child.type)
-                ),
-            }
-            for child in storage
-        ],
-    }
-
-
-def _data_type_has_blob_v2(data_type: pa.DataType) -> bool:
-    if pa.types.is_struct(data_type):
-        return any(
-            _is_blob_v2_field(field) or _data_type_has_blob_v2(field.type)
-            for field in data_type
-        )
-    if (
-        pa.types.is_list(data_type)
-        or pa.types.is_large_list(data_type)
-        or pa.types.is_fixed_size_list(data_type)
-    ):
-        field = data_type.value_field
-        return _is_blob_v2_field(field) or _data_type_has_blob_v2(field.type)
-    if pa.types.is_map(data_type):
-        return any(
-            _is_blob_v2_field(field) or _data_type_has_blob_v2(field.type)
-            for field in (data_type.key_field, data_type.item_field)
-        )
-    return False
-
-
-def _exact_arrow_field(
-    field: pa.Field, *, inside_collection: bool = False
-) -> dict[str, Any]:
+def _exact_arrow_field(field: pa.Field) -> dict[str, Any]:
    _validate_exact_arrow_field(field)
    if _is_blob_v2_field(field):
-        if inside_collection:
-            raise TypeError(_NESTED_BLOB_COLLECTION_ERROR)
-        return {
-            "name": field.name,
-            "nullable": field.nullable,
-            "type": _exact_blob_storage_type(field),
-            "metadata": {
-                _ARROW_EXTENSION_NAME_KEY: _BLOB_V2_EXTENSION_NAME,
-            },
-        }
+        raise TypeError(
+            "unsupported Arrow type for Function signature: nested Blob v2 "
+            "fields are not supported; declare Blob parameters or named result "
+            "fields directly"
+        )
    value = {
        "name": field.name,
        "nullable": field.nullable,
-        "type": _exact_arrow_type(field.type, inside_collection=inside_collection),
+        "type": _exact_arrow_type(field.type),
    }
    return value


-def _exact_arrow_type(
-    data_type: pa.DataType, *, inside_collection: bool = False
-) -> dict[str, Any]:
+def _exact_arrow_type(data_type: pa.DataType) -> dict[str, Any]:
    for candidate, name in _GRAMMAR_PRIMITIVES:
        if data_type == candidate:
            return {"type": name}
@@ -765,10 +685,7 @@ def _exact_arrow_type(
            )
        return {
            "type": "struct",
-            "fields": [
-                _exact_arrow_field(field, inside_collection=inside_collection)
-                for field in fields
-            ],
+            "fields": [_exact_arrow_field(field) for field in fields],
        }
    if (
        pa.types.is_list(data_type)
@@ -793,15 +710,11 @@ def _exact_arrow_type(
                if pa.types.is_large_list(data_type)
                else "fixed_size_list"
            ),
-            "fields": [
-                _exact_arrow_field(data_type.value_field, inside_collection=True)
-            ],
+            "fields": [_exact_arrow_field(data_type.value_field)],
        }
        if pa.types.is_fixed_size_list(data_type):
            value["length"] = data_type.list_size
        return value
-    if pa.types.is_map(data_type) and _data_type_has_blob_v2(data_type):
-        raise TypeError(_NESTED_BLOB_COLLECTION_ERROR)
    raise TypeError(f"unsupported Arrow type for Function signature: {data_type}")


@@ -42,8 +42,6 @@ class MaterializedViewDefinition:
    """Cap on the number of rows the view holds."""
    inputs: List[str] = field(default_factory=list)
    """Source columns the projections and filter read."""
-    source_namespace: List[str] = field(default_factory=list)
-    """Namespace holding the source table; empty is the root namespace."""


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


@@ -749,10 +749,6 @@ class RemoteDBConnection(DBConnection):
    def get_function(self, name: str, *, version: str) -> FunctionVersion:
        return LOOP.run(self._conn.get_function(name, version=version))

-    @override
-    def list_functions(self) -> List[FunctionVersion]:
-        return LOOP.run(self._conn.list_functions())
-
    @override
    def drop_function(self, name: str, *, version: str) -> bool:
        return LOOP.run(self._conn.drop_function(name, version=version))
@@ -780,6 +776,22 @@ class RemoteDBConnection(DBConnection):
        """
        return LOOP.run(self._conn.cancel_job(job_id))

+    @override
+    def pause_job(self, job_id: str) -> str:
+        """Pause a server-side job by id.
+
+        Returns "pausing", "already_paused", or "committing".
+        """
+        return LOOP.run(self._conn.pause_job(job_id))
+
+    @override
+    def resume_job(self, job_id: str) -> str:
+        """Resume a paused server-side job by id.
+
+        Returns "resumed", "still_pausing", or "not_paused".
+        """
+        return LOOP.run(self._conn.resume_job(job_id))
+
    @override
    def job_history(self, job_id: Optional[str] = None) -> List[pa.RecordBatch]:
        """The lifecycle event history of a server-side job, as Arrow batches.
@@ -402,7 +402,6 @@ class RemoteTable(Table):
        /,
        *,
        config: IndexConfigType,
-        replace: bool = ...,
        wait_timeout: Optional[timedelta] = ...,
        name: Optional[str] = ...,
        train: bool = ...,
@@ -417,7 +416,7 @@ class RemoteTable(Table):
        index_cache_size: Optional[int] = ...,
        num_partitions: Optional[int] = ...,
        num_sub_vectors: Optional[int] = ...,
-        replace: bool = ...,
+        replace: Optional[bool] = ...,
        accelerator: Optional[str] = ...,
        index_type: Literal[
            "VECTOR", "IVF_FLAT", "IVF_SQ", "IVF_PQ", "IVF_HNSW_SQ", "IVF_HNSW_PQ"
@@ -436,7 +435,7 @@ class RemoteTable(Table):
        index_cache_size: Optional[int] = None,
        num_partitions: Optional[int] = None,
        num_sub_vectors: Optional[int] = None,
-        replace: bool = False,
+        replace: Optional[bool] = None,
        accelerator: Optional[str] = None,
        index_type="vector",
        wait_timeout: Optional[timedelta] = None,
@@ -480,6 +479,7 @@ class RemoteTable(Table):
            vector_column_name,
            accelerator,
            index_cache_size,
+            replace,
        )

        if is_legacy:
@@ -503,6 +503,12 @@ class RemoteTable(Table):
                    "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(
@@ -542,7 +548,6 @@ class RemoteTable(Table):
        LOOP.run(
            self._table.create_index(
                column,
-                replace=replace,
                config=config,
                wait_timeout=wait_timeout,
                name=name,
@@ -555,7 +560,7 @@ class RemoteTable(Table):
        column: str,
        *,
        config: IndexConfigType,
-        replace: bool = False,
+        replace: Optional[bool] = None,
        wait_timeout: Optional[timedelta] = None,
        name: Optional[str] = None,
        train: bool = True,
@@ -587,6 +592,7 @@ class RemoteTable(Table):
        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:
@@ -598,6 +604,7 @@ class RemoteTable(Table):
                num_sub_vectors,
                accelerator,
                index_cache_size,
+                replace,
            )
        ):
            return True
@@ -1132,7 +1132,7 @@ class Table(ABC):
        num_partitions: Optional[int] = None,
        num_sub_vectors: Optional[int] = None,
        vector_column_name: str = VECTOR_COLUMN_NAME,
-        replace: bool = False,
+        replace: bool = True,
        accelerator: Optional[str] = None,
        index_cache_size: Optional[int] = None,
        *,
@@ -1166,7 +1166,7 @@ class Table(ABC):
            The index configuration object. If provided, uses the new unified API.
            Can be one of: IvfFlat, IvfPq, IvfSq, IvfRq, HnswPq, HnswSq,
            BTree, Bitmap, LabelList, Fm, FTS.
-        replace : bool, default False
+        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.
@@ -1198,7 +1198,7 @@ class Table(ABC):
        column: str,
        *,
        config: IndexConfigType,
-        replace: bool = False,
+        replace: Optional[bool] = None,
        wait_timeout: Optional[timedelta] = None,
        name: Optional[str] = None,
        train: bool = True,
@@ -1260,7 +1260,7 @@ class Table(ABC):
        self,
        column: str,
        *,
-        replace: bool = False,
+        replace: bool = True,
        index_type: ScalarIndexType = "BTREE",
        wait_timeout: Optional[timedelta] = None,
        name: Optional[str] = None,
@@ -1272,7 +1272,7 @@ class Table(ABC):
        column : str
            The column to be indexed.  Must be a boolean, integer, float,
            or string column.
-        replace : bool, default False
+        replace : bool, default True
            Replace the existing index if it exists.
        index_type: Literal["BTREE", "BITMAP", "LABEL_LIST"], default "BTREE"
            The type of index to create.
@@ -1547,9 +1547,7 @@ class Table(ABC):
        on: Union[str, Iterable[str]]
            A column (or columns) to join on.  This is how records from the
            source table and target table are matched.  Typically this is some
-            kind of key or id column.  Passing several columns matches on the
-            composite key: a source row updates a target row only when it
-            agrees on every one of them.
+            kind of key or id column.

        Examples
        --------
@@ -2988,7 +2986,7 @@ class LanceTable(Table):
        num_partitions: Optional[int] = None,
        num_sub_vectors: Optional[int] = None,
        vector_column_name: str = VECTOR_COLUMN_NAME,
-        replace: bool = False,
+        replace: bool = True,
        accelerator: Optional[str] = None,
        index_cache_size: Optional[int] = None,
        num_bits: int = 8,
@@ -3030,7 +3028,7 @@ class LanceTable(Table):
            The index configuration object. If provided, uses the new unified API.
            Can be one of: IvfFlat, IvfPq, IvfSq, IvfRq, HnswPq, HnswSq,
            BTree, Bitmap, LabelList, Fm, FTS.
-        replace : bool, default False
+        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.
@@ -3169,7 +3167,7 @@ class LanceTable(Table):
        column: str,
        *,
        config: IndexConfigType,
-        replace: bool = False,
+        replace: Optional[bool] = None,
        wait_timeout: Optional[timedelta] = None,
        name: Optional[str] = None,
        train: bool = True,
@@ -3424,7 +3422,7 @@ class LanceTable(Table):
        self,
        column: str,
        *,
-        replace: bool = False,
+        replace: bool = True,
        index_type: ScalarIndexType = "BTREE",
        name: Optional[str] = None,
    ):
@@ -5313,7 +5311,7 @@ class AsyncTable:
        self,
        column: str,
        *,
-        replace: bool = False,
+        replace: Optional[bool] = None,
        config: Optional[
            Union[
                IvfFlat,
@@ -5344,14 +5342,14 @@ class AsyncTable:
        ----------
        column: str
            The column to index.
-        replace: bool, default False
+        replace: bool, default True
            Whether to replace the existing index

            If this is false, and another index already exists on the same columns
            and the same name, then an error will be returned.  This is true even if
            that index is out of date.

-            The default is False
+            The default is True
        config: default None
            For advanced configuration you can specify the type of index you would
            like to create.   You can also specify index-specific parameters when
@@ -5409,7 +5407,7 @@ class AsyncTable:
        self,
        column: str,
        *,
-        replace: bool = False,
+        replace: Optional[bool] = None,
        config: Optional[
            Union[
                IvfFlat,
@@ -5703,9 +5701,7 @@ class AsyncTable:
        on: Union[str, Iterable[str]]
            A column (or columns) to join on.  This is how records from the
            source table and target table are matched.  Typically this is some
-            kind of key or id column.  Passing several columns matches on the
-            composite key: a source row updates a target row only when it
-            agrees on every one of them.
+            kind of key or id column.

        Examples
        --------
@@ -668,167 +668,6 @@ def test_blob_fields_use_the_scalar_function_semantic_type():
    assert signature.output.arrow_type == "blob_v2"


-def test_whole_named_struct_function_can_include_a_blob_result_field():
-    @udf(
-        input_schema=pa.schema([lancedb.blob("image", nullable=False)]),
-        output_schema=pa.field(
-            "payload",
-            pa.struct(
-                [
-                    pa.field("mime_type", pa.string(), nullable=False),
-                    lancedb.blob("image", nullable=False),
-                ]
-            ),
-            nullable=False,
-        ),
-    )
-    def inspect_blob(image):
-        return {"mime_type": "image/png", "image": image}
-
-    output = inspect_blob.registration_request.signature.output
-    assert output.kind == "named_struct"
-    assert [(field.name, field.arrow_type) for field in output.fields] == [
-        ("mime_type", "utf8"),
-        ("image", "blob_v2"),
-    ]
-
-
-def test_struct_blob_signature_fields_preserve_exact_metadata_and_nullability():
-    nested_input = pa.field(
-        "payload",
-        pa.struct(
-            [
-                pa.field("mime_type", pa.string(), nullable=False),
-                pa.field(
-                    "nested",
-                    pa.struct([lancedb.blob("image", nullable=True)]),
-                    nullable=True,
-                ),
-            ]
-        ),
-        nullable=True,
-    )
-    nested_output = pa.field(
-        "result",
-        pa.struct(
-            [
-                pa.field("mime_type", pa.string(), nullable=False),
-                pa.field(
-                    "nested",
-                    pa.struct([lancedb.blob("image", nullable=True)]),
-                    nullable=False,
-                ),
-            ]
-        ),
-        nullable=False,
-    )
-
-    @udf(input_schema=pa.schema([nested_input]), output_schema=nested_output)
-    def copy_payload(payload):
-        return payload
-
-    signature = copy_payload.registration_request.signature
-    input_type = json.loads(signature.inputs[0].arrow_type)
-    assert input_type["fields"][1]["nullable"] is True
-    input_blob = input_type["fields"][1]["type"]["fields"][0]
-    assert input_blob["nullable"] is True
-    assert input_blob["metadata"] == {"ARROW:extension:name": "lance.blob.v2"}
-
-    assert signature.output.kind == "named_struct"
-    nested_result = next(
-        field for field in signature.output.fields if field.name == "nested"
-    )
-    output_type = json.loads(nested_result.arrow_type)
-    output_blob = output_type["fields"][0]
-    assert output_blob["nullable"] is True
-    assert output_blob["metadata"] == {"ARROW:extension:name": "lance.blob.v2"}
-
-
-def test_struct_blob_signature_supports_multiple_struct_levels():
-    recursive = pa.field(
-        "value",
-        pa.struct(
-            [
-                pa.field(
-                    "level_1",
-                    pa.struct(
-                        [
-                            pa.field(
-                                "level_2",
-                                pa.struct([lancedb.blob("image", nullable=False)]),
-                                nullable=False,
-                            )
-                        ]
-                    ),
-                    nullable=False,
-                )
-            ]
-        ),
-        nullable=False,
-    )
-
-    @udf(
-        input_schema=pa.schema([recursive]),
-        output_schema=pa.field("size", pa.int64(), nullable=False),
-    )
-    def blob_size(value):
-        return len(value["level_1"]["level_2"]["image"])
-
-    encoded = json.loads(blob_size.registration_request.signature.inputs[0].arrow_type)
-    blob = encoded["fields"][0]["type"]["fields"][0]["type"]["fields"][0]
-    assert blob["metadata"]["ARROW:extension:name"] == "lance.blob.v2"
-
-
-@pytest.mark.parametrize(
-    "data_type",
-    [
-        pa.list_(lancedb.blob("item", nullable=False)),
-        pa.large_list(lancedb.blob("item", nullable=False)),
-        pa.list_(lancedb.blob("item", nullable=False), 2),
-        pa.map_(pa.string(), lancedb.blob("value", nullable=False).type),
-    ],
-)
-def test_blob_signature_rejects_collection_ancestors(data_type):
-    with pytest.raises(
-        TypeError,
-        match="Blob v2 fields nested under collection types are not supported",
-    ):
-
-        @udf(
-            input_schema=pa.schema([pa.field("value", data_type, nullable=False)]),
-            output_schema=pa.field("size", pa.int64(), nullable=False),
-        )
-        def blob_size(value):
-            return len(value)
-
-
-def test_blob_signature_rejects_collection_below_a_struct():
-    nested = pa.field(
-        "value",
-        pa.struct(
-            [
-                pa.field(
-                    "images",
-                    pa.list_(lancedb.blob("item", nullable=False)),
-                    nullable=False,
-                )
-            ]
-        ),
-        nullable=False,
-    )
-    with pytest.raises(
-        TypeError,
-        match="Blob v2 fields nested under collection types are not supported",
-    ):
-
-        @udf(
-            input_schema=pa.schema([nested]),
-            output_schema=pa.field("size", pa.int64(), nullable=False),
-        )
-        def blob_size(value):
-            return len(value["images"])
-
-
 def test_named_struct_function_can_include_a_blob_result_field():
    @udf(
        input_schema=pa.schema([lancedb.blob("image", nullable=False)]),
@@ -890,6 +729,22 @@ def test_blob_marker_rejects_invalid_storage_layout():
            return len(image)


+def test_nested_blob_signature_field_has_a_clear_error():
+    nested = pa.field(
+        "value",
+        pa.struct([lancedb.blob("image", nullable=False)]),
+        nullable=False,
+    )
+    with pytest.raises(TypeError, match="nested Blob v2 fields are not supported"):
+
+        @udf(
+            input_schema=pa.schema([nested]),
+            output_schema=pa.field("size", pa.int64(), nullable=False),
+        )
+        def blob_size(value):
+            return len(value["image"])
+
+
 def test_nested_non_blob_extension_is_not_silently_unwrapped():
    class TestExtension(pa.ExtensionType):
        def __init__(self):
@@ -1162,8 +1017,6 @@ def test_local_function_catalog_operations_are_not_supported(tmp_path):
        db.create_function_async(normalize_score)
    with pytest.raises(NotImplementedError, match=message):
        db.get_function("normalize_score", version="fv_exact")
-    with pytest.raises(NotImplementedError, match=message):
-        db.list_functions()
    with pytest.raises(NotImplementedError, match=message):
        db.drop_function("normalize_score", version="fv_exact")

@@ -1211,22 +1064,6 @@ def _mock_remote_function_catalog():
                    "version": "fv_exact",
                }
                response = state["version"]
-            elif self.path == "/v1/functions/list":
-                assert body["include_definition"] is True
-                if "page_token" not in body:
-                    response = {
-                        "functions": [
-                            {
-                                "name": "normalize_score",
-                                "version": "fv_exact",
-                                "definition": state["version"],
-                            }
-                        ],
-                        "page_token": "next",
-                    }
-                else:
-                    assert body["page_token"] == "next"
-                    response = {"functions": []}
            elif self.path == "/v1/functions/drop":
                assert body == {
                    "name": "normalize_score",
@@ -1293,49 +1130,6 @@ def test_blocking_remote_registration_returns_function_version():
    ]


-def test_remote_list_functions_paginates_and_returns_typed_versions():
-    with _mock_remote_function_catalog() as (host, state):
-        db = lancedb.connect(
-            "db://dev",
-            api_key="fake",
-            host_override=host,
-            client_config={"retry_config": {"retries": 0}},
-        )
-        created = db.create_function(normalize_score)
-        state["requests"].clear()
-        functions = db.list_functions()
-
-    assert functions == [created]
-    assert state["requests"] == [
-        ("/v1/functions/list", {"include_definition": True}),
-        (
-            "/v1/functions/list",
-            {"include_definition": True, "page_token": "next"},
-        ),
-    ]
-
-
-@pytest.mark.asyncio
-async def test_async_remote_list_functions_returns_typed_versions():
-    with _mock_remote_function_catalog() as (host, state):
-        db = await lancedb.connect_async(
-            "db://dev",
-            api_key="fake",
-            host_override=host,
-            client_config={"retry_config": {"retries": 0}},
-        )
-        registration = await db.create_function_async(normalize_score)
-        created = await registration.wait()
-        state["requests"].clear()
-        functions = await db.list_functions()
-
-    assert functions == [created]
-    assert [path for path, _ in state["requests"]] == [
-        "/v1/functions/list",
-        "/v1/functions/list",
-    ]
-
-
 def test_remote_drop_function_sends_exact_version():
    with _mock_remote_function_catalog() as (host, state):
        db = lancedb.connect(
@@ -97,9 +97,6 @@ async def test_create_index_async_returns_done_job(some_table: AsyncTable):
 async def test_create_scalar_index(some_table: AsyncTable):
    # Can create
    await some_table.create_index("id")
-    # Can't recreate by default
-    with pytest.raises(RuntimeError, match="already exists"):
-        await some_table.create_index("id")
    # Can recreate if replace=True
    await some_table.create_index("id", replace=True)
    indices = await some_table.list_indices()
@@ -113,7 +110,7 @@ async def test_create_scalar_index(some_table: AsyncTable):
    with pytest.raises(RuntimeError, match="already exists"):
        await some_table.create_index("id", replace=False)
    # can also specify index type
-    await some_table.create_index("id", config=BTree(), replace=True)
+    await some_table.create_index("id", config=BTree())

    await some_table.drop_index("id_idx")
    indices = await some_table.list_indices()
@@ -354,18 +351,13 @@ async def test_full_text_search_index(some_table: AsyncTable):
 async def test_create_vector_index(some_table: AsyncTable):
    # Can create
    await some_table.create_index("vector")
-    # Can't recreate by default
-    with pytest.raises(RuntimeError, match="already exists"):
-        await some_table.create_index("vector")
    # Can recreate if replace=True
    await some_table.create_index("vector", replace=True)
    # Can't recreate if replace=False
    with pytest.raises(RuntimeError, match="already exists"):
        await some_table.create_index("vector", replace=False)
    # Can also specify index type
-    await some_table.create_index(
-        "vector", config=IvfPq(num_partitions=100), replace=True
-    )
+    await some_table.create_index("vector", config=IvfPq(num_partitions=100))
    indices = await some_table.list_indices()
    assert len(indices) == 1
    assert indices[0].index_type == "IvfPq"
@@ -266,38 +266,3 @@ async def test_async_namespace_connection_materialized_views(tmp_path):
            handle._route_pushdown_to_rust == through_namespace._route_pushdown_to_rust
        )
        assert handle._namespace_path == through_namespace._namespace_path
-
-
-def test_namespaced_select_kind_is_read_and_unknown_kinds_are_refused():
-    import json
-
-    import pyarrow as pa
-
-    from lancedb.materialized_view import _definition_from_schema
-
-    def schema_with(definition: dict) -> pa.Schema:
-        return pa.schema([pa.field("id", pa.int32())]).with_metadata(
-            {b"mv.definition": json.dumps(definition).encode()}
-        )
-
-    # "namespaced_select" is the namespaced form of "select": same shape,
-    # a separate kind so readers that predate it refuse instead of
-    # resolving the source at the root.
-    definition = _definition_from_schema(
-        schema_with(
-            {
-                "kind": "namespaced_select",
-                "source_table": "people",
-                "source_namespace": ["ns"],
-                "projections": [{"output": "name", "expression": "name"}],
-            }
-        ),
-        "v",
-    )
-    assert definition.source_table == "people"
-    assert definition.source_namespace == ["ns"]
-
-    with pytest.raises(NotImplementedError, match="cannot refresh"):
-        _definition_from_schema(
-            schema_with({"kind": "select_v3", "source_table": "people"}), "v"
-        )
@@ -820,13 +820,11 @@ def test_table_create_indices():
        scalar_req = received_requests[0]
        assert "name" in scalar_req
        assert scalar_req["name"] == "custom_scalar_idx"
-        assert scalar_req["replace"] is False

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

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

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

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

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


 def test_table_wait_for_index_timeout():
@@ -2542,6 +2534,26 @@ def test_remote_connection_jobs_surface():
            request.send_header("Content-Type", "application/json")
            request.end_headers()
            request.wfile.write(b'{"job_id": "job-1"}')
+        elif request.path == "/v1/jobs/pause":
+            if payload["job_id"] != "job-1":
+                request.send_response(404)
+                request.end_headers()
+                return
+            request.send_response(200)
+            request.send_header("Content-Type", "application/json")
+            request.end_headers()
+            request.wfile.write(b'{"job_id": "job-1", "paused": true}')
+        elif request.path == "/v1/jobs/resume":
+            if payload["job_id"] != "job-1":
+                request.send_response(404)
+                request.end_headers()
+                return
+            request.send_response(200)
+            request.send_header("Content-Type", "application/json")
+            request.end_headers()
+            request.wfile.write(
+                b'{"job_id": "job-1", "resumed": false, "still_pausing": true}'
+            )
        elif request.path == "/v1/jobs/query_events":
            assert payload["job_id"] == "job-1"
            request.send_response(200)
@@ -2570,6 +2582,9 @@ def test_remote_connection_jobs_surface():
        assert db.cancel_job("job-1") is True
        assert db.cancel_job("missing") is False

+        assert db.pause_job("job-1") == "pausing"
+        assert db.resume_job("job-1") == "still_pausing"
+
        batches = db.job_history("job-1")
        assert len(batches) == 1
        assert batches[0].num_rows == 2
@@ -1600,7 +1600,7 @@ def test_create_index_method(mock_create_index, mem_db: DBConnection):
    )
    mock_create_index.assert_called_with(
        "my_vector",
-        replace=False,
+        replace=True,
        config=expected_config,
        wait_timeout=None,
        name=None,
@@ -1620,7 +1620,7 @@ def test_create_index_method(mock_create_index, mem_db: DBConnection):
    )
    mock_create_index.assert_called_with(
        "my_vector",
-        replace=False,
+        replace=True,
        config=expected_config,
        wait_timeout=None,
        name=None,
@@ -1646,7 +1646,7 @@ def test_create_index_name_and_train_parameters(
    expected_config = IvfPq()  # Default config
    mock_create_index.assert_called_with(
        "vector",
-        replace=False,
+        replace=True,
        config=expected_config,
        wait_timeout=None,
        name="my_custom_index",
@@ -1657,7 +1657,7 @@ def test_create_index_name_and_train_parameters(
    table.create_index(vector_column_name="vector", train=False)
    mock_create_index.assert_called_with(
        "vector",
-        replace=False,
+        replace=True,
        config=expected_config,
        wait_timeout=None,
        name=None,
@@ -1668,7 +1668,7 @@ def test_create_index_name_and_train_parameters(
    table.create_index(vector_column_name="vector", name="my_index_name", train=True)
    mock_create_index.assert_called_with(
        "vector",
-        replace=False,
+        replace=True,
        config=expected_config,
        wait_timeout=None,
        name="my_index_name",
@@ -1705,7 +1705,7 @@ def test_create_index_new_api(mock_create_index, mem_db: DBConnection):
        table.create_index("vector", config=IvfPq(distance_type="l2"))
    mock_create_index.assert_called_with(
        "vector",
-        replace=False,
+        replace=True,
        config=IvfPq(distance_type="l2"),
        wait_timeout=None,
        name=None,
@@ -1716,7 +1716,7 @@ def test_create_index_new_api(mock_create_index, mem_db: DBConnection):
    table.create_index("category", config=BTree())
    mock_create_index.assert_called_with(
        "category",
-        replace=False,
+        replace=True,
        config=BTree(),
        wait_timeout=None,
        name=None,
@@ -1727,7 +1727,7 @@ def test_create_index_new_api(mock_create_index, mem_db: DBConnection):
    table.create_index("text", config=FTS(with_position=True))
    mock_create_index.assert_called_with(
        "text",
-        replace=False,
+        replace=True,
        config=FTS(with_position=True),
        wait_timeout=None,
        name=None,
@@ -2682,43 +2682,6 @@ def test_merge_insert(mem_db: DBConnection):
        )


-def test_merge_insert_composite_key(mem_db: DBConnection):
-    table = mem_db.create_table(
-        "my_table",
-        data=pa.table(
-            {
-                "shard": ["a", "a", "b"],
-                "id": [1, 2, 1],
-                "val": ["x", "y", "z"],
-            }
-        ),
-    )
-
-    # ("a", 1) matches an existing row and updates it. ("b", 2) agrees with an
-    # existing row on each key column separately but on neither pair, so it is
-    # an insert.
-    new_data = pa.table({"shard": ["a", "b"], "id": [1, 2], "val": ["X", "W"]})
-    res = (
-        table.merge_insert(["shard", "id"])
-        .when_matched_update_all()
-        .when_not_matched_insert_all()
-        .execute(new_data)
-    )
-    assert res.num_updated_rows == 1
-    assert res.num_inserted_rows == 1
-
-    expected = pa.table(
-        {
-            "shard": ["a", "a", "b", "b"],
-            "id": [1, 2, 1, 2],
-            "val": ["X", "y", "z", "W"],
-        }
-    )
-    assert table.to_arrow().sort_by([("shard", "ascending"), ("id", "ascending")]) == (
-        expected
-    )
-
-
 def test_merge_insert_nullable_pandas_into_pydantic_schema(mem_db: DBConnection):
    # Regression test for https://github.com/lancedb/lancedb/issues/2366
    pd = pytest.importorskip("pandas")
@@ -629,19 +629,6 @@ impl Connection {
        })
    }

-    pub fn list_functions(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
-        let inner = self_.get_inner()?.clone();
-        future_into_py(self_.py(), async move {
-            inner
-                .list_functions()
-                .await
-                .infer_error()?
-                .into_iter()
-                .map(|function| function.to_canonical_json().infer_error())
-                .collect::<PyResult<Vec<_>>>()
-        })
-    }
-
    pub fn drop_function(
        self_: PyRef<'_, Self>,
        name: String,
@@ -679,6 +666,30 @@ impl Connection {
        })
    }

+    pub fn pause_job(self_: PyRef<'_, Self>, job_id: String) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.get_inner()?.clone();
+        future_into_py(self_.py(), async move {
+            let status = inner.pause_job(&job_id).await.infer_error()?;
+            Ok(match status {
+                lancedb::database::PauseJobStatus::Pausing => "pausing",
+                lancedb::database::PauseJobStatus::AlreadyPaused => "already_paused",
+                lancedb::database::PauseJobStatus::Committing => "committing",
+            })
+        })
+    }
+
+    pub fn resume_job(self_: PyRef<'_, Self>, job_id: String) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.get_inner()?.clone();
+        future_into_py(self_.py(), async move {
+            let status = inner.resume_job(&job_id).await.infer_error()?;
+            Ok(match status {
+                lancedb::database::ResumeJobStatus::Resumed => "resumed",
+                lancedb::database::ResumeJobStatus::StillPausing => "still_pausing",
+                lancedb::database::ResumeJobStatus::NotPaused => "not_paused",
+            })
+        })
+    }
+
    #[pyo3(signature = (job_id=None))]
    pub fn job_history(
        self_: PyRef<'_, Self>,
@@ -1,6 +1,6 @@
 [package]
 name = "lancedb"
-version = "0.39.0-beta.1"
+version = "0.38.0"
 edition.workspace = true
 description = "LanceDB: A serverless, low-latency vector database for AI applications"
 license.workspace = true
@@ -24,7 +24,7 @@ use crate::data::scannable::Scannable;
 use crate::database::listing::ListingDatabase;
 use crate::database::{
    CloneTableRequest, Database, DatabaseOptions, JobDescription, JobInfo, OpenTableRequest,
-    ReadConsistency, TableNamesRequest,
+    PauseJobStatus, ReadConsistency, ResumeJobStatus, TableNamesRequest,
 };
 use crate::embeddings::{EmbeddingRegistry, MemoryRegistry};
 use crate::error::{Error, Result};
@@ -523,28 +523,6 @@ impl Connection {
            .await
    }

-    /// List every published immutable Function version in the remote catalog.
-    ///
-    /// Results are ordered by Function name then version. The client walks all
-    /// server pages before returning. Local databases return
-    /// [`Error::NotSupported`].
-    ///
-    /// # Example
-    ///
-    /// ```no_run
-    /// # async fn list_functions(
-    /// #     connection: &lancedb::Connection,
-    /// # ) -> Result<(), Box<dyn std::error::Error>> {
-    /// for function in connection.list_functions().await? {
-    ///     println!("{} {}", function.name(), function.version());
-    /// }
-    /// # Ok(())
-    /// # }
-    /// ```
-    pub async fn list_functions(&self) -> Result<Vec<crate::function::FunctionVersion>> {
-        self.internal.list_functions().await
-    }
-
    /// Drop one exact immutable Function version from the remote catalog.
    ///
    /// Returns `true` when the server appended a Dropped transition and
@@ -612,6 +590,18 @@ impl Connection {
        self.internal.cancel_job(job_id.as_ref()).await
    }

+    /// Pause a server-side job by id. Its workers drain and it stays parked
+    /// until resumed; see [`PauseJobStatus`] for the outcomes.
+    pub async fn pause_job(&self, job_id: impl AsRef<str>) -> Result<PauseJobStatus> {
+        self.internal.pause_job(job_id.as_ref()).await
+    }
+
+    /// Resume a paused server-side job by id. Its workers pick their work
+    /// back up from checkpoints; see [`ResumeJobStatus`] for the outcomes.
+    pub async fn resume_job(&self, job_id: impl AsRef<str>) -> Result<ResumeJobStatus> {
+        self.internal.resume_job(job_id.as_ref()).await
+    }
+
    /// The lifecycle event history of a server-side job (all jobs when
    /// `job_id` is `None`), as recorded Arrow batches.
    pub async fn job_history(&self, job_id: Option<&str>) -> Result<Vec<RecordBatch>> {
@@ -235,6 +235,29 @@ pub struct JobDescription {
    pub failure: Option<crate::error::JobFailure>,
 }

+/// The server's answer to a pause request.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum PauseJobStatus {
+    /// The pause was accepted; workers drain and the job stays parked.
+    Pausing,
+    /// The job was already paused, so a repeated pause changed nothing.
+    AlreadyPaused,
+    /// The job is finalizing its results and cannot be parked right now.
+    /// The commit is the short tail of a long job; retry shortly.
+    Committing,
+}
+
+/// The server's answer to a resume request.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ResumeJobStatus {
+    /// The job re-entered the queue and will run again.
+    Resumed,
+    /// The pause's worker drain is not confirmed yet; retry shortly.
+    StillPausing,
+    /// The job was not paused, so there was nothing to resume.
+    NotPaused,
+}
+
 fn job_op_not_supported<T>(what: &str) -> Result<T> {
    Err(crate::error::Error::NotSupported {
        message: format!("{} is not supported by this database", what),
@@ -307,10 +330,6 @@ pub trait Database:
    ) -> Result<crate::function::FunctionVersion> {
        function_catalog_not_supported()
    }
-    /// List every published immutable Function version in the remote catalog.
-    async fn list_functions(&self) -> Result<Vec<crate::function::FunctionVersion>> {
-        function_catalog_not_supported()
-    }
    /// Drop one exact immutable Function version from the remote catalog.
    async fn drop_function(&self, _name: &str, _version: &str) -> Result<bool> {
        function_catalog_not_supported()
@@ -335,6 +354,16 @@ pub trait Database:
    async fn cancel_job(&self, _job_id: &str) -> Result<bool> {
        job_op_not_supported("cancel_job")
    }
+    /// Pause a job by id. The job's workers drain and it stays parked until
+    /// resumed; see [`PauseJobStatus`] for the outcomes.
+    async fn pause_job(&self, _job_id: &str) -> Result<PauseJobStatus> {
+        job_op_not_supported("pause_job")
+    }
+    /// Resume a paused job by id. It re-enters the queue and its workers pick
+    /// their work back up from checkpoints; see [`ResumeJobStatus`].
+    async fn resume_job(&self, _job_id: &str) -> Result<ResumeJobStatus> {
+        job_op_not_supported("resume_job")
+    }
    /// The lifecycle event history of a job (all jobs when `job_id` is
    /// `None`), as recorded Arrow batches.
    async fn job_history(&self, _job_id: Option<&str>) -> Result<Vec<RecordBatch>> {
@@ -200,14 +200,14 @@ impl IndexBuilder {
            parent,
            index,
            columns,
-            replace: false,
+            replace: true,
            train: true,
            wait_timeout: None,
            name: None,
        }
    }

-    /// Whether to replace the existing index, the default is `false`.
+    /// Whether to replace the existing index, the default is `true`.
    ///
    /// If this is false, and another index already exists on the same columns
    /// and the same name, then an error will be returned.  This is true even if
@@ -40,7 +40,6 @@ impl TerminalResult {
        }
    }

-    #[cfg(feature = "remote")]
    pub(crate) fn remote(value: Option<Value>, request_id: String) -> Self {
        Self {
            value,
@@ -48,46 +47,30 @@ impl TerminalResult {
        }
    }

-    #[cfg(feature = "remote")]
    pub(crate) fn value(&self) -> Option<&Value> {
        self.value.as_ref()
    }

    fn decode<T: DeserializeOwned>(self) -> Result<T> {
-        let value = self.value.ok_or_else(|| {
-            #[cfg(feature = "remote")]
-            if let Some(request_id) = &self.request_id {
-                return Error::Http {
-                    source: "successful typed job response did not contain a result".into(),
-                    request_id: request_id.clone(),
-                    status_code: None,
-                };
-            }
-            Error::Runtime {
+        let value = self.value.ok_or_else(|| match &self.request_id {
+            Some(request_id) => Error::Http {
+                source: "successful typed job response did not contain a result".into(),
+                request_id: request_id.clone(),
+                status_code: None,
+            },
+            None => Error::Runtime {
                message: "successful typed job did not contain a result".to_string(),
-            }
+            },
        })?;
-        serde_json::from_value(value).map_err(|error| {
-            #[cfg(feature = "remote")]
-            {
-                match self.request_id {
-                    Some(request_id) => Error::Http {
-                        source: format!("failed to parse typed job result: {error}").into(),
-                        request_id,
-                        status_code: None,
-                    },
-                    None => Error::Runtime {
-                        message: format!("failed to parse typed job result: {error}"),
-                    },
-                }
-            }
-            #[cfg(not(feature = "remote"))]
-            {
-                let _ = self.request_id;
-                Error::Runtime {
-                    message: format!("failed to parse typed job result: {error}"),
-                }
-            }
+        serde_json::from_value(value).map_err(|error| match self.request_id {
+            Some(request_id) => Error::Http {
+                source: format!("failed to parse typed job result: {error}").into(),
+                request_id,
+                status_code: None,
+            },
+            None => Error::Runtime {
+                message: format!("failed to parse typed job result: {error}"),
+            },
        })
    }
 }
@@ -134,7 +117,6 @@ impl Job<()> {
        }
    }

-    #[cfg(feature = "remote")]
    pub(crate) fn new(handle: Box<dyn JobHandle>) -> Self {
        Self {
            inner: JobInner::Handle {
@@ -74,15 +74,8 @@ const EMBEDDING_FUNCTIONS_META_KEY: &str = "embedding_functions";
 const COLUMN_DEFINITIONS_META_KEY: &str = "lancedb::column_definitions";

 /// Value of the definition's `kind` tag for the projected `select` form.
-/// Reserved for root-namespace sources; see [`NAMESPACED_SELECT_KIND`].
 pub const SELECT_KIND: &str = "select";

-/// The `select` form over a namespaced source: its own kind, because released
-/// readers drop unknown fields and resolve a `select` source at the root, so
-/// this routes them to the [`MaterializedViewKind::Unrecognized`] refusal
-/// instead of a wrong-table refresh.
-pub const NAMESPACED_SELECT_KIND: &str = "namespaced_select";
-
 /// Which view outputs each source column is projected to directly. A column
 /// may be projected more than once, so each carries every name the view gives
 /// it, in projection order.
@@ -102,10 +95,6 @@ pub struct ViewProjection {
 pub struct MaterializedViewDefinition {
    /// Name of the source table, in the same database as the view.
    pub source_table: String,
-    /// Namespace path holding the source table; empty is the root namespace.
-    /// A definition written before namespaced sources reads as root.
-    #[serde(default, skip_serializing_if = "Vec::is_empty")]
-    pub source_namespace: Vec<String>,
    /// The projected output columns, in view schema order.
    pub projections: Vec<ViewProjection>,
    /// SQL predicate selecting the source rows the view holds.
@@ -140,12 +129,7 @@ pub(crate) fn definition_to_metadata(definition: &MaterializedViewDefinition) ->
    let mut value = serde_json::to_value(definition).map_err(|e| Error::Runtime {
        message: format!("failed to serialize view definition: {e}"),
    })?;
-    let kind = if definition.source_namespace.is_empty() {
-        SELECT_KIND
-    } else {
-        NAMESPACED_SELECT_KIND
-    };
-    value["kind"] = serde_json::Value::String(kind.to_string());
+    value["kind"] = serde_json::Value::String(SELECT_KIND.to_string());
    Ok(value.to_string())
 }

@@ -166,21 +150,12 @@ pub fn materialized_view_kind(
        .get("kind")
        .and_then(|k| k.as_str())
        .ok_or_else(|| unreadable(&"missing kind tag"))?;
-    if kind != SELECT_KIND && kind != NAMESPACED_SELECT_KIND {
+    if kind != SELECT_KIND {
        return Ok(Some(MaterializedViewKind::Unrecognized {
            kind: kind.to_string(),
        }));
    }
-    let kind = kind.to_string();
-    let definition: MaterializedViewDefinition =
-        serde_json::from_value(value).map_err(|e| unreadable(&e))?;
-    // No correct writer produces a kind that disagrees with its namespace.
-    if (kind == SELECT_KIND) != definition.source_namespace.is_empty() {
-        return Err(unreadable(&format!(
-            "kind '{kind}' does not match its source namespace {:?}",
-            definition.source_namespace
-        )));
-    }
+    let definition = serde_json::from_value(value).map_err(|e| unreadable(&e))?;
    Ok(Some(MaterializedViewKind::Select(definition)))
 }

@@ -191,7 +166,6 @@ pub fn materialized_view_kind(
 pub(crate) fn plan(
    source_schema: SchemaRef,
    source_table: &str,
-    source_namespace: &[String],
    projections: &[(String, String)],
    filter: Option<&str>,
    limit: Option<u64>,
@@ -345,7 +319,6 @@ pub(crate) fn plan(

    let definition = MaterializedViewDefinition {
        source_table: source_table.to_string(),
-        source_namespace: source_namespace.to_vec(),
        projections: projections
            .into_iter()
            .map(|(output, expression)| ViewProjection { output, expression })
@@ -629,7 +602,7 @@ pub struct PreparedDeclaration {
    definition: MaterializedViewDefinition,
    /// The source's own database: the only place
    /// [`PreparedDeclaration::create`] will put the view, because refresh
-    /// resolves the recorded source coordinate through the view's database.
+    /// resolves the recorded source name through the view's database.
    database: Arc<dyn Database>,
 }

@@ -649,21 +622,10 @@ impl PreparedDeclaration {

    /// Create the view table and verify it, consuming the declaration.
    ///
-    /// The view goes at the root of the source's own database, where refresh
-    /// resolves the recorded source coordinate. Stable row ids are requested
-    /// at both levels and verified rather than trusted; nothing is rolled
-    /// back on failure.
+    /// The view goes in the source's own database, where refresh resolves the
+    /// recorded source name. Stable row ids are requested at both levels and
+    /// verified rather than trusted; nothing is rolled back on failure.
    pub async fn create(self, name: &str) -> Result<MaterializedView> {
-        self.create_in(&[], name).await
-    }
-
-    /// Create the view in `namespace_path`, empty for the root namespace.
-    /// Otherwise [`PreparedDeclaration::create`].
-    pub async fn create_in(
-        self,
-        namespace_path: &[String],
-        name: &str,
-    ) -> Result<MaterializedView> {
        let empty: Vec<std::result::Result<arrow_array::RecordBatch, arrow_schema::ArrowError>> =
            vec![];
        // Minted here, not at preparation: a declaration can be cloned and
@@ -678,7 +640,6 @@ impl PreparedDeclaration {
        let reader: Box<dyn arrow_array::RecordBatchReader + Send> =
            Box::new(arrow_array::RecordBatchIterator::new(empty, schema));
        let mut request = CreateTableRequest::new(name.to_string(), Box::new(reader));
-        request.namespace_path = namespace_path.to_vec();
        let write_params = request
            .write_options
            .lance_write_params
@@ -719,8 +680,8 @@ impl PreparedDeclaration {

 /// Validate a view declaration against its live source and hold what its
 /// creation needs. The declaration is canonicalized through the coordinate a
-/// refresh will resolve -- name and namespace both -- so a handle that does
-/// not resolve back to itself is rejected. Same creation-time checks as
+/// refresh will resolve, so a handle that does not resolve back to itself is
+/// rejected, as is a namespaced source. Same creation-time checks as
 /// [`Connection::create_materialized_view`].
 ///
 /// ```no_run
@@ -749,9 +710,17 @@ pub async fn prepare_declaration(
            message: "materialized views are supported only on local databases".into(),
        });
    };
-    // Refresh resolves the source at exactly this coordinate, so the
-    // definition records the namespace alongside the name.
-    let source_namespace = source.namespace().to_vec();
+    // The definition records the source by bare name; any other source
+    // form would be recorded as a name its refresh cannot resolve.
+    if !source.namespace().is_empty() {
+        return Err(Error::NotSupported {
+            message: format!(
+                "a namespaced source cannot be recorded in a view definition; \
+                 '{}' must be a root-namespace table",
+                source.name()
+            ),
+        });
+    }
    let database = source
        .database_opt()
        .ok_or_else(|| Error::InvalidInput {
@@ -765,7 +734,7 @@ pub async fn prepare_declaration(
    let resolved = database
        .open_table(OpenTableRequest {
            name: source.name().to_string(),
-            namespace_path: source_namespace.clone(),
+            namespace_path: vec![],
            index_cache_size: None,
            lance_read_params: None,
            location: None,
@@ -811,7 +780,6 @@ pub async fn prepare_declaration(
    let (definition, mut fields, lineage) = plan(
        source_schema.clone(),
        resolved.name(),
-        &source_namespace,
        projections,
        filter,
        limit,
@@ -871,9 +839,7 @@ fn ensure_local(connection: &Connection) -> Result<()> {
 pub struct CreateMaterializedViewBuilder {
    connection: Connection,
    name: String,
-    namespace: Vec<String>,
    source: String,
-    source_namespace: Vec<String>,
    projections: Vec<(String, String)>,
    filter: Option<String>,
    limit: Option<u64>,
@@ -884,28 +850,13 @@ impl CreateMaterializedViewBuilder {
        Self {
            connection,
            name,
-            namespace: Vec::new(),
            source,
-            source_namespace: Vec::new(),
            projections: Vec::new(),
            filter: None,
            limit: None,
        }
    }

-    /// The namespace to create the view in. Defaults to the root namespace.
-    pub fn namespace(mut self, namespace_path: Vec<String>) -> Self {
-        self.namespace = namespace_path;
-        self
-    }
-
-    /// The namespace holding the source table; recorded in the definition
-    /// for refresh to resolve. Defaults to the root namespace.
-    pub fn source_namespace(mut self, namespace_path: Vec<String>) -> Self {
-        self.source_namespace = namespace_path;
-        self
-    }
-
    /// The view's columns, as `(name, SQL expression)` pairs. Not calling
    /// this selects every source column, expanded at creation time.
    pub fn select(
@@ -936,12 +887,7 @@ impl CreateMaterializedViewBuilder {
    /// provenance across compaction, and cannot be enabled later.
    pub async fn execute(self) -> Result<MaterializedView> {
        ensure_local(&self.connection)?;
-        let source = self
-            .connection
-            .open_table(&self.source)
-            .namespace(self.source_namespace.clone())
-            .execute()
-            .await?;
+        let source = self.connection.open_table(&self.source).execute().await?;
        let prepared = prepare_declaration(
            &source,
            &self.projections,
@@ -949,7 +895,7 @@ impl CreateMaterializedViewBuilder {
            self.limit,
        )
        .await?;
-        prepared.create_in(&self.namespace, &self.name).await
+        prepared.create(&self.name).await
    }
 }

@@ -1206,7 +1152,6 @@ mod tests {
            view.definition(),
            &MaterializedViewDefinition {
                source_table: "people".into(),
-                source_namespace: Vec::new(),
                projections: vec![
                    ViewProjection {
                        output: "name".into(),
@@ -2138,138 +2083,33 @@ mod tests {
            .await
            .unwrap_err();
        assert!(err.to_string().contains("custom_loc"), "{err}");
-    }

-    /// A view declared over a namespaced source records that namespace, and
-    /// refresh resolves the source through it -- the coordinate round-trips.
-    #[tokio::test]
-    async fn a_namespaced_source_round_trips_through_refresh() {
-        use lance_namespace::models::CreateNamespaceRequest;
-
-        let tmp = tempfile::tempdir().unwrap();
-        let mut properties = std::collections::HashMap::new();
-        properties.insert("root".to_string(), tmp.path().to_str().unwrap().to_string());
-        let conn = crate::connect_namespace("dir", properties)
-            .execute()
-            .await
-            .unwrap();
-        conn.create_namespace(CreateNamespaceRequest {
-            id: Some(vec!["ns".into()]),
-            ..Default::default()
-        })
+        // A namespaced source cannot be recorded in the definition: the
+        // bare name refresh resolves would reach a different table or none.
+        let namespaced = crate::table::NativeTable::create(
+            "memory://ns_src",
+            "ns_src",
+            vec!["ns".to_string()],
+            Box::new(arrow_array::RecordBatchIterator::new(
+                vec![],
+                std::sync::Arc::new(arrow_schema::Schema::new(vec![arrow_schema::Field::new(
+                    "id",
+                    arrow_schema::DataType::Int32,
+                    true,
+                )])),
+            )) as Box<dyn arrow_array::RecordBatchReader + Send>,
+            None,
+            None,
+            None,
+            None,
+            std::collections::HashSet::new(),
+        )
        .await
        .unwrap();
-
-        let batch = record_batch!(
-            ("name", Utf8, ["ada", "grace", "alan"]),
-            ("age", Int32, [36, 85, 41])
-        )
-        .unwrap();
-        conn.create_table("people", batch)
-            .namespace(vec!["ns".to_string()])
-            .write_options(stable_row_ids())
-            .execute()
+        let namespaced = Table::new(std::sync::Arc::new(namespaced), conn.database().clone());
+        let err = prepare_declaration(&namespaced, &[], None, None)
            .await
-            .unwrap();
-
-        // A decoy of the same name at the root: resolving the source at the
-        // wrong namespace materializes one row here instead of three.
-        let decoy = record_batch!(("name", Utf8, ["mallory"]), ("age", Int32, [42])).unwrap();
-        conn.create_table("people", decoy)
-            .write_options(stable_row_ids())
-            .execute()
-            .await
-            .unwrap();
-
-        let view = conn
-            .create_materialized_view("adults", "people")
-            .namespace(vec!["ns".to_string()])
-            .source_namespace(vec!["ns".to_string()])
-            .select([("name", "name")])
-            .only_if("age >= 18")
-            .execute()
-            .await
-            .unwrap();
-
-        assert_eq!(view.definition().source_table, "people");
-        assert_eq!(view.definition().source_namespace, vec!["ns".to_string()]);
-        assert_eq!(view.table().namespace(), &["ns"]);
-
-        // Refresh resolves the source at the recorded namespace, not at root.
-        let result = view.refresh().execute().await.unwrap();
-        assert_eq!(result.rows_written, 3);
-    }
-
-    /// A definition stored before namespaced sources existed carries no
-    /// namespace key and must read as the root namespace.
-    #[test]
-    fn a_definition_without_a_namespace_reads_as_root() {
-        let stored =
-            r#"{"source_table":"people","projections":[{"output":"name","expression":"name"}]}"#;
-        let definition: MaterializedViewDefinition = serde_json::from_str(stored).unwrap();
-        assert!(definition.source_namespace.is_empty());
-    }
-
-    fn definition(source_namespace: Vec<String>) -> MaterializedViewDefinition {
-        MaterializedViewDefinition {
-            source_table: "people".to_string(),
-            source_namespace,
-            projections: vec![ViewProjection {
-                output: "name".to_string(),
-                expression: "name".to_string(),
-            }],
-            filter: None,
-            limit: None,
-            inputs: vec!["name".to_string()],
-        }
-    }
-
-    /// A root definition keeps the pre-namespace `select` form byte-stably;
-    /// a namespaced one moves off `select`, which sends pre-namespace readers
-    /// to the `Unrecognized` refusal instead of a root resolve.
-    #[test]
-    fn a_namespaced_definition_is_refused_by_the_pre_namespace_reader() {
-        let root = definition_to_metadata(&definition(Vec::new())).unwrap();
-        let root: serde_json::Value = serde_json::from_str(&root).unwrap();
-        assert_eq!(root["kind"], "select");
-        assert!(
-            root.get("source_namespace").is_none(),
-            "a root definition must not grow new keys: {root}"
-        );
-
-        let stored = definition_to_metadata(&definition(vec!["ns".to_string()])).unwrap();
-        let value: serde_json::Value = serde_json::from_str(&stored).unwrap();
-        // The pre-namespace discriminator is `kind == "select"`; anything
-        // else lands in its Unrecognized refusal rather than in a root open.
-        assert_eq!(value["kind"], "namespaced_select");
-
-        // The current reader round-trips the coordinate.
-        let metadata = HashMap::from([(DEFINITION_META_KEY.to_string(), stored)]);
-        match materialized_view_kind(&metadata).unwrap() {
-            Some(MaterializedViewKind::Select(read)) => {
-                assert_eq!(read.source_namespace, vec!["ns".to_string()])
-            }
-            other => panic!("expected the namespaced select form, got {other:?}"),
-        }
-    }
-
-    /// A kind that disagrees with its namespace is an error, not a view:
-    /// under `select` it is the shape old readers would resolve at the root.
-    #[test]
-    fn a_kind_namespace_mismatch_is_refused() {
-        for (kind, namespace) in [
-            (SELECT_KIND, vec!["ns".to_string()]),
-            (NAMESPACED_SELECT_KIND, Vec::new()),
-        ] {
-            let mut value = serde_json::to_value(definition(namespace)).unwrap();
-            value["kind"] = serde_json::Value::String(kind.to_string());
-            let metadata = HashMap::from([(DEFINITION_META_KEY.to_string(), value.to_string())]);
-            let err = materialized_view_kind(&metadata).unwrap_err();
-            assert!(
-                err.to_string()
-                    .contains("does not match its source namespace"),
-                "kind '{kind}': {err}"
-            );
-        }
+            .unwrap_err();
+        assert!(err.to_string().contains("namespaced source"), "{err}");
    }
 }
@@ -170,7 +170,6 @@ pub(crate) async fn execute_refresh(
    let (replanned, mut planned_fields, _renames) = super::plan(
        source_schema,
        &definition.source_table,
-        &definition.source_namespace,
        &projections,
        definition.filter.as_deref(),
        definition.limit,
@@ -591,7 +590,7 @@ async fn open_source(view: &Table, definition: &MaterializedViewDefinition) -> R
    let source = database
        .open_table(OpenTableRequest {
            name: definition.source_table.clone(),
-            namespace_path: definition.source_namespace.clone(),
+            namespace_path: Vec::new(),
            index_cache_size: None,
            lance_read_params: None,
            location: None,
@@ -2920,7 +2919,6 @@ mod tests {

        let replacement = crate::materialized_view::MaterializedViewDefinition {
            source_table: "src".into(),
-            source_namespace: Vec::new(),
            projections: vec![
                crate::materialized_view::ViewProjection {
                    output: "x".into(),
@@ -2960,7 +2958,6 @@ mod tests {

        let narrower = crate::materialized_view::MaterializedViewDefinition {
            source_table: "src".into(),
-            source_namespace: Vec::new(),
            projections: vec![crate::materialized_view::ViewProjection {
                output: "x".into(),
                expression: "x".into(),
@@ -1,7 +1,7 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright The LanceDB Authors

-use std::collections::{HashMap, HashSet};
+use std::collections::HashMap;
 use std::sync::Arc;

 use async_trait::async_trait;
@@ -26,7 +26,9 @@ use crate::database::{
 use crate::error::Result;
 use crate::function::{FunctionRegistrationRequest, FunctionVersion};
 use crate::job::Job;
-use crate::remote::job::{DescribeJobResponse, RemoteJob, job_state_to_client};
+use crate::remote::job::{
+    DescribeJobResponse, PauseJobResponse, RemoteJob, ResumeJobResponse, job_state_to_client,
+};
 use crate::remote::util::stream_as_body;
 use crate::table::BaseTable;

@@ -533,19 +535,6 @@ struct RemoteListJobsResponse {
    page_token: Option<String>,
 }

-#[derive(serde::Deserialize)]
-struct RemoteListedFunctionVersion {
-    definition: FunctionVersion,
-}
-
-#[derive(serde::Deserialize)]
-struct RemoteListFunctionsResponse {
-    #[serde(default)]
-    functions: Vec<RemoteListedFunctionVersion>,
-    #[serde(default)]
-    page_token: Option<String>,
-}
-
 #[derive(serde::Deserialize)]
 struct RemoteDropFunctionResponse {
    dropped: bool,
@@ -601,43 +590,6 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
        response.json().await.err_to_http(request_id)
    }

-    async fn list_functions(&self) -> Result<Vec<FunctionVersion>> {
-        let mut functions = Vec::new();
-        let mut page_token: Option<String> = None;
-        let mut seen_page_tokens = HashSet::new();
-        loop {
-            let mut body = serde_json::json!({ "include_definition": true });
-            if let Some(token) = &page_token {
-                body["page_token"] = serde_json::Value::String(token.clone());
-            }
-            let req = self.client.post("/v1/functions/list").json(&body);
-            let (request_id, response) = self.client.send(req).await?;
-            let response = self.client.check_response(&request_id, response).await?;
-            let status = response.status();
-            let response: RemoteListFunctionsResponse =
-                response.json().await.err_to_http(request_id.clone())?;
-            functions.extend(
-                response
-                    .functions
-                    .into_iter()
-                    .map(|listed| listed.definition),
-            );
-            let Some(next_page_token) = response.page_token.filter(|token| !token.is_empty())
-            else {
-                break;
-            };
-            if !seen_page_tokens.insert(next_page_token.clone()) {
-                return Err(Error::Http {
-                    source: "Function listing response repeated a page_token".into(),
-                    request_id,
-                    status_code: Some(status),
-                });
-            }
-            page_token = Some(next_page_token);
-        }
-        Ok(functions)
-    }
-
    async fn drop_function(&self, name: &str, version: &str) -> Result<bool> {
        let req = self
            .client
@@ -734,6 +686,40 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
        }
    }

+    async fn pause_job(&self, job_id: &str) -> Result<crate::database::PauseJobStatus> {
+        let req = self
+            .client
+            .post("/v1/jobs/pause")
+            .json(&serde_json::json!({ "job_id": job_id }));
+        let (request_id, rsp) = self.client.send(req).await?;
+        let rsp = self.client.check_response(&request_id, rsp).await?;
+        let body: PauseJobResponse = rsp.json().await.err_to_http(request_id)?;
+        Ok(if body.paused {
+            crate::database::PauseJobStatus::Pausing
+        } else if body.committing {
+            crate::database::PauseJobStatus::Committing
+        } else {
+            crate::database::PauseJobStatus::AlreadyPaused
+        })
+    }
+
+    async fn resume_job(&self, job_id: &str) -> Result<crate::database::ResumeJobStatus> {
+        let req = self
+            .client
+            .post("/v1/jobs/resume")
+            .json(&serde_json::json!({ "job_id": job_id }));
+        let (request_id, rsp) = self.client.send(req).await?;
+        let rsp = self.client.check_response(&request_id, rsp).await?;
+        let body: ResumeJobResponse = rsp.json().await.err_to_http(request_id)?;
+        Ok(if body.resumed {
+            crate::database::ResumeJobStatus::Resumed
+        } else if body.still_pausing {
+            crate::database::ResumeJobStatus::StillPausing
+        } else {
+            crate::database::ResumeJobStatus::NotPaused
+        })
+    }
+
    async fn job_history(&self, job_id: Option<&str>) -> Result<Vec<arrow_array::RecordBatch>> {
        let mut body = serde_json::json!({});
        if let Some(job_id) = job_id {
@@ -2669,6 +2655,45 @@ mod tests {
        assert!(!conn.cancel_job("nope").await.unwrap());
    }

+    #[tokio::test]
+    async fn test_pause_and_resume_job() {
+        use crate::database::{PauseJobStatus, ResumeJobStatus};
+        let conn = Connection::new_with_handler(|request| {
+            assert_eq!(request.url().path(), "/v1/jobs/pause");
+            http::Response::builder()
+                .status(200)
+                .body(r#"{"job_id": "job-1", "paused": true}"#)
+                .unwrap()
+        });
+        assert_eq!(
+            conn.pause_job("job-1").await.unwrap(),
+            PauseJobStatus::Pausing
+        );
+
+        let conn = Connection::new_with_handler(|_| {
+            http::Response::builder()
+                .status(200)
+                .body(r#"{"job_id": "job-1", "paused": false, "committing": true}"#)
+                .unwrap()
+        });
+        assert_eq!(
+            conn.pause_job("job-1").await.unwrap(),
+            PauseJobStatus::Committing
+        );
+
+        let conn = Connection::new_with_handler(|request| {
+            assert_eq!(request.url().path(), "/v1/jobs/resume");
+            http::Response::builder()
+                .status(200)
+                .body(r#"{"job_id": "job-1", "resumed": false, "still_pausing": true}"#)
+                .unwrap()
+        });
+        assert_eq!(
+            conn.resume_job("job-1").await.unwrap(),
+            ResumeJobStatus::StillPausing
+        );
+    }
+
    #[tokio::test]
    async fn test_job_history_parses_arrow_stream() {
        let schema = Arc::new(Schema::new(vec![Field::new(
@@ -2758,119 +2783,6 @@ mod tests {
        assert_eq!(version.version(), "fv_01K3EXACT");
    }

-    #[tokio::test]
-    async fn test_list_functions_requests_definitions_and_paginates() {
-        const VERSION: &str = include_str!(
-            "../../tests/fixtures/first_class_functions/v1/remote_function_version.canonical.json"
-        );
-        let version: serde_json::Value = serde_json::from_str(VERSION).unwrap();
-        let page = Arc::new(AtomicUsize::new(0));
-        let conn = Connection::new_with_handler(move |request| {
-            assert_eq!(request.method(), &reqwest::Method::POST);
-            assert_eq!(request.url().path(), "/v1/functions/list");
-            let body: serde_json::Value =
-                serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()).unwrap();
-            assert_eq!(body["include_definition"], true);
-            match page.fetch_add(1, Ordering::SeqCst) {
-                0 => {
-                    assert!(body.get("page_token").is_none());
-                    http::Response::builder()
-                        .status(200)
-                        .body(r#"{"functions": [], "page_token": "next"}"#.to_string())
-                        .unwrap()
-                }
-                _ => {
-                    assert_eq!(body["page_token"], "next");
-                    http::Response::builder()
-                        .status(200)
-                        .body(
-                            serde_json::json!({
-                                "functions": [{
-                                    "name": "embed",
-                                    "version": "fv_01K3EXACT",
-                                    "definition": version.clone(),
-                                }],
-                            })
-                            .to_string(),
-                        )
-                        .unwrap()
-                }
-            }
-        });
-        let functions = conn.list_functions().await.unwrap();
-        assert_eq!(functions.len(), 1);
-        assert_eq!(functions[0].name(), "embed");
-        assert_eq!(functions[0].version(), "fv_01K3EXACT");
-    }
-
-    #[tokio::test]
-    async fn test_list_functions_stops_on_an_empty_page_token() {
-        let requests = Arc::new(AtomicUsize::new(0));
-        let seen = requests.clone();
-        let conn = Connection::new_with_handler(move |request| {
-            seen.fetch_add(1, Ordering::SeqCst);
-            let body: serde_json::Value =
-                serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()).unwrap();
-            assert!(body.get("page_token").is_none());
-            http::Response::builder()
-                .status(200)
-                .body(r#"{"functions": [], "page_token": ""}"#)
-                .unwrap()
-        });
-
-        let functions = conn.list_functions().await.unwrap();
-        assert!(functions.is_empty());
-        assert_eq!(requests.load(Ordering::SeqCst), 1);
-    }
-
-    #[tokio::test]
-    async fn test_list_functions_rejects_a_page_token_cycle() {
-        let page = Arc::new(AtomicUsize::new(0));
-        let requests = page.clone();
-        let conn = Connection::new_with_handler(move |request| {
-            let body: serde_json::Value =
-                serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()).unwrap();
-            let next_page_token = match page.fetch_add(1, Ordering::SeqCst) {
-                0 => {
-                    assert!(body.get("page_token").is_none());
-                    "one"
-                }
-                1 => {
-                    assert_eq!(body["page_token"], "one");
-                    "two"
-                }
-                2 => {
-                    assert_eq!(body["page_token"], "two");
-                    "one"
-                }
-                page => panic!("unexpected page: {page}"),
-            };
-            http::Response::builder()
-                .status(200)
-                .body(
-                    serde_json::json!({
-                        "functions": [],
-                        "page_token": next_page_token,
-                    })
-                    .to_string(),
-                )
-                .unwrap()
-        });
-
-        let error = conn.list_functions().await.unwrap_err();
-        assert!(
-            matches!(
-                &error,
-                Error::Http {
-                    status_code: Some(http::StatusCode::OK),
-                    ..
-                }
-            ),
-            "got {error:?}"
-        );
-        assert_eq!(requests.load(Ordering::SeqCst), 3);
-    }
-
    #[tokio::test]
    async fn test_drop_function_sends_exact_version_and_decodes_replay() {
        let conn = Connection::new_with_handler(|request| {
@@ -73,6 +73,28 @@ pub(super) struct ReportedFailure {
    retryable: Option<bool>,
 }

+/// Forward-compatible `/v1/jobs/pause` wire envelope.
+#[derive(Deserialize)]
+pub(super) struct PauseJobResponse {
+    /// False when the job was already paused, so a repeated pause changed
+    /// nothing.
+    #[serde(default)]
+    pub(super) paused: bool,
+    /// The job is finalizing its results and cannot be parked right now.
+    #[serde(default)]
+    pub(super) committing: bool,
+}
+
+/// Forward-compatible `/v1/jobs/resume` wire envelope.
+#[derive(Deserialize)]
+pub(super) struct ResumeJobResponse {
+    #[serde(default)]
+    pub(super) resumed: bool,
+    /// The pause's worker drain is not confirmed yet.
+    #[serde(default)]
+    pub(super) still_pausing: bool,
+}
+
 /// Forward-compatible `/v1/jobs/describe` wire envelope.
 #[derive(Deserialize)]
 pub(super) struct DescribeJobResponse {
@@ -72,7 +72,7 @@ use lance_datafusion::exec::{OneShotExec, execute_plan};
 use reqwest::{RequestBuilder, Response};
 use serde::{Deserialize, Serialize};
 use serde_json::Number;
-use std::collections::{HashMap, HashSet};
+use std::collections::HashMap;
 use std::io::Cursor;
 use std::pin::Pin;
 use std::sync::{Arc, Mutex};
@@ -524,8 +524,7 @@ impl<S: HttpSend> RemoteTable<S> {
            _ => resolve_arrow_field_path(&schema, &column)?,
        };
        let mut body = serde_json::json!({
-            "column": canonical_column,
-            "replace": index.replace,
+            "column": canonical_column
        });

        // Add name parameter if provided (for backwards compatibility, only include if Some)
@@ -3648,12 +3647,7 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {

 #[derive(Serialize, Clone, Debug)]
 pub struct MergeInsertRequest {
-    // Sent as one repeated `on` query parameter per column, which is how the
-    // namespace spec encodes an array-valued `on`. serde_urlencoded (which
-    // reqwest's `query()` uses) cannot serialize a sequence nested in a struct,
-    // so this field is emitted separately by [`Self::on_query_params`].
-    #[serde(skip_serializing)]
-    on: Vec<String>,
+    on: String,
    when_matched_update_all: bool,
    when_matched_update_all_filt: Option<String>,
    when_not_matched_insert_all: bool,
@@ -3669,17 +3663,6 @@ pub struct MergeInsertRequest {
    use_lsm: Option<bool>,
 }

-impl MergeInsertRequest {
-    /// The `on` columns as repeated query parameters: `?on=a&on=b`.
-    ///
-    /// A single column serializes to `?on=a`, exactly what clients sent before
-    /// `on` became a list, so a server that predates composite keys sees no
-    /// change from a single-column caller.
-    pub(crate) fn on_query_params(&self) -> Vec<(&str, &str)> {
-        self.on.iter().map(|col| ("on", col.as_str())).collect()
-    }
-}
-
 fn is_true(b: &bool) -> bool {
    *b
 }
@@ -3692,15 +3675,12 @@ impl TryFrom<MergeInsertBuilder> for MergeInsertRequest {
            return Err(Error::InvalidInput {
                message: "MergeInsertBuilder missing required 'on' field".into(),
            });
-        }
-        // The server rejects a repeated column with a 400; catching it here
-        // names the offending column and costs no round trip.
-        let mut seen = HashSet::with_capacity(value.on.len());
-        if let Some(dup) = value.on.iter().find(|col| !seen.insert(*col)) {
-            return Err(Error::InvalidInput {
-                message: format!("MergeInsertBuilder 'on' column '{dup}' is repeated"),
+        } else if value.on.len() > 1 {
+            return Err(Error::NotSupported {
+                message: "MergeInsertBuilder only supports a single 'on' column".into(),
            });
        }
+        let on = value.on[0].clone();

        let when_matched_update_all_filt = match value.when_matched_update_all_filt {
            Some(MergeFilter::Sql(sql)) => Some(sql),
@@ -3724,7 +3704,7 @@ impl TryFrom<MergeInsertBuilder> for MergeInsertRequest {
            };

        Ok(Self {
-            on: value.on,
+            on,
            when_matched_update_all: value.when_matched_update_all,
            when_matched_update_all_filt,
            when_not_matched_insert_all: value.when_not_matched_insert_all,
@@ -4568,76 +4548,6 @@ mod tests {
        }
    }

-    #[tokio::test]
-    async fn test_merge_insert_composite_key() {
-        let batch = RecordBatch::try_new(
-            Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, false)])),
-            vec![Arc::new(Int32Array::from(vec![1, 2, 3]))],
-        )
-        .unwrap();
-        let data: Box<dyn RecordBatchReader + Send> = Box::new(RecordBatchIterator::new(
-            [Ok(batch.clone())],
-            batch.schema(),
-        ));
-
-        let table = Table::new_with_handler("my_table", move |request| {
-            assert_eq!(request.url().path(), "/v1/table/my_table/merge_insert/");
-
-            // One repeated `on` per column, in the order the caller gave them.
-            let on = request
-                .url()
-                .query_pairs()
-                .filter(|(key, _)| key == "on")
-                .map(|(_, value)| value.into_owned())
-                .collect::<Vec<_>>();
-            assert_eq!(on, vec!["shard_key".to_string(), "id".to_string()]);
-
-            let params = request.url().query_pairs().collect::<HashMap<_, _>>();
-            assert_eq!(params["when_matched_update_all"], "true");
-            assert_eq!(params["when_not_matched_insert_all"], "true");
-
-            http::Response::builder()
-                .status(200)
-                .body(r#"{"version": 43, "num_deleted_rows": 0, "num_inserted_rows": 3, "num_updated_rows": 0}"#)
-                .unwrap()
-        });
-
-        let mut merge = table.merge_insert(&["shard_key", "id"]);
-        merge.when_matched_update_all(None);
-        merge.when_not_matched_insert_all();
-        let result = table.base_table().merge_insert(merge, data).await.unwrap();
-
-        assert_eq!(result.num_inserted_rows, 3);
-    }
-
-    #[tokio::test]
-    async fn test_merge_insert_rejects_repeated_on_column() {
-        let batch = RecordBatch::try_new(
-            Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, false)])),
-            vec![Arc::new(Int32Array::from(vec![1]))],
-        )
-        .unwrap();
-        let data: Box<dyn RecordBatchReader + Send> = Box::new(RecordBatchIterator::new(
-            [Ok(batch.clone())],
-            batch.schema(),
-        ));
-
-        let table = Table::new_with_handler::<&str>("my_table", |request| {
-            panic!("Unexpected request: {}", request.url());
-        });
-
-        let merge = table.merge_insert(&["id", "id"]);
-        let err = table
-            .base_table()
-            .merge_insert(merge, data)
-            .await
-            .unwrap_err();
-        assert!(
-            matches!(&err, Error::InvalidInput { message } if message.contains("'id' is repeated")),
-            "unexpected error: {err}"
-        );
-    }
-
    #[tokio::test]
    async fn test_merge_insert_retries_on_409() {
        let batch = RecordBatch::try_new(
@@ -6306,7 +6216,6 @@ mod tests {
                            let mut expected_body = expected_body.clone();
                            expected_body["column"] = "a".into();
                            expected_body[INDEX_TYPE_KEY] = index_type.into();
-                            expected_body["replace"] = false.into();

                            assert_eq!(body, expected_body);

@@ -6324,73 +6233,6 @@ mod tests {
        }
    }

-    #[tokio::test]
-    async fn test_create_index_forwards_default_replace_false_on_existing_route() {
-        let table = Table::new_with_handler("my_table", move |request| {
-            assert_eq!(request.method(), "POST");
-            match request.url().path() {
-                "/v1/table/my_table/describe/" => {
-                    let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]);
-                    http::Response::builder()
-                        .status(200)
-                        .body(describe_response(&schema))
-                        .unwrap()
-                }
-                "/v1/table/my_table/create_index/" => {
-                    let body = request.body().unwrap().as_bytes().unwrap();
-                    let body: serde_json::Value = serde_json::from_slice(body).unwrap();
-                    assert_eq!(body["replace"], json!(false));
-
-                    http::Response::builder()
-                        .status(200)
-                        .body("{}".to_string())
-                        .unwrap()
-                }
-                path => panic!("Unexpected path: {}", path),
-            }
-        });
-
-        table
-            .create_index(&["a"], Index::BTree(Default::default()))
-            .execute()
-            .await
-            .unwrap();
-    }
-
-    #[tokio::test]
-    async fn test_create_index_forwards_explicit_replace_true_on_existing_route() {
-        let table = Table::new_with_handler("my_table", move |request| {
-            assert_eq!(request.method(), "POST");
-            match request.url().path() {
-                "/v1/table/my_table/describe/" => {
-                    let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]);
-                    http::Response::builder()
-                        .status(200)
-                        .body(describe_response(&schema))
-                        .unwrap()
-                }
-                "/v1/table/my_table/create_index/" => {
-                    let body = request.body().unwrap().as_bytes().unwrap();
-                    let body: serde_json::Value = serde_json::from_slice(body).unwrap();
-                    assert_eq!(body["replace"], json!(true));
-
-                    http::Response::builder()
-                        .status(200)
-                        .body("{}".to_string())
-                        .unwrap()
-                }
-                path => panic!("Unexpected path: {}", path),
-            }
-        });
-
-        table
-            .create_index(&["a"], Index::BTree(Default::default()))
-            .replace(true)
-            .execute()
-            .await
-            .unwrap();
-    }
-
    #[tokio::test]
    async fn test_create_index_returns_job() {
        let describe_calls = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
@@ -6616,46 +6458,38 @@ mod tests {
            json!({
                "column": "rowId",
                "index_type": "BTREE",
-                "replace": false,
            }),
            json!({
                "column": "`row-id`",
                "index_type": "BTREE",
-                "replace": false,
            }),
            json!({
                "column": "userId",
                "index_type": "BTREE",
-                "replace": false,
            }),
            json!({
                "column": "MetaData.userId",
                "index_type": "BTREE",
-                "replace": false,
            }),
            json!({
                "column": "metadata.user_id",
                "index_type": "BTREE",
-                "replace": false,
            }),
            json!({
                "column": "image.embedding",
                "index_type": "IVF_PQ",
                "metric_type": "l2",
-                "replace": false,
            }),
            {
                let mut body = serde_json::to_value(InvertedIndexParams::default()).unwrap();
                body["column"] = "payload.text".into();
                body["index_type"] = "FTS".into();
-                body["replace"] = false.into();
                body
            },
            {
                let mut body = serde_json::to_value(InvertedIndexParams::default()).unwrap();
                body["column"] = "docs.content".into();
                body["index_type"] = "FTS".into();
-                body["replace"] = false.into();
                body
            },
            {
@@ -6663,18 +6497,15 @@ mod tests {
                body["column"] = "docs.content".into();
                body["index_type"] = "FTS".into();
                body["document_granularity"] = "list_element".into();
-                body["replace"] = false.into();
                body
            },
            json!({
                "column": "`meta-data`.`user-id`",
                "index_type": "BTREE",
-                "replace": false,
            }),
            json!({
                "column": "literal.`a.b`",
                "index_type": "BTREE",
-                "replace": false,
            }),
        ]);
        let request_idx = Arc::new(AtomicUsize::new(0));
@@ -734,7 +734,6 @@ impl<S: HttpSend + 'static> ExecutionPlan for RemoteWriteExec<S> {
                WriteOp::MergeInsert { query, timeout } => {
                    let mut request = client
                        .post(&format!("/v1/table/{}/merge_insert/", identifier))
-                        .query(&query.on_query_params())
                        .query(query)
                        .header(CONTENT_TYPE, ARROW_STREAM_CONTENT_TYPE);
                    if let Some(timeout) = timeout {
@@ -1490,7 +1489,7 @@ mod tests {
        });

        let query = MergeInsertRequest {
-            on: vec!["id".to_string()],
+            on: "id".to_string(),
            when_matched_update_all: false,
            when_matched_update_all_filt: None,
            when_not_matched_insert_all: false,
@@ -1506,9 +1506,7 @@ impl Table {
    ///
    /// * `on` One or more columns to join on.  This is how records from the
    ///   source table and target table are matched.  Typically this is some
-    ///   kind of key or id column.  Several columns match on the composite
-    ///   key: a source row updates a target row only when it agrees on every
-    ///   one of them.
+    ///   kind of key or id column.
    ///
    /// # Examples
    ///
@@ -589,13 +589,16 @@ fn canonical_input_arrow_type(field: &JsonArrowField) -> Result<String> {
        .and_then(|metadata| metadata.get(ARROW_EXT_NAME_KEY))
        .map(String::as_str)
        == Some(BLOB_V2_EXT_NAME);
-    if is_blob_v2 || field.r#type.fields.is_some() {
+    if is_blob_v2 {
        let arrow_field = lance_namespace::schema::convert_json_arrow_field(field)
            .map_err(|e| invalid_function(format!("invalid Function input field: {e}")))?;
-        validate_function_blob_nesting(&arrow_field, false)?;
-        if is_blob_v2 {
-            return Ok(FUNCTION_BLOB_V2_TYPE.to_string());
+        if !has_supported_blob_v2_layout(&arrow_field) {
+            return Err(invalid_function(format!(
+                "Function input '{}' has an invalid Blob v2 storage layout",
+                arrow_field.name()
+            )));
        }
+        return Ok(FUNCTION_BLOB_V2_TYPE.to_string());
    }
    if field.r#type.fields.is_none() && field.r#type.length.is_none() {
        Ok(field.r#type.r#type.clone())
@@ -614,34 +617,6 @@ fn has_supported_blob_v2_layout(field: &ArrowField) -> bool {
        )
 }

-fn validate_function_blob_nesting(field: &ArrowField, inside_collection: bool) -> Result<()> {
-    if field.is_blob_v2() {
-        if inside_collection {
-            return Err(invalid_function(format!(
-                "Function field '{}' nests Blob v2 under a collection, which Function signatures do not support",
-                field.name()
-            )));
-        }
-        if !has_supported_blob_v2_layout(field) {
-            return Err(invalid_function(format!(
-                "Function field '{}' has an invalid Blob v2 storage layout",
-                field.name()
-            )));
-        }
-        return Ok(());
-    }
-    match field.data_type() {
-        DataType::Struct(fields) => fields
-            .iter()
-            .try_for_each(|field| validate_function_blob_nesting(field, inside_collection)),
-        DataType::List(field)
-        | DataType::LargeList(field)
-        | DataType::FixedSizeList(field, _)
-        | DataType::Map(field, _) => validate_function_blob_nesting(field, true),
-        _ => Ok(()),
-    }
-}
-
 /// `fixed_size_list<item, size>` -> (`item`, `size`); the comma must sit outside
 /// any nested `<...>`.
 fn split_fixed_size_list(raw: &str) -> Option<(&str, i32)> {
@@ -722,22 +697,21 @@ fn parse_output_arrow_type(raw: &str) -> Result<JsonArrowDataType> {
 }

 fn function_output_field(name: &str, nullable: bool, raw: &str) -> Result<JsonArrowField> {
-    let field = if raw == FUNCTION_BLOB_V2_TYPE {
-        lance_namespace::schema::arrow_schema_to_json(&ArrowSchema::new(vec![crate::blob(
-            name, nullable,
-        )]))
+    if raw == FUNCTION_BLOB_V2_TYPE {
+        return lance_namespace::schema::arrow_schema_to_json(&ArrowSchema::new(vec![
+            crate::blob(name, nullable),
+        ]))
        .map_err(|e| invalid_function(format!("could not encode Blob v2 output field: {e}")))?
        .fields
        .into_iter()
        .next()
-        .ok_or_else(|| invalid_function("Blob v2 output field is missing"))?
-    } else {
-        JsonArrowField::new(name.to_string(), nullable, parse_output_arrow_type(raw)?)
-    };
-    let arrow_field = lance_namespace::schema::convert_json_arrow_field(&field)
-        .map_err(|e| invalid_function(format!("invalid Function output field: {e}")))?;
-    validate_function_blob_nesting(&arrow_field, false)?;
-    Ok(field)
+        .ok_or_else(|| invalid_function("Blob v2 output field is missing"));
+    }
+    Ok(JsonArrowField::new(
+        name.to_string(),
+        nullable,
+        parse_output_arrow_type(raw)?,
+    ))
 }

 fn function_output_field_matches(expected: &ArrowField, actual: &ArrowField) -> bool {
@@ -2745,28 +2719,6 @@ mod tests {
        .unwrap()
    }

-    fn exact_arrow_type(field: ArrowField) -> String {
-        let json =
-            lance_namespace::schema::arrow_schema_to_json(&ArrowSchema::new(vec![field])).unwrap();
-        serde_json::to_string(json.fields[0].r#type.as_ref()).unwrap()
-    }
-
-    fn single_input_application(path: &str) -> FunctionApplication {
-        FunctionApplication::from_json(
-            &serde_json::json!({
-                "function": {"name": "inspect", "version": "fv_nested_blob"},
-                "inputs": [{
-                    "parameter": "value",
-                    "kind": "column",
-                    "value": {"path": path}
-                }],
-                "output": {"kind": "scalar", "arrow_type": "int64", "nullable": false}
-            })
-            .to_string(),
-        )
-        .unwrap()
-    }
-
    fn binding_from_plan(plan: &FunctionDeclarationPlan) -> FunctionBinding {
        let inputs = plan
            .input_bindings
@@ -3206,128 +3158,6 @@ mod tests {
        assert_eq!(fields[1].data_type(), &DataType::Int32);
    }

-    #[test]
-    fn test_struct_blob_input_preserves_exact_schema_and_nullability() {
-        let payload = ArrowField::new(
-            "payload",
-            DataType::Struct(Fields::from(vec![
-                ArrowField::new("mime_type", DataType::Utf8, false),
-                ArrowField::new(
-                    "nested",
-                    DataType::Struct(Fields::from(vec![crate::blob("image", true)])),
-                    true,
-                ),
-            ])),
-            true,
-        );
-        let plan = plan_function_application(
-            &ArrowSchema::new(vec![payload]),
-            &single_input_application("payload"),
-            Some("size"),
-        )
-        .unwrap();
-
-        let declared: JsonArrowDataType =
-            serde_json::from_str(&plan.input_bindings[0].arrow_type).unwrap();
-        let DataType::Struct(fields) =
-            lance_namespace::schema::convert_json_arrow_type(&declared).unwrap()
-        else {
-            panic!("expected a struct Function input")
-        };
-        assert!(fields[1].is_nullable());
-        let DataType::Struct(nested) = fields[1].data_type() else {
-            panic!("expected a recursive struct Function input")
-        };
-        assert!(nested[0].is_blob_v2());
-        assert!(nested[0].is_nullable());
-
-        let exact = lance_namespace::schema::convert_json_arrow_schema(&plan.input_schema).unwrap();
-        let DataType::Struct(fields) = exact.field(0).data_type() else {
-            panic!("expected exact input schema to retain the struct")
-        };
-        let DataType::Struct(nested) = fields[1].data_type() else {
-            panic!("expected exact input schema to retain the nested struct")
-        };
-        assert!(nested[0].is_blob_v2());
-    }
-
-    #[test]
-    fn test_recursive_blob_result_plans_one_whole_named_struct_column() {
-        let details_type = exact_arrow_type(ArrowField::new(
-            "details",
-            DataType::Struct(Fields::from(vec![crate::blob("image", true)])),
-            false,
-        ));
-        let application = FunctionApplication::from_json(
-            &serde_json::json!({
-                "function": {"name": "inspect", "version": "fv_nested_blob"},
-                "inputs": [],
-                "output": {
-                    "kind": "named_struct",
-                    "fields": [
-                        {"name": "mime_type", "arrow_type": "utf8", "nullable": false},
-                        {"name": "details", "arrow_type": details_type, "nullable": false}
-                    ]
-                }
-            })
-            .to_string(),
-        )
-        .unwrap();
-        let plan = plan_function_application(&ArrowSchema::empty(), &application, Some("payload"))
-            .unwrap();
-
-        assert_eq!(plan.outputs.len(), 1);
-        assert_eq!(plan.outputs[0].result_field, WHOLE_RESULT_FIELD);
-        let schema =
-            lance_namespace::schema::convert_json_arrow_schema(&plan.output_schema).unwrap();
-        assert_eq!(schema.field(0).name(), "payload");
-        let DataType::Struct(fields) = schema.field(0).data_type() else {
-            panic!("whole named result must be one struct column")
-        };
-        assert_eq!(
-            fields.iter().map(|field| field.name()).collect::<Vec<_>>(),
-            ["mime_type", "details"]
-        );
-        let DataType::Struct(details) = fields[1].data_type() else {
-            panic!("expected recursive result struct")
-        };
-        assert!(details[0].is_blob_v2());
-        assert!(!fields.iter().any(|field| field.name() == "payload"));
-    }
-
-    #[test]
-    fn test_blob_children_under_collections_are_rejected() {
-        let collections = vec![
-            DataType::List(Arc::new(crate::blob("item", false))),
-            DataType::LargeList(Arc::new(crate::blob("item", false))),
-            DataType::FixedSizeList(Arc::new(crate::blob("item", false)), 2),
-            DataType::Map(
-                Arc::new(ArrowField::new(
-                    "entries",
-                    DataType::Struct(Fields::from(vec![
-                        ArrowField::new("key", DataType::Utf8, false),
-                        crate::blob("value", false),
-                    ])),
-                    false,
-                )),
-                false,
-            ),
-        ];
-        for data_type in collections {
-            let schema = ArrowSchema::new(vec![ArrowField::new("value", data_type, false)]);
-            let error = plan_function_application(
-                &schema,
-                &single_input_application("value"),
-                Some("size"),
-            )
-            .unwrap_err();
-            assert!(
-                error.to_string().contains("under a collection"),
-                "got: {error}"
-            );
-        }
-    }
-
    #[test]
    fn test_blob_whole_struct_binding_accepts_full_logical_layout() {
        let input = crate::blob("image", false);
@@ -133,7 +133,7 @@ impl NativeTable {
                    ),
                });
            }
-            (resolved.canonical_path, resolved.terminal_field)
+            (resolved.canonical_path, resolved.field)
        } else {
            Self::resolve_index_field(dataset.schema(), &opts.columns[0])?
        };
@@ -439,8 +439,7 @@ mod tests {
    use arrow_array::record_batch;
    use arrow_array::{
        Array, ArrayRef, BinaryArray, BooleanArray, FixedSizeListArray, Float32Array, Int32Array,
-        LargeBinaryArray, LargeStringArray, ListArray, RecordBatch, StringArray, StructArray,
-        UInt32Array,
+        LargeBinaryArray, LargeStringArray, RecordBatch, StringArray, StructArray,
    };
    use arrow_data::ArrayDataBuilder;
    use arrow_schema::{DataType, Field, Schema};
@@ -459,7 +458,6 @@ mod tests {
    use crate::query::{ExecutableQuery, QueryBase};
    use crate::table::optimize::{CompactionOptions, OptimizeAction};
    use lance_index::scalar::FullTextSearchQuery;
-    use lance_index::scalar::inverted::query::{FtsQuery, MatchQuery};

    fn create_fixed_size_list<T: Array>(
        values: T,
@@ -601,80 +599,6 @@ mod tests {
        assert!(invalid_granularity.is_err());
    }

-    #[tokio::test]
-    async fn test_nested_list_fts_uses_deepest_document_coordinates() {
-        let conn = connect("memory://").execute().await.unwrap();
-        let mut docs = ListBuilder::new(ListBuilder::new(StringBuilder::new()));
-
-        docs.values().values().append_value("alpha");
-        docs.values().values().append_value("beta");
-        docs.values().append(true);
-        docs.values().values().append_value("gamma");
-        docs.values().values().append_value("alpha delta");
-        docs.values().append(true);
-        docs.append(true);
-
-        docs.values().append(true);
-        docs.values().values().append_value("alpha");
-        docs.values().append(true);
-        docs.append(true);
-
-        let batch = RecordBatch::try_from_iter(vec![
-            ("id", Arc::new(Int32Array::from(vec![0, 1])) as ArrayRef),
-            ("docs", Arc::new(docs.finish()) as ArrayRef),
-        ])
-        .unwrap();
-        let table = conn.create_table("nested", batch).execute().await.unwrap();
-
-        let job = table
-            .create_index(
-                &["docs"],
-                Index::FTS(
-                    FtsIndexBuilder::default()
-                        .document_granularity(DocumentGranularity::ListElement),
-                ),
-            )
-            .execute_async()
-            .await
-            .unwrap();
-        job.wait().await.unwrap();
-
-        let query = FullTextSearchQuery::new_query(FtsQuery::Match(
-            MatchQuery::new("alpha".to_string())
-                .with_column(Some("docs".to_string()))
-                .with_document_granularity(DocumentGranularity::ListElement),
-        ));
-        let batches = table
-            .query()
-            .full_text_search(query)
-            .limit(10)
-            .execute()
-            .await
-            .unwrap()
-            .try_collect::<Vec<_>>()
-            .await
-            .unwrap();
-
-        let mut hits = Vec::new();
-        for batch in batches {
-            let ids = batch["id"].as_any().downcast_ref::<Int32Array>().unwrap();
-            let coordinates = batch["_doc_index"]
-                .as_any()
-                .downcast_ref::<ListArray>()
-                .unwrap();
-            for row in 0..batch.num_rows() {
-                let coordinate = coordinates.value(row);
-                let coordinate = coordinate.as_any().downcast_ref::<UInt32Array>().unwrap();
-                hits.push((ids.value(row), coordinate.values().to_vec()));
-            }
-        }
-        hits.sort_unstable();
-        assert_eq!(
-            hits,
-            vec![(0, vec![0, 0]), (0, vec![1, 1]), (1, vec![1, 0])]
-        );
-    }
-
    /// Concurrent waiters, and a wait issued after the job settled, all
    /// succeed once the build does.
    #[tokio::test]
@@ -725,11 +649,12 @@ mod tests {
            .await
            .unwrap();

-        // Rebuilding the same index without explicit replace fails once the build
+        // Rebuilding the same index without replace fails once the build
        // starts, so the failure reaches the job rather than execute_async.
        let job = Arc::new(
            table
                .create_index(&["id"], Index::BTree(BTreeIndexBuilder::default()))
+                .replace(false)
                .execute_async()
                .await
                .unwrap(),
@@ -768,6 +693,7 @@ mod tests {

        let job = table
            .create_index(&["id"], Index::BTree(BTreeIndexBuilder::default()))
+            .replace(false)
            .execute_async()
            .await
            .unwrap();
@@ -1104,7 +1030,6 @@ mod tests {
        // Can also specify btree
        table
            .create_index(&["i"], Index::BTree(BTreeIndexBuilder::default()))
-            .replace(true)
            .execute()
            .await
            .unwrap();
@@ -227,7 +227,7 @@ pub(crate) fn resolve_arrow_field_path(schema: &Schema, column: &str) -> Result<

 pub(crate) struct ResolvedFtsField {
    pub canonical_path: String,
-    pub terminal_field: Field,
+    pub field: Field,
    pub list_depth: usize,
 }

@@ -309,7 +309,7 @@ pub(crate) fn resolve_lance_fts_field_path(
    );
    Ok(ResolvedFtsField {
        canonical_path,
-        terminal_field: Field::from(terminal),
+        field: Field::from(field),
        list_depth,
    })
 }
@@ -375,7 +375,7 @@ pub(crate) fn resolve_arrow_fts_field_path(
            message: format!("Invalid schema: {}", e),
        })?;
    let resolved = resolve_lance_fts_field_path(&lance_schema, column)?;
-    Ok((resolved.canonical_path, resolved.terminal_field))
+    Ok((resolved.canonical_path, resolved.field))
 }

 pub fn supported_btree_data_type(dtype: &DataType) -> bool {
@@ -647,9 +647,8 @@ mod tests {
            Field::new("docs", text_list(), true),
        ]);

-        let (path, field) = resolve_arrow_fts_field_path(&schema, "docs.content").unwrap();
+        let (path, _) = resolve_arrow_fts_field_path(&schema, "docs.content").unwrap();
        assert_eq!(path, "docs.content");
-        assert_eq!(field.data_type(), &DataType::Utf8);

        let lance_schema = lance_core::datatypes::Schema::try_from(&schema).unwrap();
        let field_id = lance_schema