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
46 changed files with 2328 additions and 204 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)"]],
@@ -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",
@@ -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,6 +150,7 @@ 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]: ...
@@ -607,6 +608,7 @@ class FullTextQuery:
 class PyQueryRequest:
    limit: Optional[int]
    offset: Optional[int]
+    take_offsets: Optional[List[int]]
    filter: Optional[Union[str, bytes]]
    full_text_search: Optional[FullTextQuery]
    select: Optional[Union[str, List[str]]]
@@ -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.

@@ -1423,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))
@@ -2257,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)
@@ -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", []),
    )


@@ -109,6 +109,7 @@ def _query_is_plain_scan(query: Query) -> bool:
    return (
        query.vector is None
        and query.full_text_query is None
+        and query.take_offsets is None
        and not query.postfilter
        and not query.order_by
    )
@@ -804,6 +805,10 @@ class Query(pydantic.BaseModel):
    # offset to start fetching results from
    offset: Optional[int] = None

+    # Dataset offsets whose duplicate occurrences must be restored after lookup.
+    # This is populated when a take query is converted to this serializable form.
+    take_offsets: Optional[List[int]] = None
+
    # if true, will only search the indexed data
    fast_search: Optional[bool] = None

@@ -825,6 +830,7 @@ class Query(pydantic.BaseModel):
        query = cls()
        query.limit = req.limit
        query.offset = req.offset
+        query.take_offsets = req.take_offsets
        query.filter = req.filter
        query.full_text_query = req.full_text_search
        query.columns = req.select
@@ -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))
@@ -548,6 +548,7 @@ class RemoteTable(Table):
        LOOP.run(
            self._table.create_index(
                column,
+                replace=replace,
                config=config,
                wait_timeout=wait_timeout,
                name=name,
@@ -1678,9 +1678,9 @@ class Table(ABC):
        Offsets are mostly useful for sampling as the set of all valid offsets is easily
        known in advance to be [0, len(table)).

-        No guarantees are made regarding the order in which results are returned.  If
-        you desire an output order that matches the order of the given offsets, you will
-        need to add the row offset column to the output and align it yourself.
+        No guarantees are made regarding the order in which results are returned.
+        Repeated offsets produce repeated rows, which makes this method suitable for
+        sampling with replacement.

        Parameters
        ----------
@@ -4090,6 +4090,7 @@ class LanceTable(Table):
            )
            and not self._route_pushdown_to_rust
            and self.current_branch() is None
+            and query.take_offsets is None
        ):
            from lancedb.namespace import _execute_server_side_query

@@ -5983,7 +5984,23 @@ class AsyncTable:

    def _sync_query_to_async(
        self, query: Query
-    ) -> AsyncHybridQuery | AsyncFTSQuery | AsyncVectorQuery | AsyncQuery:
+    ) -> (
+        AsyncHybridQuery
+        | AsyncFTSQuery
+        | AsyncVectorQuery
+        | AsyncQuery
+        | AsyncTakeQuery
+    ):
+        if query.take_offsets is not None:
+            take_query = self.take_offsets(query.take_offsets)
+            if query.columns:
+                take_query = take_query.select(query.columns)
+            if query.use_lsm is not None:
+                take_query = take_query.use_lsm(query.use_lsm)
+            if query.with_row_id:
+                take_query = take_query.with_row_id()
+            return take_query
+
        async_query = self.query()
        if query.limit is not None:
            async_query = async_query.limit(query.limit)
@@ -6048,6 +6065,7 @@ class AsyncTable:
                self._namespace_client, self._pushdown_operations
            )
            and not self._route_pushdown_to_rust
+            and query.take_offsets is None
        ):
            from lancedb.namespace import _execute_server_side_query

@@ -6545,6 +6563,9 @@ class AsyncTable:
        Offsets are mostly useful for sampling as the set of all valid offsets is easily
        known in advance to be [0, len(table)).

+        No guarantees are made regarding the order in which results are returned.
+        Repeated offsets produce repeated rows.
+
        Parameters
        ----------
        offsets: list[int]
@@ -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"
+        )
@@ -1923,6 +1923,21 @@ def test_take_queries(tmp_path):
        17,
    ]

+    # Duplicate offsets are occurrences, not set members. Ordering is unspecified.
+    assert sorted(table.take_offsets([5, 2, 5, 17]).to_pandas()["idx"].to_list()) == [
+        2,
+        5,
+        5,
+        17,
+    ]
+
+    # Converting a take builder to its serializable query representation must
+    # retain occurrence metadata and execute with the same multiplicity.
+    query = table.take_offsets([5, 2, 5, 17]).select(["idx"]).to_query_object()
+    assert query.take_offsets == [5, 2, 5, 17]
+    converted = table._execute_query(query).read_all()
+    assert sorted(converted["idx"].to_pylist()) == [2, 5, 5, 17]
+
    # Take by row id
    assert list(
        sorted(table.take_row_ids([5, 2, 17]).to_pandas()["idx"].to_list())
@@ -479,24 +479,49 @@ def test_remote_permutation_is_picklable():
                match = re.search(
                    r"_rowoffset\s+in\s+\((.*?)\)", body["filter"], re.IGNORECASE
                )
-                offsets = [int(o.strip()) for o in match.group(1).split(",")]
+                offsets = list(
+                    dict.fromkeys(int(o.strip()) for o in match.group(1).split(","))
+                )
            else:
                offsets = list(range(len(rows)))
-            table = pa.table({"a": [rows[offset] for offset in offsets]})
+            columns = body.get("columns") or ["a"]
+            table = pa.table(
+                {
+                    column: (
+                        [rows[offset] for offset in offsets]
+                        if column == "a"
+                        else offsets
+                    )
+                    for column in columns
+                }
+            )

            request.send_response(200)
            request.send_header("Content-Type", "application/vnd.apache.arrow.file")
            request.end_headers()
            with pa.ipc.new_file(request.wfile, schema=table.schema) as writer:
-                writer.write_table(table)
+                writer.write_table(table, max_chunksize=2)
        else:
            request.send_response(404)
            request.end_headers()

    with mock_lancedb_connection(handler) as db:
-        permutation = Permutation.identity(db.open_table("test"))
+        table = db.open_table("test")
+        assert table.take_offsets([0, 2, 0, 4]).to_list() == [
+            {"a": 0},
+            {"a": 0},
+            {"a": 2},
+            {"a": 4},
+        ]
+
+        permutation = Permutation.identity(table)
        restored = pickle.loads(pickle.dumps(permutation))
-        assert restored.__getitems__([0, 2, 4]) == [{"a": 0}, {"a": 2}, {"a": 4}]
+        assert restored.__getitems__([0, 2, 0, 4]) == [
+            {"a": 0},
+            {"a": 2},
+            {"a": 0},
+            {"a": 4},
+        ]


 def test_create_table_exist_ok():
@@ -795,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"]

@@ -807,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(
@@ -1079,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"):
@@ -1088,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():
@@ -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,
@@ -323,6 +323,7 @@ impl<'py> IntoPyObject<'py> for PyQueryVectors {
 pub struct PyQueryRequest {
    pub limit: Option<usize>,
    pub offset: Option<usize>,
+    pub take_offsets: Option<Vec<u64>>,
    pub filter: Option<PyQueryFilter>,
    pub full_text_search: Option<PyLanceDB<FtsQuery>>,
    pub select: PySelect,
@@ -353,6 +354,7 @@ impl From<AnyQuery> for PyQueryRequest {
            AnyQuery::Query(query_request) => Self {
                limit: query_request.limit,
                offset: query_request.offset,
+                take_offsets: query_request.take_offsets,
                filter: query_request.filter.map(PyQueryFilter),
                full_text_search: query_request
                    .full_text_search
@@ -381,6 +383,7 @@ impl From<AnyQuery> for PyQueryRequest {
            AnyQuery::VectorQuery(vector_query) => Self {
                limit: vector_query.base.limit,
                offset: vector_query.base.offset,
+                take_offsets: vector_query.base.take_offsets,
                filter: vector_query.base.filter.map(PyQueryFilter),
                full_text_search: None,
                select_source_columns: PySelect::source_columns(&vector_query.base.select),
@@ -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
@@ -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
@@ -307,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()
@@ -31,7 +31,7 @@ use lance::io::RecordBatchStream;
 use lance_arrow::RecordBatchExt;
 use lance_core::ROW_ID;
 use lance_core::error::LanceOptionExt;
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::sync::Arc;

 /// Reads a permutation of a source table based on row IDs stored in a separate table
@@ -234,7 +234,14 @@ impl PermutationReader {
            .expect_ok()?
            .values();

-        let in_list: Vec<Expr> = row_ids.iter().map(|id| lit(*id)).collect();
+        let mut unique_row_ids = HashSet::with_capacity(num_rows);
+        let in_list: Vec<Expr> = row_ids
+            .iter()
+            .copied()
+            .filter(|row_id| unique_row_ids.insert(*row_id))
+            .map(lit)
+            .collect();
+        let num_unique_row_ids = unique_row_ids.len();

        let base_query = QueryRequest {
            filter: Some(QueryFilter::Datafusion(col(ROW_ID).in_list(in_list, false))),
@@ -247,7 +254,7 @@ impl PermutationReader {
            .query(
                &AnyQuery::Query(base_query),
                QueryExecutionOptions {
-                    max_batch_length: num_rows as u32,
+                    max_batch_length: num_unique_row_ids as u32,
                    ..Default::default()
                },
            )
@@ -262,9 +269,9 @@ impl PermutationReader {
            });
        }

-        if batches.iter().map(|b| b.num_rows()).sum::<usize>() != num_rows {
+        if batches.iter().map(|b| b.num_rows()).sum::<usize>() != num_unique_row_ids {
            return Err(Error::InvalidInput {
-                message: "Base table returned different number of rows than the number of row IDs"
+                message: "Base table returned a different number of rows than the number of unique row IDs"
                    .to_string(),
            });
        }
@@ -504,6 +511,7 @@ impl PermutationReader {
            let table = Table::from(self.base_table.clone());
            let batches = table
                .take_offsets(offsets.to_vec())
+                .preserve_order()
                .select(selection.clone())
                .execute()
                .await?
@@ -803,10 +811,10 @@ mod tests {
        .unwrap();

        // Take offsets in reverse order and verify returned rows match that order
-        let offsets = vec![5, 3, 1, 0];
+        let offsets = vec![5, 3, 5, 1, 0];
        let batch = reader.take_offsets(&offsets, Select::All).await.unwrap();

-        assert_eq!(batch.num_rows(), 4);
+        assert_eq!(batch.num_rows(), 5);

        let idx_values = batch
            .column(0)
@@ -820,6 +828,52 @@ mod tests {
        assert_eq!(idx_values, expected);
    }

+    #[tokio::test]
+    async fn test_take_offsets_preserves_repeated_rows_in_permutation() {
+        let base_table = lance_datagen::gen_batch()
+            .col("idx", lance_datagen::array::step::<Int32Type>())
+            .into_mem_table("tbl", RowCount::from(5), BatchCount::from(1))
+            .await;
+        let base_row_ids = collect_column::<UInt64Type>(&base_table, "_rowid").await;
+        let permutation_row_ids = vec![
+            base_row_ids[3],
+            base_row_ids[1],
+            base_row_ids[3],
+            base_row_ids[2],
+        ];
+        let permutation_batch = RecordBatch::try_new(
+            Arc::new(Schema::new(vec![
+                Field::new("row_id", DataType::UInt64, false),
+                Field::new(SPLIT_ID_COLUMN, DataType::UInt64, false),
+            ])),
+            vec![
+                Arc::new(UInt64Array::from(permutation_row_ids)),
+                Arc::new(UInt64Array::from(vec![0; 4])),
+            ],
+        )
+        .unwrap();
+        let permutation_table = virtual_table("row_ids", &permutation_batch).await;
+        let reader = PermutationReader::try_from_tables(
+            base_table.base_table().clone(),
+            permutation_table.base_table().clone(),
+            0,
+        )
+        .await
+        .unwrap();
+
+        let batch = reader
+            .take_offsets(&[0, 1, 2, 3], Select::All)
+            .await
+            .unwrap();
+        let idx_values = batch
+            .column(0)
+            .as_primitive::<Int32Type>()
+            .values()
+            .to_vec();
+
+        assert_eq!(idx_values, vec![3, 1, 3, 2]);
+    }
+
    #[tokio::test]
    async fn test_take_offsets_with_column_selection() {
        let (base_table, row_ids_table, row_ids) = setup_permutation_tables(10).await;
@@ -883,17 +937,17 @@ mod tests {
            .unwrap();

        // With no permutation table, take_offsets uses the base table directly
-        let offsets = vec![0, 2, 4, 6];
+        let offsets = vec![0, 2, 0, 4, 6];
        let batch = reader.take_offsets(&offsets, Select::All).await.unwrap();

-        assert_eq!(batch.num_rows(), 4);
+        assert_eq!(batch.num_rows(), 5);

        let idx_values = batch
            .column(0)
            .as_primitive::<Int32Type>()
            .values()
            .to_vec();
-        assert_eq!(idx_values, vec![0, 2, 4, 6]);
+        assert_eq!(idx_values, vec![0, 2, 0, 4, 6]);
    }

    #[tokio::test]
@@ -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,21 +1,37 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright The LanceDB Authors

+use std::collections::{HashMap, HashSet};
+use std::pin::Pin;
 use std::sync::Arc;
 use std::{future::Future, time::Duration};

 use arrow::compute::concat_batches;
-use arrow_array::{Array, Float16Array, Float32Array, Float64Array, RecordBatch, make_array};
+use arrow_array::{
+    Array, Float16Array, Float32Array, Float64Array, RecordBatch, UInt64Array,
+    cast::AsArray,
+    make_array,
+    types::{Int64Type, UInt64Type},
+};
 use arrow_schema::{DataType, SchemaRef};
+use datafusion_common::{DataFusionError, Result as DataFusionResult};
+use datafusion_execution::TaskContext;
 use datafusion_expr::{Expr, col, lit};
-use datafusion_physical_plan::ExecutionPlan;
-use futures::{FutureExt, TryFutureExt, TryStreamExt, stream, try_join};
+use datafusion_physical_expr::{EquivalenceProperties, Partitioning};
+use datafusion_physical_plan::{
+    DisplayAs, DisplayFormatType, ExecutionPlan, ExecutionPlanProperties, PlanProperties,
+    coalesce_partitions::CoalescePartitionsExec,
+    execution_plan::{Boundedness, EmissionType},
+    limit::GlobalLimitExec,
+    stream::RecordBatchStreamAdapter,
+};
+use futures::{FutureExt, StreamExt, TryFutureExt, TryStreamExt, stream, try_join};
 use half::f16;
 /// Re-export Lance ColumnOrdering type for use in query ordering
 pub use lance::dataset::scanner::ColumnOrdering;
 use lance::dataset::{ROW_ID, scanner::DatasetRecordBatchStream};
 use lance_arrow::RecordBatchExt;
-use lance_datafusion::exec::execute_plan;
+use lance_datafusion::exec::{execute_plan, format_plan as format_analyzed_plan};
 use lance_index::scalar::FullTextSearchQuery;
 use lance_index::scalar::inverted::SCORE_COL;
 use lance_index::vector::DIST_COL;
@@ -825,6 +841,14 @@ pub struct QueryRequest {
    /// Offset of the query.
    pub offset: Option<usize>,

+    /// Dataset offsets whose occurrence multiplicity must be restored after
+    /// executing the physical lookup represented by this request.
+    ///
+    /// This is client-side execution metadata used when a [`TakeQuery`] is
+    /// converted into a request. It is not sent to remote services.
+    #[doc(hidden)]
+    pub take_offsets: Option<Vec<u64>>,
+
    /// Apply filter to the returned rows.
    pub filter: Option<QueryFilter>,

@@ -893,6 +917,7 @@ impl Default for QueryRequest {
        Self {
            limit: None,
            offset: None,
+            take_offsets: None,
            filter: None,
            filter_error: None,
            full_text_search: None,
@@ -1529,6 +1554,302 @@ impl HasQuery for VectorQuery {
    }
 }

+fn take_occurrences(offsets: &[u64]) -> HashMap<u64, usize> {
+    let mut occurrences = HashMap::with_capacity(offsets.len());
+    for offset in offsets {
+        *occurrences.entry(*offset).or_insert(0) += 1;
+    }
+    occurrences
+}
+
+fn restore_take_batch_with_occurrences(
+    batch: RecordBatch,
+    offsets: &[u64],
+    occurrences: &HashMap<u64, usize>,
+    ordering_column: &str,
+    drop_ordering_column: bool,
+    preserve_order: bool,
+) -> Result<RecordBatch> {
+    let actual_offsets = batch
+        .column_by_name(ordering_column)
+        .ok_or_else(|| Error::Schema {
+            message: format!(
+                "take query result did not include ordering column '{ordering_column}'"
+            ),
+        })?;
+    let actual_offsets = match actual_offsets.data_type() {
+        DataType::UInt64 => actual_offsets
+            .as_primitive::<UInt64Type>()
+            .values()
+            .to_vec(),
+        DataType::Int64 => actual_offsets
+            .as_primitive::<Int64Type>()
+            .values()
+            .iter()
+            .map(|offset| {
+                u64::try_from(*offset).map_err(|_| Error::Schema {
+                    message: format!(
+                        "take query ordering column '{ordering_column}' contained a negative offset"
+                    ),
+                })
+            })
+            .collect::<Result<Vec<_>>>()?,
+        data_type => {
+            return Err(Error::Schema {
+                message: format!(
+                    "take query ordering column '{ordering_column}' had unsupported type {data_type}"
+                ),
+            });
+        }
+    };
+
+    let mut desired_order = Vec::with_capacity(offsets.len());
+    if preserve_order {
+        let ordering = actual_offsets
+            .iter()
+            .copied()
+            .enumerate()
+            .map(|(index, offset)| (offset, index as u64))
+            .collect::<HashMap<_, _>>();
+        // Missing offsets retain the filter-based behavior of returning no row.
+        desired_order.extend(
+            offsets
+                .iter()
+                .filter_map(|offset| ordering.get(offset).copied()),
+        );
+    } else {
+        // Public take queries do not guarantee output order. Preserve the lookup's
+        // existing order and only restore the multiplicity of each matching row.
+        for (index, offset) in actual_offsets.iter().enumerate() {
+            if let Some(count) = occurrences.get(offset) {
+                desired_order.extend(std::iter::repeat_n(index as u64, *count));
+            }
+        }
+    }
+
+    let mut ordered_batch = if desired_order.len() == batch.num_rows()
+        && desired_order
+            .iter()
+            .enumerate()
+            .all(|(index, desired)| *desired == index as u64)
+    {
+        batch
+    } else {
+        arrow_select::take::take_record_batch(&batch, &UInt64Array::from(desired_order))?
+    };
+
+    if drop_ordering_column {
+        ordered_batch = ordered_batch.drop_column(ordering_column)?;
+    }
+
+    Ok(ordered_batch)
+}
+
+#[cfg(test)]
+fn restore_take_batch(
+    batch: RecordBatch,
+    offsets: &[u64],
+    ordering_column: &str,
+    drop_ordering_column: bool,
+    preserve_order: bool,
+) -> Result<RecordBatch> {
+    restore_take_batch_with_occurrences(
+        batch,
+        offsets,
+        &take_occurrences(offsets),
+        ordering_column,
+        drop_ordering_column,
+        preserve_order,
+    )
+}
+
+/// Restores the logical offset occurrence sequence above the physical lookup plan.
+///
+/// The lookup plan returns each matching row at most once. For ordinary unordered
+/// takes this operator expands each input batch incrementally and preserves the
+/// lookup's partitioning. The explicitly ordered reader path collects one coalesced
+/// input before restoring requested order. Pagination must remain above this operator
+/// so it applies to occurrences.
+#[derive(Debug)]
+struct TakeRestoreExec {
+    input: Arc<dyn ExecutionPlan>,
+    offsets: Vec<u64>,
+    occurrences: Arc<HashMap<u64, usize>>,
+    ordering_column: String,
+    drop_ordering_column: bool,
+    preserve_order: bool,
+    schema: SchemaRef,
+    properties: Arc<PlanProperties>,
+}
+
+impl TakeRestoreExec {
+    fn try_new(
+        input: Arc<dyn ExecutionPlan>,
+        offsets: Vec<u64>,
+        ordering_column: String,
+        drop_ordering_column: bool,
+        preserve_order: bool,
+    ) -> Result<Self> {
+        let schema = if drop_ordering_column {
+            RecordBatch::new_empty(input.schema())
+                .drop_column(&ordering_column)?
+                .schema()
+        } else {
+            input.schema()
+        };
+        let partition_count = if preserve_order {
+            1
+        } else {
+            input.output_partitioning().partition_count()
+        };
+        let emission_type = if preserve_order {
+            EmissionType::Final
+        } else {
+            EmissionType::Incremental
+        };
+        let properties = Arc::new(PlanProperties::new(
+            EquivalenceProperties::new(schema.clone()),
+            Partitioning::UnknownPartitioning(partition_count),
+            emission_type,
+            Boundedness::Bounded,
+        ));
+
+        Ok(Self {
+            input,
+            occurrences: Arc::new(take_occurrences(&offsets)),
+            offsets,
+            ordering_column,
+            drop_ordering_column,
+            preserve_order,
+            schema,
+            properties,
+        })
+    }
+}
+
+impl DisplayAs for TakeRestoreExec {
+    fn fmt_as(
+        &self,
+        _display_type: DisplayFormatType,
+        formatter: &mut std::fmt::Formatter<'_>,
+    ) -> std::fmt::Result {
+        write!(
+            formatter,
+            "TakeRestoreExec: occurrences={}",
+            self.offsets.len()
+        )
+    }
+}
+
+impl ExecutionPlan for TakeRestoreExec {
+    fn name(&self) -> &str {
+        "TakeRestoreExec"
+    }
+
+    fn properties(&self) -> &Arc<PlanProperties> {
+        &self.properties
+    }
+
+    fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
+        vec![&self.input]
+    }
+
+    fn maintains_input_order(&self) -> Vec<bool> {
+        vec![!self.preserve_order]
+    }
+
+    fn benefits_from_input_partitioning(&self) -> Vec<bool> {
+        vec![false]
+    }
+
+    fn with_new_children(
+        self: Arc<Self>,
+        children: Vec<Arc<dyn ExecutionPlan>>,
+    ) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
+        if children.len() != 1 {
+            return Err(DataFusionError::Internal(format!(
+                "TakeRestoreExec expected one child, got {}",
+                children.len()
+            )));
+        }
+        let child = children.into_iter().next().unwrap();
+        let plan = Self::try_new(
+            child,
+            self.offsets.clone(),
+            self.ordering_column.clone(),
+            self.drop_ordering_column,
+            self.preserve_order,
+        )
+        .map_err(|error| DataFusionError::External(Box::new(error)))?;
+        Ok(Arc::new(plan))
+    }
+
+    fn execute(
+        &self,
+        partition: usize,
+        context: Arc<TaskContext>,
+    ) -> DataFusionResult<datafusion_physical_plan::SendableRecordBatchStream> {
+        let partition_count = self.input.output_partitioning().partition_count();
+        if partition >= partition_count || (self.preserve_order && partition != 0) {
+            return Err(DataFusionError::Internal(format!(
+                "TakeRestoreExec cannot execute partition {partition}; input has {partition_count} partitions"
+            )));
+        }
+
+        let input = self.input.execute(partition, context)?;
+        let output_schema = self.schema.clone();
+        let offsets = self.offsets.clone();
+        let occurrences = self.occurrences.clone();
+        let ordering_column = self.ordering_column.clone();
+        let drop_ordering_column = self.drop_ordering_column;
+        let preserve_order = self.preserve_order;
+        let stream: Pin<Box<dyn futures::Stream<Item = DataFusionResult<RecordBatch>> + Send>> =
+            if preserve_order {
+                let input_schema = input.schema();
+                Box::pin(stream::once(async move {
+                    let batches = input.try_collect::<Vec<_>>().await?;
+                    let batch = if batches.is_empty() {
+                        RecordBatch::new_empty(input_schema.clone())
+                    } else {
+                        concat_batches(&input_schema, &batches)?
+                    };
+                    restore_take_batch_with_occurrences(
+                        batch,
+                        &offsets,
+                        &occurrences,
+                        &ordering_column,
+                        drop_ordering_column,
+                        true,
+                    )
+                    .map_err(|error| DataFusionError::External(Box::new(error)))
+                }))
+            } else {
+                Box::pin(input.map(move |batch| {
+                    batch.and_then(|batch| {
+                        restore_take_batch_with_occurrences(
+                            batch,
+                            &offsets,
+                            &occurrences,
+                            &ordering_column,
+                            drop_ordering_column,
+                            false,
+                        )
+                        .map_err(|error| DataFusionError::External(Box::new(error)))
+                    })
+                }))
+            };
+
+        Ok(Box::pin(RecordBatchStreamAdapter::new(
+            output_schema,
+            stream,
+        )))
+    }
+
+    fn supports_limit_pushdown(&self) -> bool {
+        false
+    }
+}
+
 /// A builder for LanceDB take queries.
 ///
 /// See [`crate::Table::query`] for more details on queries
@@ -1545,6 +1866,8 @@ impl HasQuery for VectorQuery {
 pub struct TakeQuery {
    parent: Arc<dyn BaseTable>,
    request: QueryRequest,
+    offsets: Option<Vec<u64>>,
+    preserve_order: bool,
 }

 impl TakeQuery {
@@ -1552,15 +1875,24 @@ impl TakeQuery {
    ///
    /// See [`crate::Table::take_offsets`] for more details.
    pub fn from_offsets(parent: Arc<dyn BaseTable>, offsets: Vec<u64>) -> Self {
-        let in_list: Vec<Expr> = offsets.iter().map(|o| lit(*o)).collect();
+        let mut seen = HashSet::with_capacity(offsets.len());
+        let in_list: Vec<Expr> = offsets
+            .iter()
+            .copied()
+            .filter(|offset| seen.insert(*offset))
+            .map(lit)
+            .collect();
        Self {
            parent,
            request: QueryRequest {
                filter: Some(QueryFilter::Datafusion(
                    col("_rowoffset").in_list(in_list, false),
                )),
+                take_offsets: Some(offsets.clone()),
                ..Default::default()
            },
+            offsets: Some(offsets),
+            preserve_order: false,
        }
    }

@@ -1575,9 +1907,181 @@ impl TakeQuery {
                filter: Some(QueryFilter::Datafusion(col(ROW_ID).in_list(in_list, false))),
                ..Default::default()
            },
+            offsets: None,
+            preserve_order: false,
        }
    }

+    /// Preserve the requested offset order when restoring duplicate occurrences.
+    ///
+    /// This is reserved for readers whose API explicitly guarantees ordering.
+    pub(crate) fn preserve_order(mut self) -> Self {
+        debug_assert!(self.offsets.is_some());
+        self.preserve_order = true;
+        self
+    }
+
+    async fn request_with_row_offset(
+        parent: &dyn BaseTable,
+        request: &QueryRequest,
+    ) -> Result<(QueryRequest, String, bool)> {
+        const ROW_OFFSET: &str = "_rowoffset";
+        const INTERNAL_ROW_OFFSET: &str = "__lancedb_take_row_offset";
+
+        let mut request = request.clone();
+        // The physical lookup must not recursively restore occurrences. The
+        // wrapper above this request owns that logical operation.
+        request.take_offsets = None;
+        let (ordering_column, drop_ordering_column) = match &mut request.select {
+            Select::All => {
+                let mut columns = parent
+                    .schema()
+                    .await?
+                    .fields()
+                    .iter()
+                    .map(|field| field.name().clone())
+                    .collect::<Vec<_>>();
+                columns.push(ROW_OFFSET.to_string());
+                request.select = Select::Columns(columns);
+                (ROW_OFFSET.to_string(), true)
+            }
+            Select::Columns(columns) => {
+                if columns.iter().any(|column| column == ROW_OFFSET) {
+                    (ROW_OFFSET.to_string(), false)
+                } else {
+                    columns.push(ROW_OFFSET.to_string());
+                    (ROW_OFFSET.to_string(), true)
+                }
+            }
+            Select::Dynamic(columns) => {
+                let mut ordering_column = INTERNAL_ROW_OFFSET.to_string();
+                while columns.iter().any(|(name, _)| name == &ordering_column) {
+                    ordering_column.push('_');
+                }
+                columns.push((ordering_column.clone(), ROW_OFFSET.to_string()));
+                (ordering_column, true)
+            }
+            Select::Expr(columns) => {
+                let mut ordering_column = INTERNAL_ROW_OFFSET.to_string();
+                while columns.iter().any(|(name, _)| name == &ordering_column) {
+                    ordering_column.push('_');
+                }
+                columns.push((ordering_column.clone(), col(ROW_OFFSET)));
+                (ordering_column, true)
+            }
+        };
+
+        Ok((request, ordering_column, drop_ordering_column))
+    }
+
+    async fn prepare_offsets_lookup(
+        parent: &dyn BaseTable,
+        request: &QueryRequest,
+    ) -> Result<(QueryRequest, String, bool, usize, Option<usize>)> {
+        let (mut request, ordering_column, drop_ordering_column) =
+            Self::request_with_row_offset(parent, request).await?;
+        // The lookup operates on distinct physical rows. Pagination is a logical
+        // operation over occurrences and must be applied only after restoration.
+        let output_offset = request.offset.take().unwrap_or_default();
+        let output_limit = request.limit.take();
+
+        Ok((
+            request,
+            ordering_column,
+            drop_ordering_column,
+            output_offset,
+            output_limit,
+        ))
+    }
+
+    fn wrap_offsets_plan(
+        lookup: Arc<dyn ExecutionPlan>,
+        offsets: &[u64],
+        ordering_column: String,
+        drop_ordering_column: bool,
+        output_offset: usize,
+        output_limit: Option<usize>,
+        preserve_order: bool,
+    ) -> Result<Arc<dyn ExecutionPlan>> {
+        let lookup = if preserve_order {
+            Arc::new(CoalescePartitionsExec::new(lookup)) as Arc<dyn ExecutionPlan>
+        } else {
+            lookup
+        };
+        let restored: Arc<dyn ExecutionPlan> = Arc::new(TakeRestoreExec::try_new(
+            lookup,
+            offsets.to_vec(),
+            ordering_column,
+            drop_ordering_column,
+            preserve_order,
+        )?);
+
+        if output_offset > 0 || output_limit.is_some() {
+            Ok(Arc::new(GlobalLimitExec::new(
+                restored,
+                output_offset,
+                output_limit,
+            )))
+        } else {
+            Ok(restored)
+        }
+    }
+
+    fn wrap_offsets_explanation(
+        lookup: &str,
+        occurrence_count: usize,
+        output_offset: usize,
+        output_limit: Option<usize>,
+        preserve_order: bool,
+    ) -> String {
+        fn indent(plan: &str, spaces: usize) -> String {
+            let indentation = " ".repeat(spaces);
+            plan.lines()
+                .map(|line| format!("{indentation}{line}"))
+                .collect::<Vec<_>>()
+                .join("\n")
+        }
+
+        let restored = if preserve_order {
+            format!(
+                "TakeRestoreExec: occurrences={occurrence_count}\n  CoalescePartitionsExec\n{}",
+                indent(lookup, 4)
+            )
+        } else {
+            format!(
+                "TakeRestoreExec: occurrences={occurrence_count}\n{}",
+                indent(lookup, 2)
+            )
+        };
+
+        if output_offset > 0 || output_limit.is_some() {
+            let fetch = output_limit
+                .map(|limit| limit.to_string())
+                .unwrap_or_else(|| "None".to_string());
+            format!(
+                "GlobalLimitExec: skip={output_offset}, fetch={fetch}\n{}",
+                indent(&restored, 2)
+            )
+        } else {
+            restored
+        }
+    }
+
+    async fn create_offsets_plan(
+        &self,
+        offsets: &[u64],
+        options: QueryExecutionOptions,
+    ) -> Result<Arc<dyn ExecutionPlan>> {
+        create_take_offsets_plan(
+            self.parent.as_ref(),
+            &self.request,
+            offsets,
+            options,
+            self.preserve_order,
+        )
+        .await
+    }
+
    /// Convert the `TakeQuery` into a `QueryRequest`.
    pub fn into_request(self) -> QueryRequest {
        self.request
@@ -1622,6 +2126,63 @@ impl TakeQuery {
    }
 }

+pub(crate) async fn create_take_offsets_plan(
+    parent: &dyn BaseTable,
+    request: &QueryRequest,
+    offsets: &[u64],
+    options: QueryExecutionOptions,
+    preserve_order: bool,
+) -> Result<Arc<dyn ExecutionPlan>> {
+    let (request, ordering_column, drop_ordering_column, output_offset, output_limit) =
+        TakeQuery::prepare_offsets_lookup(parent, request).await?;
+    let lookup_options = if preserve_order {
+        options.without_output_batch_length_limit()
+    } else {
+        options
+    };
+    let lookup = parent
+        .create_plan(&AnyQuery::Query(request), lookup_options)
+        .await?;
+
+    TakeQuery::wrap_offsets_plan(
+        lookup,
+        offsets,
+        ordering_column,
+        drop_ordering_column,
+        output_offset,
+        output_limit,
+        preserve_order,
+    )
+}
+
+pub(crate) async fn explain_take_offsets_plan(
+    parent: &dyn BaseTable,
+    request: &QueryRequest,
+    offsets: &[u64],
+    verbose: bool,
+) -> Result<String> {
+    let (request, _, _, output_offset, output_limit) =
+        TakeQuery::prepare_offsets_lookup(parent, request).await?;
+    let lookup = parent
+        .explain_plan(&AnyQuery::Query(request), verbose)
+        .await?;
+    Ok(TakeQuery::wrap_offsets_explanation(
+        &lookup,
+        offsets.len(),
+        output_offset,
+        output_limit,
+        false,
+    ))
+}
+
+pub(crate) async fn prepare_take_offsets_request(
+    parent: &dyn BaseTable,
+    request: &QueryRequest,
+) -> Result<QueryRequest> {
+    let (request, _, _, _, _) = TakeQuery::prepare_offsets_lookup(parent, request).await?;
+    Ok(request)
+}
+
 impl HasQuery for TakeQuery {
    fn mut_query(&mut self) -> &mut QueryRequest {
        &mut self.request
@@ -1630,6 +2191,10 @@ impl HasQuery for TakeQuery {

 impl ExecutableQuery for TakeQuery {
    async fn create_plan(&self, options: QueryExecutionOptions) -> Result<Arc<dyn ExecutionPlan>> {
+        if let Some(offsets) = &self.offsets {
+            return self.create_offsets_plan(offsets, options).await;
+        }
+
        let req = AnyQuery::Query(self.request.clone());
        self.parent.clone().create_plan(&req, options).await
    }
@@ -1638,6 +2203,18 @@ impl ExecutableQuery for TakeQuery {
        &self,
        options: QueryExecutionOptions,
    ) -> Result<SendableRecordBatchStream> {
+        if self.offsets.is_some() {
+            let plan = self.create_plan(options.clone()).await?;
+            let inner = execute_plan(plan, Default::default())?;
+            let inner = MaxBatchLengthStream::new_boxed(inner, options.max_batch_length as usize);
+            let inner = if let Some(timeout) = options.timeout {
+                TimeoutStream::new_boxed(inner, timeout)
+            } else {
+                inner
+            };
+            return Ok(DatasetRecordBatchStream::new(inner).into());
+        }
+
        let query = AnyQuery::Query(self.request.clone());
        Ok(SendableRecordBatchStream::from(
            self.parent.clone().query(&query, options).await?,
@@ -1645,11 +2222,51 @@ impl ExecutableQuery for TakeQuery {
    }

    async fn explain_plan(&self, verbose: bool) -> Result<String> {
+        if let Some(offsets) = &self.offsets {
+            let (request, _, _, output_offset, output_limit) =
+                Self::prepare_offsets_lookup(self.parent.as_ref(), &self.request).await?;
+            // Ask the backend to explain only the distinct-row lookup. This keeps
+            // remote explanation non-executing while still showing the client-side
+            // operators that create_plan and execution place above that lookup.
+            let lookup = self
+                .parent
+                .explain_plan(&AnyQuery::Query(request), verbose)
+                .await?;
+            return Ok(Self::wrap_offsets_explanation(
+                &lookup,
+                offsets.len(),
+                output_offset,
+                output_limit,
+                self.preserve_order,
+            ));
+        }
+
        let query = AnyQuery::Query(self.request.clone());
        self.parent.explain_plan(&query, verbose).await
    }

    async fn analyze_plan_with_options(&self, options: QueryExecutionOptions) -> Result<String> {
+        if self.offsets.is_some() {
+            if self.parent.analyze_plan_is_remote() {
+                let (request, _, _, _, _) =
+                    Self::prepare_offsets_lookup(self.parent.as_ref(), &self.request).await?;
+                // Remote analysis is owned by the service. The current wire
+                // request represents only the distinct-row lookup, so return
+                // the service report unchanged instead of fabricating metrics
+                // for client-side restoration operators.
+                return self
+                    .parent
+                    .analyze_plan(&AnyQuery::Query(request), options)
+                    .await;
+            }
+
+            let plan = self.create_plan(options).await?;
+            execute_plan(plan.clone(), Default::default())?
+                .try_collect::<Vec<_>>()
+                .await?;
+            return Ok(format_analyzed_plan(plan));
+        }
+
        let query = AnyQuery::Query(self.request.clone());
        self.parent.analyze_plan(&query, options).await
    }
@@ -1670,6 +2287,7 @@ mod tests {
        StringArray, cast::AsArray, types::Float32Type,
    };
    use arrow_schema::{DataType, Field as ArrowField, Schema as ArrowSchema};
+    use datafusion_physical_plan::display::DisplayableExecutionPlan;
    use futures::{StreamExt, TryStreamExt};
    use lance_testing::datagen::{BatchGenerator, IncrementingInt32, RandomVector};
    use rand::seq::IndexedRandom;
@@ -2924,6 +3542,218 @@ mod tests {
        assert_eq!(results[0].num_columns(), 1);
    }

+    #[tokio::test]
+    async fn test_take_offsets_preserves_duplicate_multiplicity() {
+        let tmp_dir = tempdir().unwrap();
+        let table = make_test_table(&tmp_dir).await;
+
+        let results = table
+            .take_offsets(vec![5, 1, 5, 17])
+            .select(Select::Columns(vec!["id".to_string()]))
+            .execute_with_options(QueryExecutionOptions {
+                max_batch_length: 2,
+                ..Default::default()
+            })
+            .await
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+
+        assert_eq!(results.len(), 2);
+        assert!(results.iter().all(|batch| batch.num_columns() == 1));
+        let mut ids = results
+            .iter()
+            .flat_map(|batch| {
+                batch
+                    .column_by_name("id")
+                    .unwrap()
+                    .as_primitive::<Int32Type>()
+                    .values()
+                    .to_vec()
+            })
+            .collect::<Vec<_>>();
+        ids.sort_unstable();
+        assert_eq!(ids, vec![1, 5, 5, 17]);
+    }
+
+    #[tokio::test]
+    async fn test_take_offsets_plan_is_incremental() {
+        let tmp_dir = tempdir().unwrap();
+        let table = make_test_table(&tmp_dir).await;
+
+        let plan = table
+            .take_offsets(vec![5, 1, 17])
+            .create_plan(QueryExecutionOptions {
+                max_batch_length: 1,
+                ..Default::default()
+            })
+            .await
+            .unwrap();
+
+        assert_eq!(plan.properties().emission_type, EmissionType::Incremental);
+        let displayed = DisplayableExecutionPlan::new(plan.as_ref())
+            .indent(false)
+            .to_string();
+        assert!(displayed.contains("TakeRestoreExec"));
+        assert!(!displayed.contains("CoalescePartitionsExec"));
+    }
+
+    #[tokio::test]
+    async fn test_take_into_request_preserves_duplicate_multiplicity() {
+        let tmp_dir = tempdir().unwrap();
+        let table = make_test_table(&tmp_dir).await;
+        let request = table.take_offsets(vec![5, 5]).into_request();
+        assert_eq!(request.take_offsets, Some(vec![5, 5]));
+
+        let batches = table
+            .base_table()
+            .query(&AnyQuery::Query(request), QueryExecutionOptions::default())
+            .await
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+
+        assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 2);
+    }
+
+    #[test]
+    fn test_restore_take_batch_only_reorders_when_requested() {
+        let batch = RecordBatch::try_from_iter([
+            (
+                "id",
+                Arc::new(Int32Array::from(vec![17, 5, 1])) as Arc<dyn Array>,
+            ),
+            (
+                "_rowoffset",
+                Arc::new(UInt64Array::from(vec![17, 5, 1])) as Arc<dyn Array>,
+            ),
+        ])
+        .unwrap();
+
+        let restored =
+            restore_take_batch(batch.clone(), &[5, 1, 5, 17], "_rowoffset", true, false).unwrap();
+        assert_eq!(
+            restored
+                .column_by_name("id")
+                .unwrap()
+                .as_primitive::<Int32Type>()
+                .values(),
+            &[17, 5, 5, 1]
+        );
+
+        let ordered = restore_take_batch(batch, &[5, 1, 5, 17], "_rowoffset", true, true).unwrap();
+        assert_eq!(
+            ordered
+                .column_by_name("id")
+                .unwrap()
+                .as_primitive::<Int32Type>()
+                .values(),
+            &[5, 1, 5, 17]
+        );
+    }
+
+    #[tokio::test]
+    async fn test_take_offsets_applies_pagination_after_restoration() {
+        let tmp_dir = tempdir().unwrap();
+        let table = make_test_table(&tmp_dir).await;
+
+        let limited = table
+            .take_offsets(vec![0, 1, 0, 2])
+            .select(Select::Columns(vec!["id".to_string()]))
+            .limit(3)
+            .execute()
+            .await
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+        let limited = concat_batches(&limited[0].schema(), &limited).unwrap();
+        assert_eq!(limited.num_rows(), 3);
+        assert!(
+            limited
+                .column_by_name("id")
+                .unwrap()
+                .as_primitive::<Int32Type>()
+                .values()
+                .iter()
+                .all(|id| [0, 1, 2].contains(id))
+        );
+
+        let offset = table
+            .take_offsets(vec![5, 1, 5, 17])
+            .select(Select::Columns(vec!["id".to_string()]))
+            .offset(1)
+            .execute()
+            .await
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+        let offset = concat_batches(&offset[0].schema(), &offset).unwrap();
+        assert_eq!(offset.num_rows(), 3);
+        assert!(
+            offset
+                .column_by_name("id")
+                .unwrap()
+                .as_primitive::<Int32Type>()
+                .values()
+                .iter()
+                .all(|id| [1, 5, 17].contains(id))
+        );
+    }
+
+    #[tokio::test]
+    async fn test_take_offsets_create_plan_restores_occurrences() {
+        let tmp_dir = tempdir().unwrap();
+        let table = make_test_table(&tmp_dir).await;
+        let take = table
+            .take_offsets(vec![5, 1, 5, 17])
+            .select(Select::Columns(vec!["id".to_string()]));
+
+        let plan = take
+            .create_plan(QueryExecutionOptions::default())
+            .await
+            .unwrap();
+        assert_eq!(plan.schema().fields().len(), 1);
+        assert_eq!(plan.schema().field(0).name(), "id");
+        let planned = execute_plan(plan, Default::default())
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+        let planned = concat_batches(&planned[0].schema(), &planned).unwrap();
+        let mut ids = planned
+            .column_by_name("id")
+            .unwrap()
+            .as_primitive::<Int32Type>()
+            .values()
+            .to_vec();
+        ids.sort_unstable();
+        assert_eq!(ids, vec![1, 5, 5, 17]);
+    }
+
+    #[tokio::test]
+    async fn test_take_offsets_plan_introspection_shows_restoration() {
+        let tmp_dir = tempdir().unwrap();
+        let table = make_test_table(&tmp_dir).await;
+        let take = table
+            .take_offsets(vec![0, 1, 0, 2])
+            .select(Select::Columns(vec!["id".to_string()]))
+            .limit(3);
+
+        let explained = take.explain_plan(false).await.unwrap();
+        assert!(explained.contains("GlobalLimitExec"));
+        assert!(explained.contains("TakeRestoreExec"));
+        assert!(!explained.contains("CoalescePartitionsExec"));
+
+        let analyzed = take.analyze_plan().await.unwrap();
+        assert!(analyzed.contains("GlobalLimitExec"));
+        assert!(analyzed.contains("TakeRestoreExec"));
+        assert!(!analyzed.contains("CoalescePartitionsExec"));
+    }
+
    #[tokio::test]
    async fn test_take_row_ids() {
        let tmp_dir = tempdir().unwrap();
@@ -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;
@@ -533,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,
@@ -588,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
@@ -2708,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| {
@@ -40,8 +40,8 @@ use crate::table::{
 use crate::table::{AnyQuery, Filter, Predicate, PreprocessingOutput, TableStatistics};
 use crate::utils::background_cache::BackgroundCache;
 use crate::utils::{
-    resolve_arrow_field_path, resolve_arrow_fts_field_path, supported_btree_data_type,
-    supported_vector_data_type,
+    MaxBatchLengthStream, TimeoutStream, resolve_arrow_field_path, resolve_arrow_fts_field_path,
+    supported_btree_data_type, supported_vector_data_type,
 };
 use crate::{DistanceType, Error};
 use crate::{
@@ -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());
@@ -2022,6 +2026,9 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
    fn as_any(&self) -> &dyn std::any::Any {
        self
    }
+    fn analyze_plan_is_remote(&self) -> bool {
+        true
+    }
    fn name(&self) -> &str {
        &self.name
    }
@@ -2594,6 +2601,13 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
        query: &AnyQuery,
        options: QueryExecutionOptions,
    ) -> Result<Arc<dyn ExecutionPlan>> {
+        if let AnyQuery::Query(request) = query
+            && let Some(offsets) = &request.take_offsets
+        {
+            return crate::query::create_take_offsets_plan(self, request, offsets, options, false)
+                .await;
+        }
+
        let streams = self.execute_query(query, &options).await?;
        if streams.len() == 1 {
            let stream = streams.into_iter().next().unwrap();
@@ -2612,6 +2626,27 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
        query: &AnyQuery,
        options: QueryExecutionOptions,
    ) -> Result<DatasetRecordBatchStream> {
+        if let AnyQuery::Query(request) = query
+            && let Some(offsets) = &request.take_offsets
+        {
+            let plan = crate::query::create_take_offsets_plan(
+                self,
+                request,
+                offsets,
+                options.clone(),
+                false,
+            )
+            .await?;
+            let inner = execute_plan(plan, Default::default())?;
+            let inner = MaxBatchLengthStream::new_boxed(inner, options.max_batch_length as usize);
+            let inner = if let Some(timeout) = options.timeout {
+                TimeoutStream::new_boxed(inner, timeout)
+            } else {
+                inner
+            };
+            return Ok(DatasetRecordBatchStream::new(inner));
+        }
+
        let streams = self.execute_query(query, &options).await?;

        if streams.len() == 1 {
@@ -2649,6 +2684,12 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
    }

    async fn explain_plan(&self, query: &AnyQuery, verbose: bool) -> Result<String> {
+        if let AnyQuery::Query(request) = query
+            && let Some(offsets) = &request.take_offsets
+        {
+            return crate::query::explain_take_offsets_plan(self, request, offsets, verbose).await;
+        }
+
        let base_request = self
            .client
            .post(&format!("/v1/table/{}/explain_plan/", self.identifier));
@@ -2701,6 +2742,17 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
        query: &AnyQuery,
        options: QueryExecutionOptions,
    ) -> Result<String> {
+        let prepared_query = if let AnyQuery::Query(request) = query
+            && request.take_offsets.is_some()
+        {
+            Some(AnyQuery::Query(
+                crate::query::prepare_take_offsets_request(self, request).await?,
+            ))
+        } else {
+            None
+        };
+        let query = prepared_query.as_ref().unwrap_or(query);
+
        let mut request = self
            .client
            .post(&format!("/v1/table/{}/analyze_plan/", self.identifier));
@@ -3690,7 +3742,7 @@ mod tests {
    };
    use arrow_schema::{DataType, Field, Schema};
    use chrono::{DateTime, Utc};
-    use futures::{StreamExt, TryFutureExt, future::BoxFuture};
+    use futures::{StreamExt, TryFutureExt, TryStreamExt, future::BoxFuture};
    use lance_index::scalar::inverted::{DocumentGranularity, query::MatchQuery};
    use lance_index::scalar::{FullTextSearchQuery, InvertedIndexParams};
    use reqwest::Body;
@@ -5611,6 +5663,114 @@ mod tests {
        assert_eq!(result, "analyzed plan");
    }

+    #[tokio::test]
+    async fn test_take_offsets_explain_plan_does_not_execute_query() {
+        let table = Table::new_with_handler("my_table", |request| {
+            assert_eq!(request.method(), "POST");
+            assert_eq!(request.url().path(), "/v1/table/my_table/explain_plan/");
+
+            http::Response::builder()
+                .status(200)
+                .body(r#""RemoteLookupExec""#)
+                .unwrap()
+        });
+
+        let explained = table
+            .take_offsets(vec![0, 1, 0, 2])
+            .select(crate::query::Select::columns(&["id"]))
+            .limit(3)
+            .explain_plan(false)
+            .await
+            .unwrap();
+
+        assert!(explained.contains("GlobalLimitExec"));
+        assert!(explained.contains("TakeRestoreExec"));
+        assert!(!explained.contains("CoalescePartitionsExec"));
+        assert!(explained.contains("RemoteLookupExec"));
+    }
+
+    #[tokio::test]
+    async fn test_converted_take_request_restores_remote_occurrences() {
+        let table = Table::new_with_handler("my_table", |request| {
+            assert_eq!(request.method(), "POST");
+            assert_eq!(request.url().path(), "/v1/table/my_table/query/");
+
+            let body: serde_json::Value =
+                serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()).unwrap();
+            assert_eq!(body["columns"], json!(["id", "_rowoffset"]));
+
+            let data = RecordBatch::try_new(
+                Arc::new(Schema::new(vec![
+                    Field::new("id", DataType::Int32, false),
+                    Field::new("_rowoffset", DataType::UInt64, false),
+                ])),
+                vec![
+                    Arc::new(Int32Array::from(vec![5])),
+                    Arc::new(arrow_array::UInt64Array::from(vec![5])),
+                ],
+            )
+            .unwrap();
+            http::Response::builder()
+                .status(200)
+                .header(CONTENT_TYPE, ARROW_FILE_CONTENT_TYPE)
+                .body(write_ipc_file(&data))
+                .unwrap()
+        });
+
+        let request = table
+            .take_offsets(vec![5, 5])
+            .select(crate::query::Select::columns(&["id"]))
+            .into_request();
+        let batches = table
+            .base_table()
+            .query(&AnyQuery::Query(request), QueryExecutionOptions::default())
+            .await
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+
+        assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 2);
+        assert!(
+            batches
+                .iter()
+                .all(|batch| batch.schema().fields().len() == 1)
+        );
+    }
+
+    #[tokio::test]
+    async fn test_take_offsets_analyze_plan_delegates_to_remote() {
+        let table = Table::new_with_handler("my_table", |request| {
+            assert_eq!(request.method(), "POST");
+            assert_eq!(request.url().path(), "/v1/table/my_table/analyze_plan/");
+            assert_eq!(
+                request
+                    .url()
+                    .query_pairs()
+                    .find(|(key, _)| key == "distributed_metrics"),
+                Some(("distributed_metrics".into(), "per_worker".into()))
+            );
+
+            http::Response::builder()
+                .status(200)
+                .body(r#""Remote analyzed plan: worker metrics""#)
+                .unwrap()
+        });
+
+        let analyzed = table
+            .take_offsets(vec![0, 1, 0, 2])
+            .select(crate::query::Select::columns(&["id"]))
+            .limit(3)
+            .analyze_plan_with_options(QueryExecutionOptions {
+                analyze_plan_distributed_metrics: AnalyzePlanDistributedMetrics::PerWorker,
+                ..Default::default()
+            })
+            .await
+            .unwrap();
+
+        assert_eq!(analyzed, "Remote analyzed plan: worker metrics");
+    }
+
    #[tokio::test]
    async fn test_query_structured_fts() {
        let table =
@@ -6077,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));
@@ -595,6 +595,14 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
        query: &AnyQuery,
        options: QueryExecutionOptions,
    ) -> Result<String>;
+    /// Whether [`BaseTable::analyze_plan`] is provided by a remote service.
+    ///
+    /// Client-side query wrappers use this to preserve backend metrics and
+    /// distributed-analysis options instead of replacing them with a local plan.
+    #[doc(hidden)]
+    fn analyze_plan_is_remote(&self) -> bool {
+        false
+    }

    /// Add new records to the table.
    async fn add(&self, add: AddDataBuilder) -> Result<AddResult>;
@@ -1652,9 +1660,9 @@ impl Table {
    /// Offsets are useful for sampling as the set of all valid offsets is easily
    /// known in advance to be [0, len(table)).
    ///
-    /// No guarantees are made regarding the order in which results are returned.  If you
-    /// desire an output order that matches the order of the given offsets, you will need
-    /// to add the row offset column to the output and align it yourself.
+    /// No guarantees are made regarding the order in which results are returned.
+    /// Repeated offsets produce repeated rows, which makes this method suitable for
+    /// sampling with replacement.
    ///
    /// Parameters
    /// ----------
@@ -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);
@@ -110,7 +110,7 @@ fn requires_local_namespace_execution(query: &AnyQuery) -> bool {
    // pushing these down would silently ignore the user's setting. For use_lsm that
    // is worse than a tuning miss: MemWAL read routing lives only in `create_plan`,
    // so a pushed-down query would return stale base-only data with no error.
-    if query.base().use_lsm.is_some() {
+    if query.base().use_lsm.is_some() || query.base().take_offsets.is_some() {
        return true;
    }
    matches!(
@@ -154,6 +154,13 @@ pub async fn create_plan(
    options: QueryExecutionOptions,
 ) -> Result<Arc<dyn ExecutionPlan>> {
    let query = query.canonicalized()?;
+    if let AnyQuery::Query(request) = &query
+        && let Some(offsets) = &request.take_offsets
+    {
+        return crate::query::create_take_offsets_plan(table, request, offsets, options, false)
+            .await;
+    }
+
    let query = match query {
        AnyQuery::VectorQuery(query) => query,
        AnyQuery::Query(query) => VectorQueryRequest::from_plain_query(query),
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
lancedb-gatefixer[bot]	19232f9c50	fix: preserve duplicate take offsets (#4024 ) ## Summary - preserve repeated table offsets without adding a public ordering guarantee - retain exact requested ordering in identity and persisted permutations - cover local, projected, multi-batch, and mocked-remote query paths ## Root cause Take queries lowered offsets to a set-like IN predicate and discarded repeated occurrences. Persisted permutation loading also compared the distinct base-table result count with the requested occurrence count, rejecting repeated row IDs before its existing reordering step could expand them. ## Fix The shared take-query path now deduplicates the predicate for efficient lookup, requests row-offset metadata internally, and expands each matching row to the requested multiplicity in backend result order. An internal opt-in keeps exact requested order for identity PermutationReader reads, while persisted permutations continue using their existing ordering map. ## Validation - cargo test --quiet --features remote --tests - cargo check --quiet --features remote --tests --examples - cargo clippy --quiet --features remote --tests --examples - targeted Python local and mocked-remote regression tests - exact issue reproduction Fixes #2820 <!-- lance-gatekeeper-fix:v1 agent=75acf840afa6f4be4bff98b567b504bd generation=1 --> --------- Co-authored-by: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com> Co-authored-by: Xuanwo <github@xuanwo.io>	2026-09-01 20:08:47 +08:00