Compare commits

..

1 Commits

Author SHA1 Message Date
Lance Release b4947fe6dc Bump version: 0.38.0-beta.12 → 0.38.0-beta.13 2026-08-30 06:08:51 +00:00
28 changed files with 241 additions and 1407 deletions
+1 -1
View File
@@ -1,5 +1,5 @@
[tool.bumpversion]
current_version = "0.38.0"
current_version = "0.38.0-beta.13"
parse = """(?x)
(?P<major>0|[1-9]\\d*)\\.
(?P<minor>0|[1-9]\\d*)\\.
Generated
+45 -68
View File
@@ -3455,9 +3455,8 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"
[[package]]
name = "fsst"
version = "11.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f727719438dfdb74f358a347c91ff81b6e7084a6421f34de3e473ce271f10caa"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
dependencies = [
"arrow-array",
"rand 0.9.5",
@@ -4816,9 +4815,8 @@ checksum = "e037a2e1d8d5fdbd49b16a4ea09d5d6401c1f29eca5ff29d03d3824dba16256a"
[[package]]
name = "lance"
version = "11.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "be781f40c7a75f9eae2188a2f71174acb7a360dca97163db40b041d0828dea48"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
dependencies = [
"arc-swap",
"arrow",
@@ -4890,9 +4888,8 @@ dependencies = [
[[package]]
name = "lance-arrow"
version = "11.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fb97fd9875f3036d7c2561aa5b16eb87b80ccabaa4eeb5e6099b19cc662f1cd8"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -4914,8 +4911,7 @@ dependencies = [
[[package]]
name = "lance-arrow-scalar"
version = "58.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "771f68b04b47f3addf781116f65061808de94b05e1e9411c23c18f32d14ebe79"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -4929,20 +4925,17 @@ dependencies = [
[[package]]
name = "lance-arrow-stats"
version = "58.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dd47ec33c90bf29f688fd02118e37d3a5ad5c339caa3163f89e417dc0867001f"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
dependencies = [
"arrow-array",
"arrow-schema",
"half",
"lance-arrow-scalar",
]
[[package]]
name = "lance-bitpacking"
version = "11.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6f45658c5b2dc9aada41b66ee44b83af3fa888b7385ae414bae951b12a9f1cd3"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
dependencies = [
"arrayref",
"crunchy",
@@ -4952,9 +4945,8 @@ dependencies = [
[[package]]
name = "lance-core"
version = "11.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "27af3df3a7d08897efccd04461df31cedf0880c4b86a055ddce48e423d27f967"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -4991,9 +4983,8 @@ dependencies = [
[[package]]
name = "lance-datafusion"
version = "11.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c377f837df5296e92f9fad724c83c1bef4e74d5af6e5a9312e9307e1dead8614"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
dependencies = [
"arrow",
"arrow-array",
@@ -5022,9 +5013,8 @@ dependencies = [
[[package]]
name = "lance-datagen"
version = "11.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "778e1a5065fa4bc184e36e32681f10f8f4680ad8cedc9377b4c088dce8c5b8da"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
dependencies = [
"arrow",
"arrow-array",
@@ -5041,9 +5031,8 @@ dependencies = [
[[package]]
name = "lance-derive"
version = "11.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "13e5e95e0fd3d74f7938f4bee623041421b323b5c61f242c8622a1f48a202527"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
dependencies = [
"proc-macro2",
"quote",
@@ -5052,9 +5041,8 @@ dependencies = [
[[package]]
name = "lance-encoding"
version = "11.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1625653c55c65f3426e281f6e29b54c603f38a40bd4cebd707bd4f3ea48be6c5"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
dependencies = [
"arrow-arith",
"arrow-array",
@@ -5087,9 +5075,8 @@ dependencies = [
[[package]]
name = "lance-file"
version = "11.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b7e13c9266b478fc98f36ee19347c4658f7a6613fed77778b1a455fe1b88552e"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
dependencies = [
"arrow-arith",
"arrow-array",
@@ -5120,9 +5107,8 @@ dependencies = [
[[package]]
name = "lance-index"
version = "11.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b0e0cb95f2c4f341c4dd04ac60f6a89ea26a6f75e09570225cbda4854c8b088e"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
dependencies = [
"arc-swap",
"arrow",
@@ -5186,9 +5172,8 @@ dependencies = [
[[package]]
name = "lance-index-core"
version = "11.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "79ccd371977c1f7168da259d66ad37154f23146f093d46136bc7f79559f00f2c"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
dependencies = [
"arrow-array",
"arrow-schema",
@@ -5210,9 +5195,8 @@ dependencies = [
[[package]]
name = "lance-io"
version = "11.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "414d50997391b1ac83dc183c1612fdff88f58b959806078dc4c5e465154566de"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
dependencies = [
"arrow",
"arrow-array",
@@ -5252,9 +5236,8 @@ dependencies = [
[[package]]
name = "lance-linalg"
version = "11.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6ff55b152ef23a56d7ba7e4d1b2c9cf0cc79aef6ee607c115597557ea4059f41"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
dependencies = [
"arrow-array",
"arrow-schema",
@@ -5268,9 +5251,8 @@ dependencies = [
[[package]]
name = "lance-namespace"
version = "11.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "09991c13ab282b731e323619613914e08da9cc82b312f904e58c128b23f2f0e3"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
dependencies = [
"arrow",
"async-trait",
@@ -5282,9 +5264,8 @@ dependencies = [
[[package]]
name = "lance-namespace-impls"
version = "11.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0ec0bc005f6bb8f120774eb4a9ba02e10463d8338167a46cbb1391d46680a174"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
dependencies = [
"arrow",
"arrow-ipc",
@@ -5337,9 +5318,8 @@ dependencies = [
[[package]]
name = "lance-select"
version = "11.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e8f676a2a1837cc85b77feb5326d3296827da964e40d67144f646563302a6ce9"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -5353,9 +5333,8 @@ dependencies = [
[[package]]
name = "lance-table"
version = "11.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bd33054347395048b1d842dfb85a13f7801392c2da5f39425db62f00a481744b"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
dependencies = [
"arrow",
"arrow-array",
@@ -5395,9 +5374,8 @@ dependencies = [
[[package]]
name = "lance-testing"
version = "11.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ecc9ad9ae24f045dfddd538a39e55a28fa7e1ca6ad9f23e20d4087eaf2bb66f7"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
dependencies = [
"arrow-array",
"arrow-schema",
@@ -5410,9 +5388,8 @@ dependencies = [
[[package]]
name = "lance-tokenizer"
version = "11.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3bfa6f0164c8b7056150f5682ce4d415a335b59b04c479873fda04b200117d27"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
dependencies = [
"frostem",
"icu_segmenter",
@@ -5425,7 +5402,7 @@ dependencies = [
[[package]]
name = "lancedb"
version = "0.38.0"
version = "0.38.0-beta.12"
dependencies = [
"ahash",
"anyhow",
@@ -5513,7 +5490,7 @@ dependencies = [
[[package]]
name = "lancedb-nodejs"
version = "0.38.0"
version = "0.38.0-beta.12"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -5538,7 +5515,7 @@ dependencies = [
[[package]]
name = "lancedb-python"
version = "0.38.0"
version = "0.38.0-beta.12"
dependencies = [
"arrow",
"async-trait",
+14 -14
View File
@@ -13,20 +13,20 @@ categories = ["database-implementations"]
rust-version = "1.91.0"
[workspace.dependencies]
lance = { "version" = "=11.0.0", default-features = false }
lance-core = "=11.0.0"
lance-datagen = "=11.0.0"
lance-file = "=11.0.0"
lance-io = { "version" = "=11.0.0", default-features = false }
lance-index = "=11.0.0"
lance-linalg = "=11.0.0"
lance-namespace = "=11.0.0"
lance-namespace-impls = { "version" = "=11.0.0", default-features = false }
lance-table = "=11.0.0"
lance-testing = "=11.0.0"
lance-datafusion = "=11.0.0"
lance-encoding = "=11.0.0"
lance-arrow = "=11.0.0"
lance = { "version" = "=12.0.0-beta.5", default-features = false, "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
lance-core = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
lance-datagen = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
lance-file = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
lance-io = { "version" = "=12.0.0-beta.5", default-features = false, "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
lance-index = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
lance-linalg = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
lance-namespace = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
lance-namespace-impls = { "version" = "=12.0.0-beta.5", default-features = false, "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
lance-table = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
lance-testing = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
lance-datafusion = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
lance-encoding = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
lance-arrow = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
lancedb = { path = "rust/lancedb", default-features = false }
ahash = "0.8"
# Note that this one does not include pyarrow
+1 -1
View File
@@ -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.38.0-beta.13</version>
</dependency>
```
+1 -1
View File
@@ -8,7 +8,7 @@
<parent>
<groupId>com.lancedb</groupId>
<artifactId>lancedb-parent</artifactId>
<version>0.38.0-final.0</version>
<version>0.38.0-beta.13</version>
<relativePath>../pom.xml</relativePath>
</parent>
+2 -2
View File
@@ -6,7 +6,7 @@
<groupId>com.lancedb</groupId>
<artifactId>lancedb-parent</artifactId>
<version>0.38.0-final.0</version>
<version>0.38.0-beta.13</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>11.0.0</lance-core.version>
<lance-core.version>12.0.0-beta.5</lance-core.version>
<spotless.skip>false</spotless.skip>
<spotless.version>2.30.0</spotless.version>
<spotless.java.googlejavaformat.version>1.7</spotless.java.googlejavaformat.version>
+1 -1
View File
@@ -1,7 +1,7 @@
[package]
name = "lancedb-nodejs"
edition.workspace = true
version = "0.38.0"
version = "0.38.0-beta.13"
publish = false
license.workspace = true
description.workspace = true
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-darwin-arm64",
"version": "0.38.0",
"version": "0.38.0-beta.13",
"os": ["darwin"],
"cpu": ["arm64"],
"main": "lancedb.darwin-arm64.node",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-linux-arm64-gnu",
"version": "0.38.0",
"version": "0.38.0-beta.13",
"os": ["linux"],
"cpu": ["arm64"],
"main": "lancedb.linux-arm64-gnu.node",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-linux-arm64-musl",
"version": "0.38.0",
"version": "0.38.0-beta.13",
"os": ["linux"],
"cpu": ["arm64"],
"main": "lancedb.linux-arm64-musl.node",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-linux-x64-gnu",
"version": "0.38.0",
"version": "0.38.0-beta.13",
"os": ["linux"],
"cpu": ["x64"],
"main": "lancedb.linux-x64-gnu.node",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-linux-x64-musl",
"version": "0.38.0",
"version": "0.38.0-beta.13",
"os": ["linux"],
"cpu": ["x64"],
"main": "lancedb.linux-x64-musl.node",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-win32-arm64-msvc",
"version": "0.38.0",
"version": "0.38.0-beta.13",
"os": [
"win32"
],
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-win32-x64-msvc",
"version": "0.38.0",
"version": "0.38.0-beta.13",
"os": ["win32"],
"cpu": ["x64"],
"main": "lancedb.win32-x64-msvc.node",
+1 -1
View File
@@ -11,7 +11,7 @@
"ann"
],
"private": false,
"version": "0.38.0",
"version": "0.38.0-beta.13",
"main": "dist/index.js",
"exports": {
".": "./dist/index.js",
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "lancedb-python"
version = "0.38.0"
version = "0.38.0-beta.13"
publish = false
edition.workspace = true
description = "Python bindings for LanceDB"
+10 -144
View File
@@ -49,25 +49,11 @@ from pydantic import (
model_validator,
)
from .schema import is_blob_v2_field as _is_blob_v2_field
_Int32 = conint(strict=True, ge=-(2**31), le=2**31 - 1)
_UInt32 = conint(strict=True, ge=0, le=2**32 - 1)
_UInt64 = conint(strict=True, ge=0, le=2**64 - 1)
def _validate_gpu_wire_marker(value: Any) -> bool:
if value is not True:
raise ValueError("runtime.gpu must be true")
return True
def _normalize_gpu_marker(value: bool) -> Optional[bool]:
if not isinstance(value, bool):
raise ValueError("gpu must be a boolean")
return True if value else None
class _FrozenDict(dict):
def _immutable(self, *args, **kwargs):
raise TypeError("remote canonical values are immutable")
@@ -253,23 +239,6 @@ class PythonRuntimeSpec(_RemoteValue):
python_version: Optional[str] = None
environment: Optional[PythonEnvironmentSpec] = None
env: Optional[Mapping[str, str]] = None
gpu: Optional[bool] = None
@model_validator(mode="before")
@classmethod
def _discard_unknown_runtime_payload(cls, value):
if isinstance(value, Mapping):
kind = value.get("kind")
if isinstance(kind, str) and kind not in {"python", "python_v2"}:
return {"kind": kind}
return value
@field_validator("gpu", mode="before")
@classmethod
def _validate_gpu_marker(cls, value):
if value is None:
return None
return _validate_gpu_wire_marker(value)
@model_validator(mode="after")
def _validate_runtime_kind(self):
@@ -278,28 +247,18 @@ class PythonRuntimeSpec(_RemoteValue):
raise ValueError("python runtime requires python_version")
if self.environment is None:
raise ValueError("python runtime requires environment")
if self.gpu is not None:
raise ValueError("python runtime with gpu requires kind='python_v2'")
elif self.kind == "python_v2":
if self.python_version is None:
raise ValueError("python_v2 runtime requires python_version")
if self.environment is None:
raise ValueError("python_v2 runtime requires environment")
if self.gpu is None:
raise ValueError("python_v2 runtime requires gpu")
else:
object.__setattr__(self, "python_version", None)
object.__setattr__(self, "environment", None)
object.__setattr__(self, "env", None)
object.__setattr__(self, "gpu", None)
return self
class FunctionVersion(_RemoteValue):
"""An exact immutable Function version returned by Enterprise.
The GPU execution requirement is part of this identity. CPU and memory sizing,
priority, concurrency, and retry policy belong to the execution platform.
Scheduling resources, priority, concurrency, and retry policy belong to
the submitting Job and are not part of this identity.
"""
name: str
@@ -520,7 +479,6 @@ class RefreshColumnResult(_RemoteValue):
_FUNCTION_NAME = re.compile(r"^[A-Za-z_][A-Za-z0-9_.-]*$")
_FUNCTION_BLOB_V2_TYPE = "blob_v2"
_GRAMMAR_PRIMITIVES = (
@@ -537,7 +495,6 @@ _GRAMMAR_PRIMITIVES = (
(pa.float32(), "float32"),
(pa.float64(), "float64"),
(pa.string(), "utf8"),
(pa.large_string(), "large_utf8"),
(pa.binary(), "binary"),
(pa.date32(), "date32"),
(pa.date64(), "date64"),
@@ -585,90 +542,20 @@ def _validate_exact_arrow_field(field: pa.Field) -> None:
"unsupported Arrow type for Function signature: field names "
"must not be empty"
)
if _is_blob_v2_field(field):
if not _has_supported_blob_v2_layout(field):
raise TypeError(
"unsupported Arrow type for Function signature: lance.blob.v2 "
f"requires a supported Blob storage layout, got {field}"
)
elif field.metadata:
if field.metadata:
raise TypeError(
"unsupported Arrow type for Function signature: field metadata "
f"is not supported, got {field}"
)
def _has_supported_blob_v2_layout(field: pa.Field) -> bool:
data_type = field.type
if isinstance(data_type, pa.ExtensionType):
data_type = data_type.storage_type
if not pa.types.is_struct(data_type):
return False
fields = tuple(data_type)
def matches(spec, compare_nullable) -> bool:
return len(fields) == len(spec) and all(
actual.name == name
and actual.type == expected_type
and (not check_nullable or actual.nullable == nullable)
for actual, (name, expected_type, nullable), check_nullable in zip(
fields, spec, compare_nullable
)
)
logical_minimal = (
("data", pa.large_binary(), True),
("uri", pa.utf8(), True),
)
logical_full = logical_minimal + (
("position", pa.uint64(), True),
("size", pa.uint64(), True),
)
prepared = (
("kind", pa.uint8(), True),
("data", pa.large_binary(), True),
("uri", pa.utf8(), True),
("blob_id", pa.uint32(), True),
("blob_size", pa.uint64(), True),
("position", pa.uint64(), True),
)
descriptor = (
("kind", pa.uint8(), False),
("position", pa.uint64(), False),
("size", pa.uint64(), False),
("blob_id", pa.uint32(), False),
("blob_uri", pa.utf8(), False),
)
return (
matches(logical_minimal, (True, True))
or matches(logical_full, (True, True, False, False))
or matches(prepared, (True,) * len(prepared))
or matches(descriptor, (False,) * len(descriptor))
)
def _canonical_arrow_field(field: pa.Field) -> str:
_validate_exact_arrow_field(field)
if _is_blob_v2_field(field):
return _FUNCTION_BLOB_V2_TYPE
return _canonical_arrow_type(field.type)
def _exact_arrow_field(field: pa.Field) -> 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"
)
value = {
return {
"name": field.name,
"nullable": field.nullable,
"type": _exact_arrow_type(field.type),
}
return value
def _exact_arrow_type(data_type: pa.DataType) -> dict[str, Any]:
@@ -792,11 +679,7 @@ def _function_output(output: pa.DataType | pa.Field | pa.Schema) -> FunctionOutp
if output.metadata:
raise TypeError("Function output schema metadata is not supported")
fields = tuple(output)
elif (
isinstance(output, pa.Field)
and not _is_blob_v2_field(output)
and pa.types.is_struct(output.type)
):
elif isinstance(output, pa.Field) and pa.types.is_struct(output.type):
_validate_exact_arrow_field(output)
if output.nullable:
raise ValueError("Function output must be non-nullable")
@@ -818,7 +701,7 @@ def _function_output(output: pa.DataType | pa.Field | pa.Schema) -> FunctionOutp
raise ValueError("Function output must be non-nullable")
return FunctionOutput(
kind="scalar",
arrow_type=_canonical_arrow_field(field),
arrow_type=_canonical_arrow_type(field.type),
nullable=False,
)
@@ -836,7 +719,7 @@ def _function_output(output: pa.DataType | pa.Field | pa.Schema) -> FunctionOutp
fields=tuple(
FunctionResultField(
name=field.name,
arrow_type=_canonical_arrow_field(field),
arrow_type=_canonical_arrow_type(field.type),
nullable=False,
)
for field in fields
@@ -870,7 +753,7 @@ def _infer_signature(
inputs = tuple(
FunctionParameter(
name=field.name,
arrow_type=_canonical_arrow_field(field),
arrow_type=_canonical_arrow_type(field.type),
nullable=field.nullable,
)
for field in input_schema
@@ -893,9 +776,7 @@ def _infer_signature(
inputs.append(
FunctionParameter(
name=parameter.name,
arrow_type=_canonical_arrow_field(
pa.field(parameter.name, data_type, nullable=nullable)
),
arrow_type=_canonical_arrow_type(data_type),
nullable=nullable,
)
)
@@ -1115,7 +996,6 @@ class UdfDefinition:
pip: tuple[str, ...],
env: Mapping[str, str],
python_version: Optional[str],
gpu: bool = False,
conda: tuple[str, ...] = (),
conda_channels: tuple[str, ...] = (),
):
@@ -1144,14 +1024,12 @@ class UdfDefinition:
signature = _infer_signature(function, input_schema, output_schema)
source = _package_source(function)
digest = f"sha256:{hashlib.sha256(source).hexdigest()}"
gpu_marker = _normalize_gpu_marker(gpu)
runtime = PythonRuntimeSpec(
kind="python_v2" if gpu_marker is not None else "python",
kind="python",
python_version=python_version
or f"{sys.version_info.major}.{sys.version_info.minor}",
environment=environment_spec,
env=environment,
gpu=gpu_marker,
)
self._function = function
self._request = FunctionRegistrationRequest(
@@ -1197,7 +1075,6 @@ def udf(
pip: tuple[str, ...] | list[str] = (),
env: Optional[Mapping[str, str]] = None,
python_version: Optional[str] = None,
gpu: bool = False,
conda: tuple[str, ...] | list[str] = (),
conda_channels: tuple[str, ...] | list[str] = (),
) -> Callable[[Callable[..., Any]], UdfDefinition]: ...
@@ -1212,7 +1089,6 @@ def udf(
pip: tuple[str, ...] | list[str] = (),
env: Optional[Mapping[str, str]] = None,
python_version: Optional[str] = None,
gpu: bool = False,
conda: tuple[str, ...] | list[str] = (),
conda_channels: tuple[str, ...] | list[str] = (),
):
@@ -1245,10 +1121,6 @@ def udf(
Environment variables included in the Function definition.
python_version : str, optional
Remote Python major/minor version. Defaults to the client version.
gpu : bool, default False
Whether every remote execution requires a GPU. The execution platform
selects one compatible GPU for each worker. The requirement is part of
the immutable Function version.
The packaged artifact is a snapshot: the function source plus exactly
the module-level names it references (modules as imports, importable
@@ -1273,11 +1145,6 @@ def udf(
... return value * 2
>>> score(1.5)
3.0
>>> @udf(pip=["cupy-cuda12x"], gpu=True)
... def gpu_score(value: int) -> int:
... return value * 2
>>> gpu_score.registration_request.runtime.gpu
True
"""
def decorate(target: Callable[..., Any]) -> UdfDefinition:
@@ -1289,7 +1156,6 @@ def udf(
pip=tuple(pip),
env={} if env is None else env,
python_version=python_version,
gpu=gpu,
conda=tuple(conda),
conda_channels=tuple(conda_channels),
)
@@ -19,13 +19,7 @@ import pyarrow as pa
import pytest
import lancedb
from lancedb.functions import (
PythonRuntimeSpec,
UdfDefinition,
_canonical_arrow_type,
_GRAMMAR_PRIMITIVES,
udf,
)
from lancedb.functions import UdfDefinition, udf
THRESHOLD = 20
_CACHE = None
@@ -95,58 +89,6 @@ def test_udf_conda_environment():
udf(name="channels", conda_channels=["conda-forge"])(lambda value: value)
def test_udf_gpu_marker_uses_gpu_runtime():
@udf(pip=["cupy-cuda12x"], gpu=True)
def double_on_gpu(value: int) -> int:
return value * 2
request = json.loads(double_on_gpu.registration_request.to_canonical_json())
assert request["runtime"]["kind"] == "python_v2"
assert request["runtime"]["gpu"] is True
@udf(pip=["pyarrow"])
def cpu_function(value: int) -> int:
return value
cpu_runtime = json.loads(cpu_function.registration_request.to_canonical_json())[
"runtime"
]
assert cpu_runtime["kind"] == "python"
assert "gpu" not in cpu_runtime
def identity(value: int) -> int:
return value
for invalid in [None, 0, 1, -1, 1.5, "", "true", "1", "H100"]:
with pytest.raises(ValueError, match="gpu must be a boolean"):
udf(name="invalid_gpu", gpu=invalid)(identity)
base_runtime = {
"kind": "python_v2",
"python_version": "3.12",
"environment": {"kind": "pip"},
}
runtime = PythonRuntimeSpec.model_validate({**base_runtime, "gpu": True})
assert runtime.gpu is True
for invalid in [False, 1, 0, "", "true", "1", "H100"]:
with pytest.raises(ValueError, match="runtime.gpu must be true"):
PythonRuntimeSpec.model_validate({**base_runtime, "gpu": invalid})
def test_unknown_runtime_discards_payload_before_known_field_validation():
for payload in [
{"kind": "python_v3", "gpu": {"model": "H100"}},
{"kind": "python_v3", "resources": []},
{
"kind": "python_v3",
"environment": {"kind": []},
"python_version": 3.15,
},
]:
runtime = PythonRuntimeSpec.model_validate(payload)
assert runtime.to_canonical_json() == '{"kind":"python_v3"}'
def test_udf_packages_attribute_access_and_body_imports():
@udf
def word_norm(body: str) -> float:
@@ -227,6 +169,8 @@ def test_udf_resolves_module_globals_before_builtins(tmp_path):
def test_canonical_arrow_type_prefers_the_compact_grammar():
from lancedb.functions import _GRAMMAR_PRIMITIVES, _canonical_arrow_type
golden = json.loads(
(
Path(__file__).parents[3]
@@ -247,6 +191,7 @@ def test_canonical_arrow_type_prefers_the_compact_grammar():
for outside in [
pa.timestamp("us"),
pa.decimal128(10, 2),
pa.large_string(),
pa.large_binary(),
pa.binary(4),
pa.duration("s"),
@@ -440,6 +385,8 @@ def test_udf_recursion_versus_a_rebound_module_name(tmp_path):
def test_canonical_arrow_type_uses_exact_json_for_list_child_properties():
from lancedb.functions import _canonical_arrow_type
nullable = pa.list_(pa.float32())
assert json.loads(_canonical_arrow_type(nullable)) == {
"type": "list",
@@ -529,7 +476,6 @@ def _arrow_type_from_golden(spec: dict) -> pa.DataType:
"null": pa.null(),
"bool": pa.bool_(),
"utf8": pa.string(),
"large_utf8": pa.large_string(),
"binary": pa.binary(),
"float16": pa.float16(),
"float32": pa.float32(),
@@ -546,6 +492,8 @@ def test_arrow_type_grammar_matches_the_shared_golden():
/ "rust/lancedb/tests/fixtures/first_class_functions/v1/arrow_types.json"
).read_text()
)
from lancedb.functions import _canonical_arrow_type
emitted = {
case["arrow_type"]: _canonical_arrow_type(_arrow_type_from_golden(case["json"]))
for case in golden["valid"]
@@ -578,151 +526,6 @@ def test_explicit_arrow_schema_is_deterministic():
assert signature.output.nullable is False
def test_blob_fields_use_the_scalar_function_semantic_type():
@udf(
input_schema=pa.schema([lancedb.blob("image", nullable=False)]),
output_schema=lancedb.blob("result", nullable=False),
)
def copy_blob(image):
return image
signature = copy_blob.registration_request.signature
assert signature.inputs[0].arrow_type == "blob_v2"
assert signature.output.kind == "scalar"
assert signature.output.arrow_type == "blob_v2"
def test_named_struct_function_can_include_a_blob_result_field():
@udf(
input_schema=pa.schema([lancedb.blob("image", nullable=False)]),
output_schema=pa.schema(
[
lancedb.blob("thumbnail", nullable=False),
pa.field("width", pa.int32(), nullable=False),
]
),
)
def inspect_blob(image):
return {"thumbnail": image, "width": 1}
output = inspect_blob.registration_request.signature.output
assert output.kind == "named_struct"
assert [(field.name, field.arrow_type) for field in output.fields] == [
("thumbnail", "blob_v2"),
("width", "int32"),
]
def test_metadata_marked_blob_field_uses_the_semantic_type():
extension = lancedb.blob("image", nullable=False).type
storage = (
extension.storage_type if isinstance(extension, pa.ExtensionType) else extension
)
metadata_blob = pa.field(
"image",
storage,
nullable=False,
metadata={"ARROW:extension:name": "lance.blob.v2"},
)
@udf(
input_schema=pa.schema([metadata_blob]),
output_schema=pa.field("size", pa.int64(), nullable=False),
)
def blob_size(image):
return len(image)
assert blob_size.registration_request.signature.inputs[0].arrow_type == "blob_v2"
def test_blob_marker_rejects_invalid_storage_layout():
malformed = pa.field(
"image",
pa.int64(),
nullable=False,
metadata={"ARROW:extension:name": "lance.blob.v2"},
)
with pytest.raises(TypeError, match="requires a supported Blob storage layout"):
@udf(
input_schema=pa.schema([malformed]),
output_schema=pa.field("size", pa.int64(), nullable=False),
)
def blob_size(image):
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):
super().__init__(pa.int64(), "test.function.extension")
def __arrow_ext_serialize__(self):
return b""
@classmethod
def __arrow_ext_deserialize__(cls, storage_type, serialized):
return cls()
nested = pa.field(
"value",
pa.struct([pa.field("extended", TestExtension(), nullable=False)]),
nullable=False,
)
with pytest.raises(TypeError, match="unsupported Arrow type"):
@udf(
input_schema=pa.schema([nested]),
output_schema=pa.field("result", pa.int64(), nullable=False),
)
def extension_value(value):
return value["extended"]
def test_explicit_large_utf8_schemas_use_the_canonical_function_name():
input_schema = pa.schema([pa.field("text", pa.large_string(), nullable=True)])
output_schema = pa.field("result", pa.large_string(), nullable=False)
@udf(input_schema=input_schema, output_schema=output_schema)
def preserve(text):
return text
signature = preserve.registration_request.signature
assert signature.inputs[0].arrow_type == "large_utf8"
assert signature.inputs[0].nullable is True
assert signature.output.arrow_type == "large_utf8"
assert signature.output.nullable is False
nested = pa.struct([pa.field("text", pa.large_string(), nullable=True)])
assert json.loads(_canonical_arrow_type(nested)) == {
"type": "struct",
"fields": [
{
"name": "text",
"nullable": True,
"type": {"type": "large_utf8"},
}
],
}
def test_nested_struct_output_uses_canonical_exact_json():
token = pa.struct(
[
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "lancedb"
version = "0.38.0"
version = "0.38.0-beta.13"
edition.workspace = true
description = "LanceDB: A serverless, low-latency vector database for AI applications"
license.workspace = true
+38 -120
View File
@@ -13,7 +13,7 @@ use lance::dataset::{ReadParams, WriteMode, builder::DatasetBuilder};
use lance::io::{ObjectStore, ObjectStoreParams, WrappingObjectStore};
use lance_datafusion::utils::StreamingWriteSource;
use lance_file::version::LanceFileVersion;
use lance_io::object_store::{StorageOptionsAccessor, StorageOptionsProvider};
use lance_io::object_store::{ReadDirOptions, StorageOptionsAccessor, StorageOptionsProvider};
use lance_table::io::commit::commit_handler_from_url;
use object_store::local::LocalFileSystem;
use snafu::ResultExt;
@@ -282,11 +282,14 @@ impl std::fmt::Display for ListingDatabase {
const LANCE_EXTENSION: &str = "lance";
/// The table a listed child directory holds, or `None` if it is not a table at all.
/// The table a listed child of the database names, or `None` if the child is not a table.
///
/// A table is the directory `<name>.lance`; a loose file or any other directory under the
/// database prefix belongs to something else. `dir_suffix` is `.lance`, built once by the
/// caller rather than per child.
/// The table a listed child directory holds, or `None` if it is not a table at all.
///
/// Only directories are considered, so a loose object named like a table is not one.
fn table_name(location: &object_store::path::Path, dir_suffix: &str) -> Option<String> {
location
.filename()?
@@ -294,75 +297,6 @@ fn table_name(location: &object_store::path::Path, dir_suffix: &str) -> Option<S
.map(String::from)
.filter(|name| !name.is_empty())
}
/// One page of the table directories under the database directory, in key order.
struct DirPage {
/// The table directories the page holds, as the store lists them.
common_prefixes: Vec<object_store::path::Path>,
/// Resumes after this page, or `None` when the page reached the end of the level.
page_token: Option<String>,
}
/// Where a listed location sits inside the database directory — the space page tokens live
/// in — or `None` if it is not a child of that directory at all. Matching both halves of the
/// prefix drops a location that merely starts with the directory's name (`dbx/y` against
/// `db/`) as well as the marker object some stores keep for the directory itself.
fn relative_key<'a>(prefix: Option<&str>, location: &'a str) -> Option<&'a str> {
let relative = match prefix {
Some(prefix) => location.strip_prefix(prefix)?,
None => location,
};
(!relative.is_empty()).then_some(relative)
}
/// One page of the table directories under `base_path`, one directory level deep.
///
/// Lance 11 exposes no paginated directory listing, so the level is listed in full and paged
/// locally: table directories go into key order (a directory's key keeps its trailing `/`,
/// so a token is never a table name), the page is the smallest `limit` of them past
/// `page_token`, and the token handed back is the key of the last directory the page took —
/// so a page that took nothing ends the listing rather than resuming from a position no page
/// ever reached. Only `<name>.lance/` directories enter the page: loose objects, other
/// directories, and a bare `.lance/` never take a page slot or name a token, which keeps a
/// page to exactly one listing of the level. Correct on every store, at the cost of that one
/// full-level listing per page.
async fn read_dir_page(
object_store: &ObjectStore,
base_path: &object_store::path::Path,
page_token: Option<String>,
limit: Option<usize>,
) -> Result<DirPage> {
let listed = object_store.list_with_delimiter(Some(base_path)).await?;
let prefix = {
let base = base_path.as_ref();
(!base.is_empty()).then(|| format!("{base}/"))
};
let table_dir_suffix = format!(".{LANCE_EXTENSION}/");
let mut children: Vec<(String, object_store::path::Path)> = listed
.common_prefixes
.into_iter()
.filter_map(|location| {
let key = format!("{}/", relative_key(prefix.as_deref(), location.as_ref())?);
(key.len() > table_dir_suffix.len() && key.ends_with(&table_dir_suffix))
.then_some((key, location))
})
.collect();
children.sort_unstable_by(|(left, _), (right, _)| left.cmp(right));
if let Some(resume) = &page_token {
children.retain(|(key, _)| key > resume);
}
let total = children.len();
children.truncate(limit.unwrap_or(total).min(total));
let page_token = match children.last() {
Some((last, _)) if children.len() < total => Some(last.clone()),
_ => None,
};
Ok(DirPage {
common_prefixes: children.into_iter().map(|(_, location)| location).collect(),
page_token,
})
}
const ENGINE: &str = "engine";
const MIRRORED_STORE: &str = "mirroredStore";
@@ -1048,7 +982,8 @@ impl Database for ListingDatabase {
let mut tables = Vec::new();
let mut page_token = request.page_token.filter(|token| !token.is_empty());
// A page of nothing: no table was handed over for a token to resume after.
// A page of nothing: the store rejects a limit of zero, and no table was handed over
// for a token to resume after.
if limit == Some(0) {
return Ok(ListTablesResponse {
context: None,
@@ -1057,21 +992,35 @@ impl Database for ListingDatabase {
});
}
// The page holds only table directories, so one call — and the one full-level
// listing behind it — fills it.
let page = read_dir_page(
&self.object_store,
&self.base_path,
page_token.take(),
limit,
)
.await?;
page_token = page.page_token;
tables.extend(
page.common_prefixes
.iter()
.filter_map(|location| table_name(location, &dir_suffix)),
);
loop {
// Ask only for what the page still has room for, so a database holding more
// than one page costs one request per page rather than one per table.
let listing = self
.object_store
.read_dir_page(
self.base_path.clone(),
ReadDirOptions {
page_token: page_token.take(),
limit: limit.map(|limit| limit - tables.len()),
},
)
.await?;
page_token = listing.page_token;
// Only child directories can be tables, and the store already separates them
// out, so the objects in the page are not looked at.
tables.extend(
listing
.result
.common_prefixes
.iter()
.filter_map(|location| table_name(location, &dir_suffix)),
);
// Children that are not tables leave the page short of the limit, so keep
// going until the page is full or the database runs out.
if page_token.is_none() || limit.is_none_or(|limit| tables.len() >= limit) {
break;
}
}
Ok(ListTablesResponse {
context: None,
@@ -1717,8 +1666,8 @@ mod tests {
}
/// Only directories named `<name>.lance` are tables; loose files and other directories
/// under the database prefix are not. They never take a page slot, so even a `limit`
/// smaller than the clutter ahead of the first table returns that table.
/// under the database prefix are not. A page spent on them is filled from the next one,
/// so a page holding only non-tables does not read as an empty database.
#[tokio::test]
async fn test_listing_ignores_non_table_children() {
let (tempdir, db) = setup_database().await;
@@ -1737,37 +1686,6 @@ mod tests {
assert_eq!(page.tables, vec!["real"]);
}
/// The Lance 11 fallback pages locally over one full-level listing, so a bounded page
/// costs exactly one listing call — clutter ahead of the first table must not buy extra
/// round trips.
#[tokio::test]
async fn test_one_full_listing_per_public_page() {
use crate::io::object_store::io_tracking::IoStatsHolder;
use lance_io::object_store::WrappingObjectStore;
let (tempdir, mut db) = setup_database().await;
create_tables(&db, &["real"]).await;
std::fs::write(tempdir.path().join("aaa-loose.lance"), b"not a table").unwrap();
create_dir_all(tempdir.path().join("aaa-scratch")).unwrap();
let io_stats = IoStatsHolder::default();
let mut tracked_store = (*db.object_store).clone();
tracked_store.inner =
io_stats.wrap(&tracked_store.store_prefix, tracked_store.inner.clone());
db.object_store = Arc::new(tracked_store);
let page = db
.list_tables(ListTablesRequest {
limit: Some(1),
..Default::default()
})
.await
.unwrap();
assert_eq!(page.tables, vec!["real"]);
assert_eq!(io_stats.incremental_stats().read_iops, 1);
}
#[tokio::test]
async fn listing_ignores_empty_table_name() {
let (tempdir, db) = setup_database().await;
+27 -153
View File
@@ -15,9 +15,6 @@ use serde_json::Value;
use crate::{Error, Result};
/// Semantic Function type for a Blob v2 value.
pub const FUNCTION_BLOB_V2_TYPE: &str = "blob_v2";
fn invalid_json(error: impl std::fmt::Display) -> Error {
Error::InvalidInput {
message: format!("invalid remote Function JSON: {error}"),
@@ -210,33 +207,6 @@ pub enum PythonRuntimeSpec {
environment: PythonEnvironmentSpec,
env: BTreeMap<String, String>,
},
/// The GPU-enabled Sophon-managed Python runtime.
///
/// # Examples
///
/// ```
/// use std::collections::BTreeMap;
/// use lancedb::function::{PythonEnvironmentSpec, PythonRuntimeSpec};
///
/// let runtime = PythonRuntimeSpec::PythonV2 {
/// python_version: "3.12".to_string(),
/// environment: PythonEnvironmentSpec {
/// kind: "pip".to_string(),
/// packages: vec!["cupy-cuda12x".to_string()],
/// channels: Vec::new(),
/// path: None,
/// modules: Vec::new(),
/// image: None,
/// },
/// env: BTreeMap::new(),
/// };
/// assert!(runtime.requires_gpu());
/// ```
PythonV2 {
python_version: String,
environment: PythonEnvironmentSpec,
env: BTreeMap<String, String>,
},
/// A runtime kind introduced by a newer server.
///
/// Unknown payload fields are intentionally not retained because the
@@ -249,27 +219,22 @@ impl PythonRuntimeSpec {
pub fn kind(&self) -> &str {
match self {
Self::Python { .. } => "python",
Self::PythonV2 { .. } => "python_v2",
Self::Unrecognized { kind } => kind,
}
}
/// The Python version for a known Python runtime, or `None` for an unknown kind.
/// The Python version for the V1 runtime, or `None` for an unknown kind.
pub fn python_version(&self) -> Option<&str> {
match self {
Self::Python { python_version, .. } | Self::PythonV2 { python_version, .. } => {
Some(python_version)
}
Self::Python { python_version, .. } => Some(python_version),
Self::Unrecognized { .. } => None,
}
}
/// The Python environment for a known Python runtime, or `None` for an unknown kind.
/// The Python environment for the V1 runtime, or `None` for an unknown kind.
pub fn environment(&self) -> Option<&PythonEnvironmentSpec> {
match self {
Self::Python { environment, .. } | Self::PythonV2 { environment, .. } => {
Some(environment)
}
Self::Python { environment, .. } => Some(environment),
Self::Unrecognized { .. } => None,
}
}
@@ -277,73 +242,38 @@ impl PythonRuntimeSpec {
/// Environment variables, or `None` for an unknown kind.
pub fn env(&self) -> Option<&BTreeMap<String, String>> {
match self {
Self::Python { env, .. } | Self::PythonV2 { env, .. } => Some(env),
Self::Python { env, .. } => Some(env),
Self::Unrecognized { .. } => None,
}
}
/// Whether the runtime requires a GPU selected by the execution platform.
pub fn requires_gpu(&self) -> bool {
matches!(self, Self::PythonV2 { .. })
}
}
#[derive(Deserialize)]
struct PythonRuntimeV1Wire {
python_version: String,
environment: PythonEnvironmentSpec,
struct PythonRuntimeWire {
kind: String,
#[serde(default)]
python_version: Option<String>,
#[serde(default)]
environment: Option<PythonEnvironmentSpec>,
#[serde(default)]
env: BTreeMap<String, String>,
#[serde(default)]
gpu: Option<Value>,
}
#[derive(Deserialize)]
struct PythonRuntimeV2Wire {
python_version: String,
environment: PythonEnvironmentSpec,
#[serde(default)]
env: BTreeMap<String, String>,
gpu: bool,
}
impl<'de> Deserialize<'de> for PythonRuntimeSpec {
fn deserialize<D: Deserializer<'de>>(deserializer: D) -> std::result::Result<Self, D::Error> {
let value = Value::deserialize(deserializer)?;
let kind = value
.get("kind")
.ok_or_else(|| de::Error::missing_field("kind"))?
.as_str()
.ok_or_else(|| de::Error::custom("runtime.kind must be a string"))?
.to_string();
match kind.as_str() {
"python" => {
let wire: PythonRuntimeV1Wire =
serde_json::from_value(value).map_err(de::Error::custom)?;
if wire.gpu.is_some() {
return Err(de::Error::custom(
"python runtime with gpu requires kind='python_v2'",
));
}
Ok(Self::Python {
python_version: wire.python_version,
environment: wire.environment,
env: wire.env,
})
}
"python_v2" => {
let wire: PythonRuntimeV2Wire =
serde_json::from_value(value).map_err(de::Error::custom)?;
if !wire.gpu {
return Err(de::Error::custom("runtime.gpu must be true"));
}
Ok(Self::PythonV2 {
python_version: wire.python_version,
environment: wire.environment,
env: wire.env,
})
}
_ => Ok(Self::Unrecognized { kind }),
let wire = PythonRuntimeWire::deserialize(deserializer)?;
if wire.kind == "python" {
Ok(Self::Python {
python_version: wire
.python_version
.ok_or_else(|| de::Error::missing_field("python_version"))?,
environment: wire
.environment
.ok_or_else(|| de::Error::missing_field("environment"))?,
env: wire.env,
})
} else {
Ok(Self::Unrecognized { kind: wire.kind })
}
}
}
@@ -357,8 +287,6 @@ impl Serialize for PythonRuntimeSpec {
environment: &'a PythonEnvironmentSpec,
#[serde(skip_serializing_if = "BTreeMap::is_empty")]
env: &'a BTreeMap<String, String>,
#[serde(skip_serializing_if = "Option::is_none")]
gpu: Option<bool>,
}
#[derive(Serialize)]
@@ -376,19 +304,6 @@ impl Serialize for PythonRuntimeSpec {
python_version,
environment,
env,
gpu: None,
}
.serialize(serializer),
Self::PythonV2 {
python_version,
environment,
env,
} => PythonRuntimeRef {
kind: "python_v2",
python_version,
environment,
env,
gpu: Some(true),
}
.serialize(serializer),
Self::Unrecognized { kind } => UnrecognizedRuntimeRef { kind }.serialize(serializer),
@@ -398,8 +313,8 @@ impl Serialize for PythonRuntimeSpec {
/// Immutable Function version returned by the Enterprise catalog.
///
/// The GPU execution requirement is part of this identity. CPU and memory sizing,
/// priority, concurrency, and retry policy belong to the execution platform.
/// Scheduling resources, priority, concurrency, and retry policy belong to
/// the submitting Job and are not part of this identity.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct FunctionVersion {
name: String,
@@ -674,7 +589,7 @@ impl_json!(RefreshColumnResult);
#[cfg(test)]
mod conda_environment_tests {
use super::{PythonEnvironmentSpec, PythonRuntimeSpec};
use super::PythonEnvironmentSpec;
#[test]
fn conda_channels_round_trip_and_pip_stays_bare() {
@@ -693,45 +608,4 @@ mod conda_environment_tests {
serde_json::from_str(r#"{"kind":"pip","packages":["numpy"]}"#).unwrap();
assert!(!serde_json::to_string(&pip).unwrap().contains("channels"));
}
#[test]
fn gpu_python_runtime_marker_round_trips_and_validates() {
let runtime: PythonRuntimeSpec = serde_json::from_str(
r#"{"kind":"python_v2","python_version":"3.12","environment":{"kind":"pip"},"gpu":true}"#,
)
.unwrap();
assert_eq!(runtime.kind(), "python_v2");
assert!(runtime.requires_gpu());
assert_eq!(
super::canonical_json(&runtime).unwrap(),
r#"{"environment":{"kind":"pip"},"gpu":true,"kind":"python_v2","python_version":"3.12"}"#
);
for invalid in [
r#"{"kind":"python","python_version":"3.12","environment":{"kind":"pip"},"gpu":true}"#,
r#"{"kind":"python_v2","python_version":"3.12","environment":{"kind":"pip"}}"#,
r#"{"kind":"python_v2","python_version":"3.12","environment":{"kind":"pip"},"gpu":1}"#,
r#"{"kind":"python_v2","python_version":"3.12","environment":{"kind":"pip"},"gpu":false}"#,
r#"{"kind":"python_v2","python_version":"3.12","environment":{"kind":"pip"},"gpu":"true"}"#,
r#"{"kind":"python_v2","python_version":"3.12","environment":{"kind":"pip"},"gpu":"H100"}"#,
] {
assert!(serde_json::from_str::<PythonRuntimeSpec>(invalid).is_err());
}
}
#[test]
fn unknown_runtime_discards_payload_before_known_field_validation() {
for encoded in [
r#"{"kind":"python_v3","gpu":{"model":"H100"}}"#,
r#"{"kind":"python_v3","resources":[]}"#,
r#"{"kind":"python_v3","python_version":3.15,"environment":{"kind":[]}}"#,
] {
let runtime: PythonRuntimeSpec = serde_json::from_str(encoded).unwrap();
assert_eq!(runtime.kind(), "python_v3");
assert_eq!(
super::canonical_json(&runtime).unwrap(),
r#"{"kind":"python_v3"}"#
);
}
}
}
+9 -1
View File
@@ -10,7 +10,7 @@ use lance::io::WrappingObjectStore;
use object_store::{
CopyOptions, Error, GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta,
ObjectStore, ObjectStoreExt, PutMultipartOptions, PutOptions, PutPayload, PutResult, Result,
UploadPart, path::Path,
UploadPart, list::PaginatedListStore, path::Path,
};
use async_trait::async_trait;
@@ -187,6 +187,14 @@ impl WrappingObjectStore for MirroringObjectStoreWrapper {
secondary: self.secondary.clone(),
})
}
fn wrap_paginated(
&self,
_store_prefix: &str,
original: Arc<dyn PaginatedListStore>,
) -> Option<Arc<dyn PaginatedListStore>> {
Some(original)
}
}
// windows pathing can't be simply concatenated
@@ -12,7 +12,7 @@ use lance::io::WrappingObjectStore;
use object_store::{
CopyOptions, GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta, ObjectStore,
PutMultipartOptions, PutOptions, PutPayload, PutResult, RenameOptions, Result as OSResult,
UploadPart, path::Path,
UploadPart, list::PaginatedListStore, path::Path,
};
#[derive(Debug, Default)]
@@ -57,6 +57,14 @@ impl WrappingObjectStore for IoStatsHolder {
stats: self.0.clone(),
})
}
fn wrap_paginated(
&self,
_store_prefix: &str,
original: Arc<dyn PaginatedListStore>,
) -> Option<Arc<dyn PaginatedListStore>> {
Some(original)
}
}
impl IoTrackingStore {
-87
View File
@@ -7464,93 +7464,6 @@ mod tests {
assert_eq!(result.version, 8);
}
#[tokio::test]
async fn test_add_function_column_allows_an_existing_binding() {
let binding = crate::function::FunctionBinding::from_json(include_str!(
"../../tests/fixtures/first_class_functions/v1/remote_function_binding.json"
))
.unwrap();
let binding_metadata = crate::table::computed_columns::function_bindings_metadata(
std::slice::from_ref(&binding),
)
.unwrap();
let mut fields = vec![
Field::new("title", DataType::Utf8, true),
Field::new("body", DataType::Utf8, true),
];
fields.extend(binding.outputs().iter().map(|output| {
let data_type = match output.arrow_type.as_str() {
"utf8" => DataType::Utf8,
"int64" => DataType::Int64,
other => panic!("unexpected fixture output type {other}"),
};
Field::new(&output.output_name, data_type, true).with_metadata(
crate::table::computed_columns::function_computed_column_metadata(
binding.binding_id(),
output.output_ordinal,
&["title".into(), "body".into()],
),
)
}));
let schema = Schema::new_with_metadata(
fields,
HashMap::from([(
crate::table::computed_columns::FUNCTION_BINDINGS_META_KEY.to_string(),
binding_metadata,
)]),
);
let table =
Table::new_with_handler("my_table", move |request| match request.url().path() {
"/v1/table/my_table/describe/" => http::Response::builder()
.status(200)
.body(describe_response(&schema))
.unwrap(),
"/v1/table/my_table/add_columns/" => {
let actual: serde_json::Value =
serde_json::from_slice(request.body().unwrap().as_bytes().unwrap())
.unwrap();
assert_eq!(
actual["new_columns"],
serde_json::json!([
{"name":"secondary_text","all_null":true},
{"name":"secondary_token_count","all_null":true}
])
);
http::Response::builder()
.status(200)
.body(r#"{"version":10}"#.to_string())
.unwrap()
}
path => panic!("Unexpected path: {path}"),
});
let application = crate::function::FunctionApplication::from_json(
r#"{
"function":{"name":"text_features","version":"fv_01K3TEXT"},
"inputs":[
{"parameter":"title","kind":"column","value":{"path":"title"}},
{"parameter":"body","kind":"column","value":{"path":"body"}}
],
"output":{"kind":"named_struct","fields":[
{"name":"normalized_text","arrow_type":"utf8","nullable":false},
{"name":"token_count","arrow_type":"int64","nullable":false}
]},
"columns":{
"normalized_text":"secondary_text",
"token_count":"secondary_token_count"
}
}"#,
)
.unwrap();
let result = table
.add_columns()
.function(application)
.execute()
.await
.unwrap();
assert_eq!(result.version, 10);
}
#[tokio::test]
async fn test_add_fixed_size_list_function_column_declares_the_vector_type() {
let table = Table::new_with_handler("my_table", |request| {
+16
View File
@@ -4183,6 +4183,14 @@ mod tests {
parent_list_calls: self.parent_list_calls.clone(),
})
}
fn wrap_paginated(
&self,
_store_prefix: &str,
_original: Arc<dyn object_store::list::PaginatedListStore>,
) -> Option<Arc<dyn object_store::list::PaginatedListStore>> {
None
}
}
#[tokio::test]
@@ -4286,6 +4294,14 @@ mod tests {
self.called.store(true, Ordering::Relaxed);
original
}
fn wrap_paginated(
&self,
_store_prefix: &str,
original: Arc<dyn object_store::list::PaginatedListStore>,
) -> Option<Arc<dyn object_store::list::PaginatedListStore>> {
Some(original)
}
}
#[tokio::test]
+47 -560
View File
@@ -29,16 +29,14 @@ use datafusion_common::{ScalarValue, tree_node::TreeNode};
use datafusion_expr::Expr;
use datafusion_physical_plan::PhysicalExpr;
use lance::dataset::NewColumnTransform;
use lance_arrow::{ARROW_EXT_NAME_KEY, BLOB_V2_EXT_NAME, FieldExt};
use lance_core::datatypes::{
BLOB_V2_DESC_FIELD, BlobV2Layout, format_field_path_minimal, parse_field_path,
};
use lance_arrow::FieldExt;
use lance_core::datatypes::{BLOB_V2_DESC_FIELD, format_field_path_minimal, parse_field_path};
use lance_datafusion::planner::Planner;
use lance_namespace::models::{JsonArrowDataType, JsonArrowField, JsonArrowSchema};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use crate::function::{FUNCTION_BLOB_V2_TYPE, FunctionApplication, FunctionBinding};
use crate::function::{FunctionApplication, FunctionBinding};
use crate::utils::resolve_arrow_field_path;
use crate::{Error, Result};
@@ -549,12 +547,7 @@ fn ensure_known_binding_shape(value: &Value) -> Result<()> {
Ok(())
}
struct ResolvedFieldPath<'a> {
root: &'a ArrowField,
leaf: &'a ArrowField,
}
fn resolve_field_path<'a>(schema: &'a ArrowSchema, path: &str) -> Result<ResolvedFieldPath<'a>> {
fn resolve_field_path<'a>(schema: &'a ArrowSchema, path: &str) -> Result<&'a ArrowField> {
let parts = lance_core::datatypes::parse_field_path(path).map_err(|e| {
invalid_function(format!("invalid Function input field path '{path}': {e}"))
})?;
@@ -563,43 +556,25 @@ fn resolve_field_path<'a>(schema: &'a ArrowSchema, path: &str) -> Result<Resolve
"Function input field path cannot be empty",
));
};
let root = schema
let mut field = schema
.field_with_name(root)
.map_err(|_| invalid_function(format!("unknown Function input column '{path}'")))?;
let mut leaf = root;
for child in children {
let DataType::Struct(fields) = leaf.data_type() else {
let DataType::Struct(fields) = field.data_type() else {
return Err(invalid_function(format!(
"Function input field path '{path}' traverses a non-struct field"
)));
};
leaf = fields
field = fields
.iter()
.find(|field| field.name() == child)
.map(AsRef::as_ref)
.ok_or_else(|| invalid_function(format!("unknown Function input column '{path}'")))?;
}
Ok(ResolvedFieldPath { root, leaf })
Ok(field)
}
fn canonical_input_arrow_type(field: &JsonArrowField) -> Result<String> {
let is_blob_v2 = field
.metadata
.as_ref()
.and_then(|metadata| metadata.get(ARROW_EXT_NAME_KEY))
.map(String::as_str)
== Some(BLOB_V2_EXT_NAME);
if is_blob_v2 {
let arrow_field = lance_namespace::schema::convert_json_arrow_field(field)
.map_err(|e| invalid_function(format!("invalid Function input field: {e}")))?;
if !has_supported_blob_v2_layout(&arrow_field) {
return Err(invalid_function(format!(
"Function input '{}' has an invalid Blob v2 storage layout",
arrow_field.name()
)));
}
return Ok(FUNCTION_BLOB_V2_TYPE.to_string());
}
if field.r#type.fields.is_none() && field.r#type.length.is_none() {
Ok(field.r#type.r#type.clone())
} else {
@@ -609,14 +584,6 @@ fn canonical_input_arrow_type(field: &JsonArrowField) -> Result<String> {
}
}
fn has_supported_blob_v2_layout(field: &ArrowField) -> bool {
field.is_blob_v2()
&& matches!(
field.data_type(),
DataType::Struct(fields) if BlobV2Layout::classify(fields).is_some()
)
}
/// `fixed_size_list<item, size>` -> (`item`, `size`); the comma must sit outside
/// any nested `<...>`.
fn split_fixed_size_list(raw: &str) -> Option<(&str, i32)> {
@@ -696,81 +663,10 @@ fn parse_output_arrow_type(raw: &str) -> Result<JsonArrowDataType> {
Ok(data_type)
}
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),
]))
.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)?,
))
}
fn function_output_field_matches(expected: &ArrowField, actual: &ArrowField) -> bool {
expected.name() == actual.name()
&& expected.is_nullable() == actual.is_nullable()
&& if expected.is_blob_v2() {
has_supported_blob_v2_layout(expected) && has_supported_blob_v2_layout(actual)
} else {
function_output_type_matches(expected.data_type(), actual.data_type())
}
}
fn function_output_type_matches(expected: &DataType, actual: &DataType) -> bool {
if expected == actual {
return true;
}
match (expected, actual) {
(DataType::Struct(expected), DataType::Struct(actual)) => {
expected.len() == actual.len()
&& expected
.iter()
.zip(actual)
.all(|(expected, actual)| function_output_field_matches(expected, actual))
}
(DataType::List(expected), DataType::List(actual))
| (DataType::LargeList(expected), DataType::LargeList(actual)) => {
function_output_field_matches(expected, actual)
}
(
DataType::FixedSizeList(expected, expected_size),
DataType::FixedSizeList(actual, actual_size),
) => expected_size == actual_size && function_output_field_matches(expected, actual),
(DataType::Map(expected, expected_sorted), DataType::Map(actual, actual_sorted)) => {
expected_sorted == actual_sorted && function_output_field_matches(expected, actual)
}
_ => false,
}
}
fn function_output_type_has_blob(data_type: &DataType) -> bool {
match data_type {
DataType::Struct(fields) => fields
.iter()
.any(|field| field.is_blob_v2() || function_output_type_has_blob(field.data_type())),
DataType::List(field)
| DataType::LargeList(field)
| DataType::FixedSizeList(field, _)
| DataType::Map(field, _) => {
field.is_blob_v2() || function_output_type_has_blob(field.data_type())
}
_ => false,
}
}
fn ensure_binding_matches_schema(schema: &ArrowSchema, binding: &FunctionBinding) -> Result<()> {
let mut input_fields = Vec::with_capacity(binding.inputs().len());
for input in binding.inputs() {
let resolved = resolve_field_path(schema, &input.field_path)?;
let field = resolved.leaf;
let field = resolve_field_path(schema, &input.field_path)?;
if field
.metadata()
.get(COMPUTED_COLUMN_META_KEY)
@@ -825,11 +721,6 @@ fn ensure_binding_matches_schema(schema: &ArrowSchema, binding: &FunctionBinding
)));
}
let expected_inputs = binding
.inputs()
.iter()
.map(|input| input.field_path.clone())
.collect::<Vec<_>>();
let mut output_fields = Vec::with_capacity(binding.outputs().len());
for output in binding.outputs() {
let field = schema.field_with_name(&output.output_name).map_err(|_| {
@@ -846,61 +737,25 @@ fn ensure_binding_matches_schema(schema: &ArrowSchema, binding: &FunctionBinding
binding.binding_id()
)));
}
let (type_matches, has_semantic_blob) = if output.arrow_type == FUNCTION_BLOB_V2_TYPE {
(has_supported_blob_v2_layout(field), true)
} else {
let expected_type = parse_output_arrow_type(&output.arrow_type)?;
let expected_type = lance_namespace::schema::convert_json_arrow_type(&expected_type)
.map_err(|e| invalid_function(format!("invalid Function output type: {e}")))?;
(
function_output_type_matches(&expected_type, field.data_type()),
function_output_type_has_blob(&expected_type),
)
};
if !type_matches {
let expected_type = parse_output_arrow_type(&output.arrow_type)?;
let expected_type = lance_namespace::schema::convert_json_arrow_type(&expected_type)
.map_err(|e| invalid_function(format!("invalid Function output type: {e}")))?;
if field.data_type() != &expected_type {
return Err(invalid_function(format!(
"Function output '{}' type no longer matches binding '{}'",
output.output_name,
binding.binding_id()
)));
}
let metadata = field.metadata();
let declared_inputs = metadata
.get(INPUTS_META_KEY)
.and_then(|raw| serde_json::from_str::<Vec<String>>(raw).ok());
if metadata.get(COMPUTED_COLUMN_META_KEY).map(String::as_str) != Some("true")
|| metadata.get(KIND_META_KEY).map(String::as_str) != Some(FUNCTION_KIND)
|| metadata
.get(FUNCTION_BINDING_ID_META_KEY)
.map(String::as_str)
!= Some(binding.binding_id())
|| metadata
.get(FUNCTION_OUTPUT_ORDINAL_META_KEY)
.and_then(|value| value.parse::<u32>().ok())
!= Some(output.output_ordinal)
|| declared_inputs.as_deref() != Some(expected_inputs.as_slice())
{
return Err(invalid_function(format!(
"Function output '{}' declaration metadata does not match binding '{}'",
output.output_name,
binding.binding_id()
)));
}
if has_semantic_blob {
output_fields.push(function_output_field(
field.name(),
true,
&output.arrow_type,
)?);
} else {
let json = lance_namespace::schema::arrow_schema_to_json(&ArrowSchema::new(vec![
ArrowField::new(field.name().clone(), field.data_type().clone(), true),
]))
.map_err(|e| invalid_function(format!("invalid Function output schema: {e}")))?;
output_fields.push(json.fields.into_iter().next().unwrap());
}
output_fields.push(ArrowField::new(
field.name().clone(),
field.data_type().clone(),
true,
));
}
let output_schema = JsonArrowSchema::new(output_fields);
let output_schema =
lance_namespace::schema::arrow_schema_to_json(&ArrowSchema::new(output_fields))
.map_err(|e| invalid_function(format!("invalid Function output schema: {e}")))?;
let output_schema = serde_json::to_value(output_schema).map_err(|e| {
invalid_function(format!(
"could not encode exact Function output schema: {e}"
@@ -923,7 +778,7 @@ pub(crate) fn plan_function_application(
application: &FunctionApplication,
output_name: Option<&str>,
) -> Result<FunctionDeclarationPlan> {
ensure_supported_function_metadata(schema)?;
ensure_no_function_bindings_for_mutation(schema, "Function binding declaration")?;
if application.has_unknown_fields() {
return Err(Error::NotSupported {
message: "Function application contains fields from a newer contract".into(),
@@ -973,9 +828,8 @@ pub(crate) fn plan_function_application(
input.parameter
))
})?;
let resolved = resolve_field_path(schema, path)?;
if resolved
.root
let field = resolve_field_path(schema, path)?;
if field
.metadata()
.get(COMPUTED_COLUMN_META_KEY)
.map(String::as_str)
@@ -985,7 +839,6 @@ pub(crate) fn plan_function_application(
"Function input '{path}' is computed; computed-on-computed bindings are not supported"
)));
}
let field = resolved.leaf;
let parameter_field = ArrowField::new(
input.parameter.clone(),
field.data_type().clone(),
@@ -1029,15 +882,16 @@ pub(crate) fn plan_function_application(
"Function logical outputs must be non-nullable during NULL assignment",
));
}
let arrow_type = output.arrow_type.as_deref().ok_or_else(|| {
invalid_function("scalar Function output is missing its Arrow type")
})?;
let data_type =
parse_output_arrow_type(output.arrow_type.as_deref().ok_or_else(|| {
invalid_function("scalar Function output is missing its Arrow type")
})?)?;
outputs.push(FunctionOutputTarget {
result_field: WHOLE_RESULT_FIELD.to_string(),
output_name: name.to_string(),
output_ordinal: 0,
});
output_fields.push(function_output_field(name, true, arrow_type)?);
output_fields.push(JsonArrowField::new(name.to_string(), true, data_type));
}
"named_struct" => {
if output.fields.is_empty() {
@@ -1081,7 +935,13 @@ pub(crate) fn plan_function_application(
let fields = output
.fields
.iter()
.map(|field| function_output_field(&field.name, false, &field.arrow_type))
.map(|field| {
Ok(JsonArrowField::new(
field.name.clone(),
false,
parse_output_arrow_type(&field.arrow_type)?,
))
})
.collect::<Result<Vec<_>>>()?;
let mut data_type = JsonArrowDataType::new("struct".to_string());
data_type.fields = Some(fields);
@@ -1108,7 +968,11 @@ pub(crate) fn plan_function_application(
output_name: name.clone(),
output_ordinal: ordinal as u32,
});
output_fields.push(function_output_field(name, true, &field.arrow_type)?);
output_fields.push(JsonArrowField::new(
name.clone(),
true,
parse_output_arrow_type(&field.arrow_type)?,
));
}
}
}
@@ -1745,7 +1609,7 @@ mod tests {
}
use arrow_array::record_batch;
use arrow_schema::{DataType, TimeUnit};
use arrow_schema::DataType;
use futures::TryStreamExt;
use lance::dataset::ColumnAlteration;
@@ -2706,73 +2570,6 @@ mod tests {
.unwrap()
}
fn blob_application(output: &str) -> FunctionApplication {
FunctionApplication::from_json(&format!(
r#"{{
"function":{{"name":"blob_features","version":"fv_blob"}},
"inputs":[
{{"parameter":"image","kind":"column","value":{{"path":"image"}}}}
],
"output":{output}
}}"#
))
.unwrap()
}
fn binding_from_plan(plan: &FunctionDeclarationPlan) -> FunctionBinding {
let inputs = plan
.input_bindings
.iter()
.enumerate()
.map(|(index, input)| {
serde_json::json!({
"parameter": input.parameter,
"field_id": index,
"field_path": input.field_path,
"arrow_type": input.arrow_type,
"nullable": input.nullable,
})
})
.collect::<Vec<_>>();
let outputs = plan
.outputs
.iter()
.zip(&plan.output_schema.fields)
.enumerate()
.map(|(index, (output, field))| {
serde_json::json!({
"result_field": output.result_field,
"output_name": output.output_name,
"output_field_id": 100 + index,
"output_ordinal": output.output_ordinal,
"arrow_type": canonical_input_arrow_type(field).unwrap(),
"nullable": false,
})
})
.collect::<Vec<_>>();
FunctionBinding::from_json(
&serde_json::json!({
"binding_id": "fb_blob",
"function": plan.application.function(),
"inputs": inputs,
"outputs": outputs,
"input_schema": plan.input_schema,
"output_schema": plan.output_schema,
})
.to_string(),
)
.unwrap()
}
fn full_blob_field(name: &str, nullable: bool) -> ArrowField {
ArrowField::new(
name,
DataType::Struct(lance_core::datatypes::BLOB_V2_LOGICAL_FIELDS.clone()),
nullable,
)
.with_metadata(crate::blob(name, nullable).metadata().clone())
}
fn function_binding_schema(title_nullable: bool, body_nullable: bool) -> ArrowSchema {
ArrowSchema::new(vec![
ArrowField::new("title", DataType::Utf8, title_nullable),
@@ -2782,37 +2579,6 @@ mod tests {
])
}
fn valid_function_binding_schema(
title_nullable: bool,
body_nullable: bool,
binding: &FunctionBinding,
) -> ArrowSchema {
let mut fields = function_binding_schema(title_nullable, body_nullable)
.fields()
.iter()
.map(|field| field.as_ref().clone())
.collect::<Vec<_>>();
let inputs = binding
.inputs()
.iter()
.map(|input| input.field_path.clone())
.collect::<Vec<_>>();
for output in binding.outputs() {
let index = fields
.iter()
.position(|field| field.name() == &output.output_name)
.unwrap();
fields[index] = fields[index]
.clone()
.with_metadata(function_computed_column_metadata(
binding.binding_id(),
output.output_ordinal,
&inputs,
));
}
ArrowSchema::new(fields)
}
#[test]
fn test_non_nullable_function_inputs_can_bind_to_nullable_parameters() {
let binding = FunctionBinding::from_json(include_str!(
@@ -2820,11 +2586,7 @@ mod tests {
))
.unwrap();
ensure_binding_matches_schema(
&valid_function_binding_schema(false, false, &binding),
&binding,
)
.unwrap();
ensure_binding_matches_schema(&function_binding_schema(false, false), &binding).unwrap();
}
#[test]
@@ -2837,11 +2599,8 @@ mod tests {
raw_binding["input_schema"]["fields"][0]["nullable"] = Value::Bool(false);
let binding: FunctionBinding = serde_json::from_value(raw_binding).unwrap();
let err = ensure_binding_matches_schema(
&valid_function_binding_schema(true, false, &binding),
&binding,
)
.unwrap_err();
let err = ensure_binding_matches_schema(&function_binding_schema(true, false), &binding)
.unwrap_err();
assert!(
matches!(&err, Error::InvalidInput { message }
if message.contains("input column 'title' is nullable")
@@ -2852,73 +2611,6 @@ mod tests {
);
}
#[test]
fn test_second_binding_rejects_outputs_without_reciprocal_metadata() {
let binding = FunctionBinding::from_json(include_str!(
"../../tests/fixtures/first_class_functions/v1/remote_function_binding.json"
))
.unwrap();
let schema = ArrowSchema::new_with_metadata(
function_binding_schema(true, true).fields().to_vec(),
HashMap::from([(
FUNCTION_BINDINGS_META_KEY.to_string(),
function_bindings_metadata(std::slice::from_ref(&binding)).unwrap(),
)]),
);
let err = plan_function_application(
&schema,
&named_struct_application(
r#"{"normalized_text":"secondary_text","token_count":"secondary_token_count"}"#,
),
None,
)
.unwrap_err();
assert!(
matches!(&err, Error::InvalidInput { message }
if message.contains("declaration metadata")
&& message.contains("fb_01K3TEXT")),
"{err:?}"
);
}
#[test]
fn test_persisted_nested_input_keeps_leaf_level_validation() {
let mut raw_binding: Value = serde_json::from_str(include_str!(
"../../tests/fixtures/first_class_functions/v1/remote_function_binding.json"
))
.unwrap();
raw_binding["inputs"][0]["field_path"] = Value::String("title.value".to_string());
let binding: FunctionBinding = serde_json::from_value(raw_binding).unwrap();
let title = ArrowField::new(
"title",
DataType::Struct(vec![ArrowField::new("value", DataType::Utf8, true)].into()),
true,
)
.with_metadata(HashMap::from([
(COMPUTED_COLUMN_META_KEY.to_string(), "true".to_string()),
(KIND_META_KEY.to_string(), SQL_KIND.to_string()),
(EXPRESSION_META_KEY.to_string(), "title".to_string()),
]));
let mut fields = vec![title, ArrowField::new("body", DataType::Utf8, true)];
fields.extend(binding.outputs().iter().map(|output| {
let data_type = match output.arrow_type.as_str() {
"utf8" => DataType::Utf8,
"int64" => DataType::Int64,
other => panic!("unexpected fixture output type {other}"),
};
ArrowField::new(&output.output_name, data_type, true).with_metadata(
function_computed_column_metadata(
binding.binding_id(),
output.output_ordinal,
&["title.value".into(), "body".into()],
),
)
}));
ensure_binding_matches_schema(&ArrowSchema::new(fields), &binding).unwrap();
}
#[test]
fn test_function_binding_metadata_survives_schema_round_trip() {
let binding = FunctionBinding::from_json(include_str!(
@@ -2967,36 +2659,9 @@ mod tests {
output_ordinal: 1,
} if binding_id == "fb_01K3TEXT"
));
let dependent_application = FunctionApplication::from_json(
r#"{
"function":{"name":"dependent","version":"fv_dependent"},
"inputs":[
{"parameter":"text","kind":"column","value":{"path":"search_text"}}
],
"output":{"kind":"scalar","arrow_type":"int64","nullable":false}
}"#,
)
.unwrap();
let err = plan_function_application(&reopened, &dependent_application, Some("dependent"))
let err = plan_function_application(&reopened, &named_struct_application("{}"), None)
.unwrap_err();
assert!(
matches!(&err, Error::InvalidInput { message } if message.contains("computed-on-computed"))
);
let plan = plan_function_application(
&reopened,
&named_struct_application(
r#"{"normalized_text":"secondary_text","token_count":"secondary_token_count"}"#,
),
None,
)
.unwrap();
assert_eq!(
plan.outputs
.iter()
.map(|output| output.output_name.as_str())
.collect::<Vec<_>>(),
["secondary_text", "secondary_token_count"]
);
assert!(matches!(err, Error::NotSupported { .. }));
}
#[test]
@@ -3046,151 +2711,6 @@ mod tests {
);
}
#[test]
fn test_blob_function_plans_semantic_input_and_scalar_output() {
let schema = ArrowSchema::new(vec![crate::blob("image", false)]);
let application =
blob_application(r#"{"kind":"scalar","arrow_type":"blob_v2","nullable":false}"#);
let plan = plan_function_application(&schema, &application, Some("thumbnail")).unwrap();
assert_eq!(plan.input_bindings[0].arrow_type, FUNCTION_BLOB_V2_TYPE);
let input_schema =
lance_namespace::schema::convert_json_arrow_schema(&plan.input_schema).unwrap();
assert!(input_schema.field(0).is_blob_v2());
let output_schema =
lance_namespace::schema::convert_json_arrow_schema(&plan.output_schema).unwrap();
assert!(output_schema.field(0).is_blob_v2());
}
#[test]
fn test_blob_scalar_binding_accepts_full_logical_layout() {
let input = crate::blob("image", false);
let application =
blob_application(r#"{"kind":"scalar","arrow_type":"blob_v2","nullable":false}"#);
let plan = plan_function_application(
&ArrowSchema::new(vec![input.clone()]),
&application,
Some("thumbnail"),
)
.unwrap();
let binding = binding_from_plan(&plan);
let mut metadata = full_blob_field("thumbnail", true).metadata().clone();
metadata.extend(function_computed_column_metadata(
binding.binding_id(),
0,
&["image".into()],
));
let output = full_blob_field("thumbnail", true).with_metadata(metadata);
ensure_binding_matches_schema(&ArrowSchema::new(vec![input, output]), &binding).unwrap();
}
#[test]
fn test_blob_binding_rejects_marker_on_invalid_storage_layout() {
let input = crate::blob("image", false);
let application =
blob_application(r#"{"kind":"scalar","arrow_type":"blob_v2","nullable":false}"#);
let plan = plan_function_application(
&ArrowSchema::new(vec![input.clone()]),
&application,
Some("thumbnail"),
)
.unwrap();
let binding = binding_from_plan(&plan);
let malformed = ArrowField::new("thumbnail", DataType::Int64, true)
.with_metadata(crate::blob("thumbnail", true).metadata().clone());
ensure_binding_matches_schema(&ArrowSchema::new(vec![input, malformed]), &binding)
.unwrap_err();
}
#[test]
fn test_blob_input_rejects_marker_on_invalid_storage_layout() {
let malformed = ArrowField::new("image", DataType::Int64, false)
.with_metadata(crate::blob("image", false).metadata().clone());
let application =
blob_application(r#"{"kind":"scalar","arrow_type":"blob_v2","nullable":false}"#);
plan_function_application(
&ArrowSchema::new(vec![malformed]),
&application,
Some("thumbnail"),
)
.unwrap_err();
}
#[test]
fn test_non_blob_input_does_not_require_json_round_trip() {
let json = lance_namespace::schema::arrow_schema_to_json(&ArrowSchema::new(vec![
ArrowField::new("event_time", DataType::Time64(TimeUnit::Microsecond), false),
]))
.unwrap();
assert_eq!(
canonical_input_arrow_type(&json.fields[0]).unwrap(),
"time64"
);
}
#[test]
fn test_blob_named_struct_plans_expanded_and_whole_outputs() {
let schema = ArrowSchema::new(vec![crate::blob("image", false)]);
let application = blob_application(
r#"{"kind":"named_struct","fields":[
{"name":"thumbnail","arrow_type":"blob_v2","nullable":false},
{"name":"width","arrow_type":"int32","nullable":false}
]}"#,
);
let expanded = plan_function_application(&schema, &application, None).unwrap();
let expanded_schema =
lance_namespace::schema::convert_json_arrow_schema(&expanded.output_schema).unwrap();
assert!(expanded_schema.field(0).is_blob_v2());
assert_eq!(expanded_schema.field(1).data_type(), &DataType::Int32);
let whole = plan_function_application(&schema, &application, Some("analysis")).unwrap();
let whole_schema =
lance_namespace::schema::convert_json_arrow_schema(&whole.output_schema).unwrap();
let DataType::Struct(fields) = whole_schema.field(0).data_type() else {
panic!("whole Function output should be a struct");
};
assert!(fields[0].is_blob_v2());
assert_eq!(fields[1].data_type(), &DataType::Int32);
}
#[test]
fn test_blob_whole_struct_binding_accepts_full_logical_layout() {
let input = crate::blob("image", false);
let application = blob_application(
r#"{"kind":"named_struct","fields":[
{"name":"thumbnail","arrow_type":"blob_v2","nullable":false},
{"name":"width","arrow_type":"int32","nullable":false}
]}"#,
);
let plan = plan_function_application(
&ArrowSchema::new(vec![input.clone()]),
&application,
Some("analysis"),
)
.unwrap();
let binding = binding_from_plan(&plan);
let output = ArrowField::new(
"analysis",
DataType::Struct(Fields::from(vec![
full_blob_field("thumbnail", false),
ArrowField::new("width", DataType::Int32, false),
])),
true,
)
.with_metadata(function_computed_column_metadata(
binding.binding_id(),
0,
&["image".into()],
));
ensure_binding_matches_schema(&ArrowSchema::new(vec![input, output]), &binding).unwrap();
}
#[test]
fn test_function_mapping_and_sibling_collisions_fail_before_request() {
let unknown = named_struct_application(r#"{"missing":"renamed"}"#);
@@ -3297,38 +2817,5 @@ mod tests {
assert!(
matches!(&err, Error::InvalidInput { message } if message.contains("computed-on-computed"))
);
let nested_title = ArrowField::new(
"title",
DataType::Struct(vec![ArrowField::new("value", DataType::Utf8, true)].into()),
true,
)
.with_metadata(HashMap::from([
(COMPUTED_COLUMN_META_KEY.to_string(), "true".to_string()),
(KIND_META_KEY.to_string(), SQL_KIND.to_string()),
(
EXPRESSION_META_KEY.to_string(),
"struct('value')".to_string(),
),
]));
let nested_schema = ArrowSchema::new(vec![nested_title, schema.field(1).as_ref().clone()]);
let nested_application = FunctionApplication::from_json(
r#"{
"function":{"name":"text_features","version":"fv_exact"},
"inputs":[
{"parameter":"title","kind":"column","value":{"path":"title.value"}},
{"parameter":"body","kind":"column","value":{"path":"body"}}
],
"output":{"kind":"named_struct","fields":[
{"name":"normalized_text","arrow_type":"utf8","nullable":false},
{"name":"token_count","arrow_type":"int64","nullable":false}
]}
}"#,
)
.unwrap();
let err = plan_function_application(&nested_schema, &nested_application, None).unwrap_err();
assert!(
matches!(&err, Error::InvalidInput { message } if message.contains("computed-on-computed"))
);
}
}
+1 -1
View File
@@ -300,7 +300,7 @@ async fn build_read_context(
for shard_id in shard_ids {
let manifest_store =
ShardManifestStore::new(store.clone(), &base_path, shard_id, scan_batch_size);
if let Some(manifest) = manifest_store.read_latest().await? {
if let Some(manifest) = manifest_store.latest().await? {
snapshots.push(snapshot_from_manifest(shard_id, &manifest, &exclude));
}
}
@@ -78,12 +78,6 @@
"type": "utf8"
}
},
{
"arrow_type": "large_utf8",
"json": {
"type": "large_utf8"
}
},
{
"arrow_type": "binary",
"json": {
@@ -177,21 +171,6 @@
]
}
},
{
"arrow_type": "list<large_utf8>",
"json": {
"type": "list",
"fields": [
{
"name": "item",
"nullable": false,
"type": {
"type": "large_utf8"
}
}
]
}
},
{
"arrow_type": "large_list<utf8>",
"json": {
@@ -207,21 +186,6 @@
]
}
},
{
"arrow_type": "large_list<large_utf8>",
"json": {
"type": "large_list",
"fields": [
{
"name": "item",
"nullable": false,
"type": {
"type": "large_utf8"
}
}
]
}
},
{
"arrow_type": "fixed_size_list<float32, 384>",
"json": {
@@ -366,4 +330,4 @@
"timestamp[us]",
"struct<a: int32>"
]
}
}