fix(remote): forward create index replace flag (#4115 )

Remote create-index requests already expose `replace` on the builder, but the remote client did not consistently forward an explicit `replace=false` over REST. That meant create-only intent could be lost before it reached a remote server, even though local builders and Python APIs can express it. This PR forwards `replace=false` on the existing `create_index` endpoint and keeps the current default behavior unchanged for compatibility. This was accomplished with the following changes: - Serialize `replace: false` into the existing remote create-index request body when the builder is configured with `.replace(false)`. - Forward `replace` through the synchronous Python remote `create_index` wrapper so `RemoteTable.create_index(..., replace=False)` reaches the repaired path. - Continue omitting `replace` for the default path so existing remote create-index requests keep their current semantics. - Document `name` and `replace` on the existing OpenAPI create-index request schema. - Add coverage that verifies the remote client uses the existing `/create_index/` route and forwards `replace=false`, including the synchronous Python unified API. ### Testing - `cargo fmt --all --check` - `cargo test -p lancedb --features remote test_create_index_forwards_replace_false_on_existing_route --locked` - `uv tool run maturin develop --extras tests,dev,embeddings` - `uv run --frozen pytest python/tests/test_remote_db.py::test_remote_create_index_new_api` - `uv run ruff format --check python/lancedb/remote/table.py python/tests/test_remote_db.py` - `cargo build -p lancedb --features remote --locked` - `cargo clippy -p lancedb --features remote --all-targets --locked -- -D warnings`
chore: update lance dependency to v12.0.0-beta.9 (#4116 )
2026-09-01 19:18:38 +00:00 · 2026-09-01 11:54:53 -07:00 · 2026-09-02 00:25:27 +08:00 · 2026-09-01 23:51:08 +08:00 · 2026-09-01 23:50:57 +08:00 · 2026-09-01 13:16:03 +00:00
42 changed files with 1173 additions and 508 deletions
@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "0.38.0"
+current_version = "0.39.0-beta.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.5"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
+version = "12.0.0-beta.9"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.9#6f93e3fe389f5a661a02aff00c14df6be7bf0505"
 dependencies = [
 "arrow-array",
 "rand 0.9.5",
@@ -4815,8 +4815,8 @@ checksum = "e037a2e1d8d5fdbd49b16a4ea09d5d6401c1f29eca5ff29d03d3824dba16256a"

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

 [[package]]
 name = "lance-arrow"
-version = "12.0.0-beta.5"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
+version = "12.0.0-beta.9"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.9#6f93e3fe389f5a661a02aff00c14df6be7bf0505"
 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.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.9#6f93e3fe389f5a661a02aff00c14df6be7bf0505"
 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.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.9#6f93e3fe389f5a661a02aff00c14df6be7bf0505"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -4934,8 +4934,8 @@ dependencies = [

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

 [[package]]
 name = "lancedb-python"
-version = "0.38.0"
+version = "0.39.0-beta.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.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" }
+lance = { "version" = "=12.0.0-beta.9", default-features = false, "tag" = "v12.0.0-beta.9", "git" = "https://github.com/lance-format/lance.git" }
+lance-core = { "version" = "=12.0.0-beta.9", "tag" = "v12.0.0-beta.9", "git" = "https://github.com/lance-format/lance.git" }
+lance-datagen = { "version" = "=12.0.0-beta.9", "tag" = "v12.0.0-beta.9", "git" = "https://github.com/lance-format/lance.git" }
+lance-file = { "version" = "=12.0.0-beta.9", "tag" = "v12.0.0-beta.9", "git" = "https://github.com/lance-format/lance.git" }
+lance-io = { "version" = "=12.0.0-beta.9", default-features = false, "tag" = "v12.0.0-beta.9", "git" = "https://github.com/lance-format/lance.git" }
+lance-index = { "version" = "=12.0.0-beta.9", "tag" = "v12.0.0-beta.9", "git" = "https://github.com/lance-format/lance.git" }
+lance-linalg = { "version" = "=12.0.0-beta.9", "tag" = "v12.0.0-beta.9", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace = { "version" = "=12.0.0-beta.9", "tag" = "v12.0.0-beta.9", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace-impls = { "version" = "=12.0.0-beta.9", default-features = false, "tag" = "v12.0.0-beta.9", "git" = "https://github.com/lance-format/lance.git" }
+lance-table = { "version" = "=12.0.0-beta.9", "tag" = "v12.0.0-beta.9", "git" = "https://github.com/lance-format/lance.git" }
+lance-testing = { "version" = "=12.0.0-beta.9", "tag" = "v12.0.0-beta.9", "git" = "https://github.com/lance-format/lance.git" }
+lance-datafusion = { "version" = "=12.0.0-beta.9", "tag" = "v12.0.0-beta.9", "git" = "https://github.com/lance-format/lance.git" }
+lance-encoding = { "version" = "=12.0.0-beta.9", "tag" = "v12.0.0-beta.9", "git" = "https://github.com/lance-format/lance.git" }
+lance-arrow = { "version" = "=12.0.0-beta.9", "tag" = "v12.0.0-beta.9", "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,6 +446,15 @@ 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.38.0</version>
+    <version>0.39.0-beta.0</version>
 </dependency>
 ```

@@ -50,6 +50,16 @@ 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.38.0-final.0</version>
+      <version>0.39.0-beta.0</version>
      <relativePath>../pom.xml</relativePath>
    </parent>

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

    <groupId>com.lancedb</groupId>
    <artifactId>lancedb-parent</artifactId>
-    <version>0.38.0-final.0</version>
+    <version>0.39.0-beta.0</version>
    <packaging>pom</packaging>
    <name>${project.artifactId}</name>
    <description>LanceDB Java SDK Parent POM</description>
@@ -28,7 +28,7 @@
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <arrow.version>15.0.0</arrow.version>
-        <lance-core.version>12.0.0-beta.5</lance-core.version>
+        <lance-core.version>12.0.0-beta.9</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.38.0"
+version = "0.39.0-beta.0"
 publish = false
 license.workspace = true
 description.workspace = true
@@ -48,6 +48,28 @@ 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)"]],
@@ -987,22 +987,6 @@ describe("remote connection jobs surface", () => {
            res
              .writeHead(200, { "Content-Type": "application/json" })
              .end('{"job_id": "job-1"}');
-          } else if (req.url === "/v1/jobs/pause") {
-            if (payload["job_id"] !== "job-1") {
-              res.writeHead(404).end("no such job");
-              return;
-            }
-            res
-              .writeHead(200, { "Content-Type": "application/json" })
-              .end('{"job_id": "job-1", "paused": true}');
-          } else if (req.url === "/v1/jobs/resume") {
-            if (payload["job_id"] !== "job-1") {
-              res.writeHead(404).end("no such job");
-              return;
-            }
-            res
-              .writeHead(200, { "Content-Type": "application/json" })
-              .end('{"job_id": "job-1", "resumed": false, "still_pausing": true}');
          } else if (req.url === "/v1/jobs/query_events") {
            res
              .writeHead(200, {
@@ -1031,9 +1015,6 @@ describe("remote connection jobs surface", () => {
        expect(await db.cancelJob("job-1")).toBe(true);
        expect(await db.cancelJob("missing")).toBe(false);

-        expect(await db.pauseJob("job-1")).toEqual("pausing");
-        expect(await db.resumeJob("job-1")).toEqual("still_pausing");
-
        const history = await db.jobHistory("job-1");
        expect(history.numRows).toEqual(2);

@@ -583,24 +583,6 @@ export abstract class Connection {
   */
  abstract cancelJob(jobId: string): Promise<boolean>;

-  /**
-   * Pause a server-side job by id.
-   *
-   * The job's workers drain and it stays parked until resumed. Resolves to
-   * "pausing", "already_paused", or "committing" -- a job finalizing its
-   * results cannot be parked; retry shortly.
-   */
-  abstract pauseJob(jobId: string): Promise<string>;
-
-  /**
-   * Resume a paused server-side job by id.
-   *
-   * Its workers pick their work back up from checkpoints. Resolves to
-   * "resumed", "still_pausing" -- the pause's worker drain is not confirmed
-   * yet; retry shortly -- or "not_paused".
-   */
-  abstract resumeJob(jobId: string): Promise<string>;
-
  /**
   * The lifecycle event history of a server-side job, as an Arrow table.
   *
@@ -962,14 +944,6 @@ export class LocalConnection extends Connection {
    return this.inner.cancelJob(jobId);
  }

-  async pauseJob(jobId: string): Promise<string> {
-    return this.inner.pauseJob(jobId);
-  }
-
-  async resumeJob(jobId: string): Promise<string> {
-    return this.inner.resumeJob(jobId);
-  }
-
  async jobHistory(jobId?: string): Promise<ArrowTable> {
    const buf = await this.inner.jobHistory(jobId);
    if (buf.length === 0) {
@@ -19,6 +19,8 @@ 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[];
 }

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

@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-darwin-arm64",
-	"version": "0.38.0",
+	"version": "0.39.0-beta.0",
 	"os": ["darwin"],
 	"cpu": ["arm64"],
 	"main": "lancedb.darwin-arm64.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-gnu",
-	"version": "0.38.0",
+	"version": "0.39.0-beta.0",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-gnu.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-musl",
-	"version": "0.38.0",
+	"version": "0.39.0-beta.0",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-musl.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-gnu",
-	"version": "0.38.0",
+	"version": "0.39.0-beta.0",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-gnu.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-musl",
-	"version": "0.38.0",
+	"version": "0.39.0-beta.0",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-musl.node",
@@ -1,6 +1,6 @@
 {
  "name": "@lancedb/lancedb-win32-arm64-msvc",
-  "version": "0.38.0",
+  "version": "0.39.0-beta.0",
  "os": [
    "win32"
  ],
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-win32-x64-msvc",
-	"version": "0.38.0",
+	"version": "0.39.0-beta.0",
 	"os": ["win32"],
 	"cpu": ["x64"],
 	"main": "lancedb.win32-x64-msvc.node",
@@ -11,7 +11,7 @@
    "ann"
  ],
  "private": false,
-  "version": "0.38.0",
+  "version": "0.39.0-beta.0",
  "main": "dist/index.js",
  "exports": {
    ".": "./dist/index.js",
@@ -477,34 +477,6 @@ impl Connection {
        self.get_inner()?.cancel_job(&job_id).await.default_error()
    }

-    /// Pause a server-side job by id: its workers drain and it stays parked
-    /// until resumed. Returns "pausing", "already_paused", or "committing".
-    #[napi(catch_unwind)]
-    pub async fn pause_job(&self, job_id: String) -> napi::Result<String> {
-        let status = self.get_inner()?.pause_job(&job_id).await.default_error()?;
-        Ok(match status {
-            lancedb::database::PauseJobStatus::Pausing => "pausing".to_string(),
-            lancedb::database::PauseJobStatus::AlreadyPaused => "already_paused".to_string(),
-            lancedb::database::PauseJobStatus::Committing => "committing".to_string(),
-        })
-    }
-
-    /// Resume a paused server-side job by id. Returns "resumed",
-    /// "still_pausing", or "not_paused".
-    #[napi(catch_unwind)]
-    pub async fn resume_job(&self, job_id: String) -> napi::Result<String> {
-        let status = self
-            .get_inner()?
-            .resume_job(&job_id)
-            .await
-            .default_error()?;
-        Ok(match status {
-            lancedb::database::ResumeJobStatus::Resumed => "resumed".to_string(),
-            lancedb::database::ResumeJobStatus::StillPausing => "still_pausing".to_string(),
-            lancedb::database::ResumeJobStatus::NotPaused => "not_paused".to_string(),
-        })
-    }
-
    /// The lifecycle event history of a server-side job (all jobs when
    /// `job_id` is null), as an Arrow IPC stream buffer. Empty when there is
    /// no history.
@@ -1,6 +1,6 @@
 [package]
 name = "lancedb-python"
-version = "0.38.0"
+version = "0.39.0-beta.0"
 publish = false
 edition.workspace = true
 description = "Python bindings for LanceDB"
@@ -150,12 +150,11 @@ 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,6 +712,24 @@ 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.

@@ -753,26 +771,6 @@ 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.

@@ -1443,6 +1441,10 @@ 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))
@@ -1470,22 +1472,6 @@ 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.
@@ -2293,6 +2279,17 @@ 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)
@@ -2317,23 +2314,6 @@ 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,6 +521,12 @@ 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 = (
@@ -591,6 +597,19 @@ 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 "
@@ -655,23 +674,84 @@ def _canonical_arrow_field(field: pa.Field) -> str:
    return _canonical_arrow_type(field.type)


-def _exact_arrow_field(field: pa.Field) -> dict[str, Any]:
+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]:
    _validate_exact_arrow_field(field)
    if _is_blob_v2_field(field):
-        raise TypeError(
-            "unsupported Arrow type for Function signature: nested Blob v2 "
-            "fields are not supported; declare Blob parameters or named result "
-            "fields directly"
-        )
+        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,
+            },
+        }
    value = {
        "name": field.name,
        "nullable": field.nullable,
-        "type": _exact_arrow_type(field.type),
+        "type": _exact_arrow_type(field.type, inside_collection=inside_collection),
    }
    return value


-def _exact_arrow_type(data_type: pa.DataType) -> dict[str, Any]:
+def _exact_arrow_type(
+    data_type: pa.DataType, *, inside_collection: bool = False
+) -> dict[str, Any]:
    for candidate, name in _GRAMMAR_PRIMITIVES:
        if data_type == candidate:
            return {"type": name}
@@ -685,7 +765,10 @@ def _exact_arrow_type(data_type: pa.DataType) -> dict[str, Any]:
            )
        return {
            "type": "struct",
-            "fields": [_exact_arrow_field(field) for field in fields],
+            "fields": [
+                _exact_arrow_field(field, inside_collection=inside_collection)
+                for field in fields
+            ],
        }
    if (
        pa.types.is_list(data_type)
@@ -710,11 +793,15 @@ def _exact_arrow_type(data_type: pa.DataType) -> dict[str, Any]:
                if pa.types.is_large_list(data_type)
                else "fixed_size_list"
            ),
-            "fields": [_exact_arrow_field(data_type.value_field)],
+            "fields": [
+                _exact_arrow_field(data_type.value_field, inside_collection=True)
+            ],
        }
        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,6 +42,8 @@ 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(
@@ -53,7 +55,8 @@ def _definition_from_schema(
        raise ValueError(f"Table '{name}' is not a materialized view")
    value = json.loads(raw)
    kind = value.get("kind")
-    if kind != "select":
+    # "namespaced_select" keeps older readers from resolving the source at root.
+    if kind not in ("select", "namespaced_select"):
        raise NotImplementedError(
            f"materialized view '{name}' is defined by '{kind}', which this "
            "version of lancedb cannot refresh"
@@ -66,6 +69,7 @@ def _definition_from_schema(
        filter=value.get("filter"),
        limit=value.get("limit"),
        inputs=value.get("inputs", []),
+        source_namespace=value.get("source_namespace", []),
    )


@@ -749,6 +749,10 @@ 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))
@@ -776,22 +780,6 @@ 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.
@@ -548,6 +548,7 @@ class RemoteTable(Table):
        LOOP.run(
            self._table.create_index(
                column,
+                replace=replace,
                config=config,
                wait_timeout=wait_timeout,
                name=name,
@@ -668,6 +668,167 @@ 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)]),
@@ -729,22 +890,6 @@ 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):
@@ -1017,6 +1162,8 @@ 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")

@@ -1064,6 +1211,22 @@ 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",
@@ -1130,6 +1293,49 @@ 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(
@@ -266,3 +266,38 @@ 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,11 +820,13 @@ 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"]

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

        table.wait_for_index(["custom_scalar_idx"], timedelta(seconds=2))
        table.wait_for_index(
@@ -1104,6 +1107,9 @@ 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"):
@@ -1113,15 +1119,17 @@ def test_remote_create_index_new_api():
                num_partitions=8,
            )

-        assert len(received_requests) == 5
+        assert len(received_requests) == 6
        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():
@@ -2534,26 +2542,6 @@ 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)
@@ -2582,9 +2570,6 @@ 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
@@ -629,6 +629,19 @@ 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,
@@ -666,30 +679,6 @@ 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.38.0"
+version = "0.39.0-beta.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,
-    PauseJobStatus, ReadConsistency, ResumeJobStatus, TableNamesRequest,
+    ReadConsistency, TableNamesRequest,
 };
 use crate::embeddings::{EmbeddingRegistry, MemoryRegistry};
 use crate::error::{Error, Result};
@@ -523,6 +523,28 @@ 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
@@ -590,18 +612,6 @@ 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,29 +235,6 @@ 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),
@@ -330,6 +307,10 @@ 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()
@@ -354,16 +335,6 @@ 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>> {
@@ -74,8 +74,15 @@ 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.
@@ -95,6 +102,10 @@ 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.
@@ -129,7 +140,12 @@ 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}"),
    })?;
-    value["kind"] = serde_json::Value::String(SELECT_KIND.to_string());
+    let kind = if definition.source_namespace.is_empty() {
+        SELECT_KIND
+    } else {
+        NAMESPACED_SELECT_KIND
+    };
+    value["kind"] = serde_json::Value::String(kind.to_string());
    Ok(value.to_string())
 }

@@ -150,12 +166,21 @@ pub fn materialized_view_kind(
        .get("kind")
        .and_then(|k| k.as_str())
        .ok_or_else(|| unreadable(&"missing kind tag"))?;
-    if kind != SELECT_KIND {
+    if kind != SELECT_KIND && kind != NAMESPACED_SELECT_KIND {
        return Ok(Some(MaterializedViewKind::Unrecognized {
            kind: kind.to_string(),
        }));
    }
-    let definition = serde_json::from_value(value).map_err(|e| unreadable(&e))?;
+    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
+        )));
+    }
    Ok(Some(MaterializedViewKind::Select(definition)))
 }

@@ -166,6 +191,7 @@ 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>,
@@ -319,6 +345,7 @@ 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 })
@@ -602,7 +629,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 name through the view's database.
+    /// resolves the recorded source coordinate through the view's database.
    database: Arc<dyn Database>,
 }

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

    /// Create the view table and verify it, consuming the declaration.
    ///
-    /// 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.
+    /// 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.
    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
@@ -640,6 +678,7 @@ 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
@@ -680,8 +719,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, so a handle that does not resolve back to itself is
-/// rejected, as is a namespaced source. Same creation-time checks as
+/// refresh will resolve -- name and namespace both -- so a handle that does
+/// not resolve back to itself is rejected. Same creation-time checks as
 /// [`Connection::create_materialized_view`].
 ///
 /// ```no_run
@@ -710,17 +749,9 @@ pub async fn prepare_declaration(
            message: "materialized views are supported only on local databases".into(),
        });
    };
-    // 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()
-            ),
-        });
-    }
+    // Refresh resolves the source at exactly this coordinate, so the
+    // definition records the namespace alongside the name.
+    let source_namespace = source.namespace().to_vec();
    let database = source
        .database_opt()
        .ok_or_else(|| Error::InvalidInput {
@@ -734,7 +765,7 @@ pub async fn prepare_declaration(
    let resolved = database
        .open_table(OpenTableRequest {
            name: source.name().to_string(),
-            namespace_path: vec![],
+            namespace_path: source_namespace.clone(),
            index_cache_size: None,
            lance_read_params: None,
            location: None,
@@ -780,6 +811,7 @@ pub async fn prepare_declaration(
    let (definition, mut fields, lineage) = plan(
        source_schema.clone(),
        resolved.name(),
+        &source_namespace,
        projections,
        filter,
        limit,
@@ -839,7 +871,9 @@ 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>,
@@ -850,13 +884,28 @@ 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(
@@ -887,7 +936,12 @@ 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).execute().await?;
+        let source = self
+            .connection
+            .open_table(&self.source)
+            .namespace(self.source_namespace.clone())
+            .execute()
+            .await?;
        let prepared = prepare_declaration(
            &source,
            &self.projections,
@@ -895,7 +949,7 @@ impl CreateMaterializedViewBuilder {
            self.limit,
        )
        .await?;
-        prepared.create(&self.name).await
+        prepared.create_in(&self.namespace, &self.name).await
    }
 }

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

-        // 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(),
-        )
+    /// 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()
+        })
        .await
        .unwrap();
-        let namespaced = Table::new(std::sync::Arc::new(namespaced), conn.database().clone());
-        let err = prepare_declaration(&namespaced, &[], None, None)
+
+        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()
            .await
-            .unwrap_err();
-        assert!(err.to_string().contains("namespaced source"), "{err}");
+            .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}"
+            );
+        }
    }
 }
@@ -170,6 +170,7 @@ 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,
@@ -590,7 +591,7 @@ async fn open_source(view: &Table, definition: &MaterializedViewDefinition) -> R
    let source = database
        .open_table(OpenTableRequest {
            name: definition.source_table.clone(),
-            namespace_path: Vec::new(),
+            namespace_path: definition.source_namespace.clone(),
            index_cache_size: None,
            lance_read_params: None,
            location: None,
@@ -2919,6 +2920,7 @@ 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(),
@@ -2958,6 +2960,7 @@ 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;
+use std::collections::{HashMap, HashSet};
 use std::sync::Arc;

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

@@ -535,6 +533,19 @@ 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,
@@ -590,6 +601,43 @@ 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
@@ -686,40 +734,6 @@ 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 {
@@ -2655,45 +2669,6 @@ 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(
@@ -2783,6 +2758,119 @@ 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,28 +73,6 @@ 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 {
@@ -527,6 +527,10 @@ impl<S: HttpSend> RemoteTable<S> {
            "column": canonical_column
        });

+        if !index.replace {
+            body["replace"] = false.into();
+        }
+
        // Add name parameter if provided (for backwards compatibility, only include if Some)
        if let Some(ref name) = index.name {
            body["name"] = serde_json::Value::String(name.clone());
@@ -6233,6 +6237,40 @@ mod tests {
        }
    }

+    #[tokio::test]
+    async fn test_create_index_forwards_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()))
+            .replace(false)
+            .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));
@@ -589,16 +589,13 @@ 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 {
+    if is_blob_v2 || field.r#type.fields.is_some() {
        let arrow_field = lance_namespace::schema::convert_json_arrow_field(field)
            .map_err(|e| invalid_function(format!("invalid Function input field: {e}")))?;
-        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()
-            )));
+        validate_function_blob_nesting(&arrow_field, false)?;
+        if is_blob_v2 {
+            return Ok(FUNCTION_BLOB_V2_TYPE.to_string());
        }
-        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())
@@ -617,6 +614,34 @@ 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)> {
@@ -697,21 +722,22 @@ fn parse_output_arrow_type(raw: &str) -> Result<JsonArrowDataType> {
 }

 fn function_output_field(name: &str, nullable: bool, raw: &str) -> Result<JsonArrowField> {
-    if raw == FUNCTION_BLOB_V2_TYPE {
-        return lance_namespace::schema::arrow_schema_to_json(&ArrowSchema::new(vec![
-            crate::blob(name, nullable),
-        ]))
+    let field = if raw == FUNCTION_BLOB_V2_TYPE {
+        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"));
-    }
-    Ok(JsonArrowField::new(
-        name.to_string(),
-        nullable,
-        parse_output_arrow_type(raw)?,
-    ))
+        .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)
 }

 fn function_output_field_matches(expected: &ArrowField, actual: &ArrowField) -> bool {
@@ -2719,6 +2745,28 @@ 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
@@ -3158,6 +3206,128 @@ 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);
Author	SHA1	Message	Date
Bruno Ramirez	9a1ffb9e02	fix(remote): forward create index replace flag (#4115 ) Remote create-index requests already expose `replace` on the builder, but the remote client did not consistently forward an explicit `replace=false` over REST. That meant create-only intent could be lost before it reached a remote server, even though local builders and Python APIs can express it. This PR forwards `replace=false` on the existing `create_index` endpoint and keeps the current default behavior unchanged for compatibility. This was accomplished with the following changes: - Serialize `replace: false` into the existing remote create-index request body when the builder is configured with `.replace(false)`. - Forward `replace` through the synchronous Python remote `create_index` wrapper so `RemoteTable.create_index(..., replace=False)` reaches the repaired path. - Continue omitting `replace` for the default path so existing remote create-index requests keep their current semantics. - Document `name` and `replace` on the existing OpenAPI create-index request schema. - Add coverage that verifies the remote client uses the existing `/create_index/` route and forwards `replace=false`, including the synchronous Python unified API. ### Testing - `cargo fmt --all --check` - `cargo test -p lancedb --features remote test_create_index_forwards_replace_false_on_existing_route --locked` - `uv tool run maturin develop --extras tests,dev,embeddings` - `uv run --frozen pytest python/tests/test_remote_db.py::test_remote_create_index_new_api` - `uv run ruff format --check python/lancedb/remote/table.py python/tests/test_remote_db.py` - `cargo build -p lancedb --features remote --locked` - `cargo clippy -p lancedb --features remote --all-targets --locked -- -D warnings`	2026-09-01 11:54:53 -07:00
LanceDB Robot	f2eb4a245d	chore: update lance dependency to v12.0.0-beta.9 (#4116 ) Updates Lance dependencies from v12.0.0-beta.5 to v12.0.0-beta.9 across Rust and Java. No compatibility fixes were required; full workspace Clippy passes with all features. Lance tag: https://github.com/lance-format/lance/releases/tag/v12.0.0-beta.9	2026-09-02 00:25:27 +08:00
Xuanwo	e6867f7d04	feat: support nested blob function signatures (#4109 ) Function signatures currently reject Blob v2 fields nested inside structs, preventing UDFs from accepting or returning structured values that contain blobs. Accept canonical Blob v2 fields as direct or recursive struct children while preserving exact field metadata and nullability. Blob fields under list, large-list, fixed-size-list, or map ancestors remain rejected because collection runtime adaptation is outside the supported Function ABI. A whole named struct result can bind directly to one destination column without introducing an extra wrapper level.	2026-09-01 23:51:08 +08:00
Xuanwo	193c5e3458	feat: add list_functions client APIs (#4108 ) Function registration and exact lookup are exposed through the SDK, but clients cannot discover published versions even though the server provides `POST /v1/functions/list`. Add Rust and Python sync/async `list_functions()` APIs that return typed `FunctionVersion` values. The remote client requests canonical definitions and follows opaque page tokens until the listing is complete, including empty intermediate pages, while preserving the server's name/version ordering. Local databases retain the existing Function-catalog unsupported error. The SDK consumes protocol pagination internally so callers receive the complete catalog rather than handling server-specific page tokens.	2026-09-01 23:50:57 +08:00
Lance Release	7ebd3c222d	Bump version: 0.38.0 → 0.39.0-beta.0	2026-09-01 13:16:03 +00:00
Wyatt Alt	d118ef168b	feat: record the source namespace in a materialized view definition (#4098 ) A view definition recorded its source by bare name and refresh resolved that name at the root, so declaring a view over a namespaced source was refused outright -- materialized views were root-only for every caller. The definition now carries `source_namespace`, and refresh opens the source at that coordinate. `plan` takes the namespace too: refresh re-plans the stored definition and persists the result when it migrates, so defaulting it there would strand the view on its next rebuild. The stored kind is the version boundary. Root definitions keep the `select` form byte-for-byte, so everything written before this change reads exactly as it always did. A namespaced source is stored as `namespaced_select`: released readers drop unknown fields and resolve a `select` source at the root, so keeping the old kind would let a rolled-back worker refresh a view from a same-name root table -- the new kind routes them to their existing unrecognized-kind refusal instead. The Python and Node definition parsers learn the new kind alongside the Rust core.	2026-09-01 06:05:02 -07:00