Compare commits

..

5 Commits

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
17 changed files with 868 additions and 108 deletions
Generated
+48 -47
View File
@@ -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",
+14 -14
View File
@@ -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
+9
View File
@@ -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
+1 -1
View File
@@ -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
View File
@@ -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]: ...
+33
View File
@@ -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)
+97 -10
View File
@@ -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}")
+4
View File
@@ -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))
+1
View File
@@ -548,6 +548,7 @@ class RemoteTable(Table):
LOOP.run(
self._table.create_index(
column,
replace=replace,
config=config,
wait_timeout=wait_timeout,
name=name,
@@ -668,6 +668,167 @@ def test_blob_fields_use_the_scalar_function_semantic_type():
assert signature.output.arrow_type == "blob_v2"
def test_whole_named_struct_function_can_include_a_blob_result_field():
@udf(
input_schema=pa.schema([lancedb.blob("image", nullable=False)]),
output_schema=pa.field(
"payload",
pa.struct(
[
pa.field("mime_type", pa.string(), nullable=False),
lancedb.blob("image", nullable=False),
]
),
nullable=False,
),
)
def inspect_blob(image):
return {"mime_type": "image/png", "image": image}
output = inspect_blob.registration_request.signature.output
assert output.kind == "named_struct"
assert [(field.name, field.arrow_type) for field in output.fields] == [
("mime_type", "utf8"),
("image", "blob_v2"),
]
def test_struct_blob_signature_fields_preserve_exact_metadata_and_nullability():
nested_input = pa.field(
"payload",
pa.struct(
[
pa.field("mime_type", pa.string(), nullable=False),
pa.field(
"nested",
pa.struct([lancedb.blob("image", nullable=True)]),
nullable=True,
),
]
),
nullable=True,
)
nested_output = pa.field(
"result",
pa.struct(
[
pa.field("mime_type", pa.string(), nullable=False),
pa.field(
"nested",
pa.struct([lancedb.blob("image", nullable=True)]),
nullable=False,
),
]
),
nullable=False,
)
@udf(input_schema=pa.schema([nested_input]), output_schema=nested_output)
def copy_payload(payload):
return payload
signature = copy_payload.registration_request.signature
input_type = json.loads(signature.inputs[0].arrow_type)
assert input_type["fields"][1]["nullable"] is True
input_blob = input_type["fields"][1]["type"]["fields"][0]
assert input_blob["nullable"] is True
assert input_blob["metadata"] == {"ARROW:extension:name": "lance.blob.v2"}
assert signature.output.kind == "named_struct"
nested_result = next(
field for field in signature.output.fields if field.name == "nested"
)
output_type = json.loads(nested_result.arrow_type)
output_blob = output_type["fields"][0]
assert output_blob["nullable"] is True
assert output_blob["metadata"] == {"ARROW:extension:name": "lance.blob.v2"}
def test_struct_blob_signature_supports_multiple_struct_levels():
recursive = pa.field(
"value",
pa.struct(
[
pa.field(
"level_1",
pa.struct(
[
pa.field(
"level_2",
pa.struct([lancedb.blob("image", nullable=False)]),
nullable=False,
)
]
),
nullable=False,
)
]
),
nullable=False,
)
@udf(
input_schema=pa.schema([recursive]),
output_schema=pa.field("size", pa.int64(), nullable=False),
)
def blob_size(value):
return len(value["level_1"]["level_2"]["image"])
encoded = json.loads(blob_size.registration_request.signature.inputs[0].arrow_type)
blob = encoded["fields"][0]["type"]["fields"][0]["type"]["fields"][0]
assert blob["metadata"]["ARROW:extension:name"] == "lance.blob.v2"
@pytest.mark.parametrize(
"data_type",
[
pa.list_(lancedb.blob("item", nullable=False)),
pa.large_list(lancedb.blob("item", nullable=False)),
pa.list_(lancedb.blob("item", nullable=False), 2),
pa.map_(pa.string(), lancedb.blob("value", nullable=False).type),
],
)
def test_blob_signature_rejects_collection_ancestors(data_type):
with pytest.raises(
TypeError,
match="Blob v2 fields nested under collection types are not supported",
):
@udf(
input_schema=pa.schema([pa.field("value", data_type, nullable=False)]),
output_schema=pa.field("size", pa.int64(), nullable=False),
)
def blob_size(value):
return len(value)
def test_blob_signature_rejects_collection_below_a_struct():
nested = pa.field(
"value",
pa.struct(
[
pa.field(
"images",
pa.list_(lancedb.blob("item", nullable=False)),
nullable=False,
)
]
),
nullable=False,
)
with pytest.raises(
TypeError,
match="Blob v2 fields nested under collection types are not supported",
):
@udf(
input_schema=pa.schema([nested]),
output_schema=pa.field("size", pa.int64(), nullable=False),
)
def blob_size(value):
return len(value["images"])
def test_named_struct_function_can_include_a_blob_result_field():
@udf(
input_schema=pa.schema([lancedb.blob("image", nullable=False)]),
@@ -729,22 +890,6 @@ def test_blob_marker_rejects_invalid_storage_layout():
return len(image)
def test_nested_blob_signature_field_has_a_clear_error():
nested = pa.field(
"value",
pa.struct([lancedb.blob("image", nullable=False)]),
nullable=False,
)
with pytest.raises(TypeError, match="nested Blob v2 fields are not supported"):
@udf(
input_schema=pa.schema([nested]),
output_schema=pa.field("size", pa.int64(), nullable=False),
)
def blob_size(value):
return len(value["image"])
def test_nested_non_blob_extension_is_not_silently_unwrapped():
class TestExtension(pa.ExtensionType):
def __init__(self):
@@ -1017,6 +1162,8 @@ def test_local_function_catalog_operations_are_not_supported(tmp_path):
db.create_function_async(normalize_score)
with pytest.raises(NotImplementedError, match=message):
db.get_function("normalize_score", version="fv_exact")
with pytest.raises(NotImplementedError, match=message):
db.list_functions()
with pytest.raises(NotImplementedError, match=message):
db.drop_function("normalize_score", version="fv_exact")
@@ -1064,6 +1211,22 @@ def _mock_remote_function_catalog():
"version": "fv_exact",
}
response = state["version"]
elif self.path == "/v1/functions/list":
assert body["include_definition"] is True
if "page_token" not in body:
response = {
"functions": [
{
"name": "normalize_score",
"version": "fv_exact",
"definition": state["version"],
}
],
"page_token": "next",
}
else:
assert body["page_token"] == "next"
response = {"functions": []}
elif self.path == "/v1/functions/drop":
assert body == {
"name": "normalize_score",
@@ -1130,6 +1293,49 @@ def test_blocking_remote_registration_returns_function_version():
]
def test_remote_list_functions_paginates_and_returns_typed_versions():
with _mock_remote_function_catalog() as (host, state):
db = lancedb.connect(
"db://dev",
api_key="fake",
host_override=host,
client_config={"retry_config": {"retries": 0}},
)
created = db.create_function(normalize_score)
state["requests"].clear()
functions = db.list_functions()
assert functions == [created]
assert state["requests"] == [
("/v1/functions/list", {"include_definition": True}),
(
"/v1/functions/list",
{"include_definition": True, "page_token": "next"},
),
]
@pytest.mark.asyncio
async def test_async_remote_list_functions_returns_typed_versions():
with _mock_remote_function_catalog() as (host, state):
db = await lancedb.connect_async(
"db://dev",
api_key="fake",
host_override=host,
client_config={"retry_config": {"retries": 0}},
)
registration = await db.create_function_async(normalize_score)
created = await registration.wait()
state["requests"].clear()
functions = await db.list_functions()
assert functions == [created]
assert [path for path, _ in state["requests"]] == [
"/v1/functions/list",
"/v1/functions/list",
]
def test_remote_drop_function_sends_exact_version():
with _mock_remote_function_catalog() as (host, state):
db = lancedb.connect(
+9 -1
View File
@@ -820,11 +820,13 @@ def test_table_create_indices():
scalar_req = received_requests[0]
assert "name" in scalar_req
assert scalar_req["name"] == "custom_scalar_idx"
assert scalar_req["replace"] is False
# Check FTS index request has custom name
fts_req = received_requests[1]
assert "name" in fts_req
assert fts_req["name"] == "custom_fts_idx"
assert fts_req["replace"] is False
assert fts_req["block_size"] == 256
assert fts_req["custom_stop_words"] == ["cloud"]
@@ -832,6 +834,7 @@ def test_table_create_indices():
vector_req = received_requests[2]
assert "name" in vector_req
assert vector_req["name"] == "custom_vector_idx"
assert "replace" not in vector_req
table.wait_for_index(["custom_scalar_idx"], timedelta(seconds=2))
table.wait_for_index(
@@ -1104,6 +1107,9 @@ def test_remote_create_index_new_api():
table.create_index("text", config=FTS(block_size=256))
# IvfRq via new API
table.create_index("vector", config=IvfRq(distance_type="l2"))
table.create_index(
"vector", config=IvfPq(distance_type="l2"), replace=False
)
# Legacy index_type="IVF_RQ" routes to IvfRq config under the hood.
with pytest.warns(DeprecationWarning, match="create_index"):
@@ -1113,15 +1119,17 @@ def test_remote_create_index_new_api():
num_partitions=8,
)
assert len(received_requests) == 5
assert len(received_requests) == 6
assert [req["column"] for req in received_requests] == [
"vector",
"category",
"text",
"vector",
"vector",
"vector",
]
assert received_requests[2]["block_size"] == 256
assert received_requests[4]["replace"] is False
def test_table_wait_for_index_timeout():
+13
View File
@@ -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,
+22
View File
@@ -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
+4
View File
@@ -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()
+164 -1
View File
@@ -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| {
+38
View File
@@ -527,6 +527,10 @@ impl<S: HttpSend> RemoteTable<S> {
"column": canonical_column
});
if !index.replace {
body["replace"] = false.into();
}
// Add name parameter if provided (for backwards compatibility, only include if Some)
if let Some(ref name) = index.name {
body["name"] = serde_json::Value::String(name.clone());
@@ -6233,6 +6237,40 @@ mod tests {
}
}
#[tokio::test]
async fn test_create_index_forwards_replace_false_on_existing_route() {
let table = Table::new_with_handler("my_table", move |request| {
assert_eq!(request.method(), "POST");
match request.url().path() {
"/v1/table/my_table/describe/" => {
let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]);
http::Response::builder()
.status(200)
.body(describe_response(&schema))
.unwrap()
}
"/v1/table/my_table/create_index/" => {
let body = request.body().unwrap().as_bytes().unwrap();
let body: serde_json::Value = serde_json::from_slice(body).unwrap();
assert_eq!(body["replace"], json!(false));
http::Response::builder()
.status(200)
.body("{}".to_string())
.unwrap()
}
path => panic!("Unexpected path: {}", path),
}
});
table
.create_index(&["a"], Index::BTree(Default::default()))
.replace(false)
.execute()
.await
.unwrap();
}
#[tokio::test]
async fn test_create_index_returns_job() {
let describe_calls = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
+188 -18
View File
@@ -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);