Compare commits

..

4 Commits

Author SHA1 Message Date
Lance Release 1b0fc2c465 Bump version: 0.38.0-beta.13 → 0.38.0-beta.14 2026-08-30 15:16:33 +00:00
Xuanwo a417e46bfa feat(functions): support GPU resource requirements (#4085)
Functions can describe their Python environment today, but cannot
declare accelerator requirements. That prevents Sophon from scheduling
computed-column UDF refreshes onto GPU workers from the immutable
Function definition.

Add `num_gpus` to Python `@udf` through a typed
`FunctionResourceRequirements` value and represent resource-aware
definitions with the `python_v2` runtime discriminator. CPU Functions
retain their existing `python` encoding and canonical identity.

The new discriminator is intentional for mixed-version safety:
deployments that do not understand execution resources reject the
runtime instead of accepting a new field and silently running the
Function on CPU. Required resources are part of Function version
identity; priority, concurrency, and retry policy remain Job concerns.
The actual resource scheduling remains owned by Sophon.
2026-08-30 08:10:08 -07:00
Jack Ye fcdc3f949e fix: allow multiple function bindings per table (#4090)
Allow a remote Function declaration when the table already contains
valid, supported Function binding metadata. Existing bindings remain
fully validated, including fail-closed handling for newer or
inconsistent contracts, while other schema mutations retain their
existing no-binding guard. Add planner and remote request-path
regression coverage for a second binding and reject dependent Function
inputs, including nested paths.
2026-08-30 01:16:57 -07:00
Lance Release 0c4e0667bc Bump version: 0.38.0-beta.12 → 0.38.0-beta.13 2026-08-30 06:09:21 +00:00
21 changed files with 580 additions and 63 deletions
+1 -1
View File
@@ -1,5 +1,5 @@
[tool.bumpversion]
current_version = "0.38.0-beta.13"
current_version = "0.38.0-beta.14"
parse = """(?x)
(?P<major>0|[1-9]\\d*)\\.
(?P<minor>0|[1-9]\\d*)\\.
Generated
+3 -3
View File
@@ -5402,7 +5402,7 @@ dependencies = [
[[package]]
name = "lancedb"
version = "0.38.0-beta.12"
version = "0.38.0-beta.14"
dependencies = [
"ahash",
"anyhow",
@@ -5490,7 +5490,7 @@ dependencies = [
[[package]]
name = "lancedb-nodejs"
version = "0.38.0-beta.12"
version = "0.38.0-beta.14"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -5515,7 +5515,7 @@ dependencies = [
[[package]]
name = "lancedb-python"
version = "0.38.0-beta.12"
version = "0.38.0-beta.14"
dependencies = [
"arrow",
"async-trait",
+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-beta.13</version>
<version>0.38.0-beta.14</version>
</dependency>
```
+1 -1
View File
@@ -8,7 +8,7 @@
<parent>
<groupId>com.lancedb</groupId>
<artifactId>lancedb-parent</artifactId>
<version>0.38.0-beta.13</version>
<version>0.38.0-beta.14</version>
<relativePath>../pom.xml</relativePath>
</parent>
+1 -1
View File
@@ -6,7 +6,7 @@
<groupId>com.lancedb</groupId>
<artifactId>lancedb-parent</artifactId>
<version>0.38.0-beta.13</version>
<version>0.38.0-beta.14</version>
<packaging>pom</packaging>
<name>${project.artifactId}</name>
<description>LanceDB Java SDK Parent POM</description>
+1 -1
View File
@@ -1,7 +1,7 @@
[package]
name = "lancedb-nodejs"
edition.workspace = true
version = "0.38.0-beta.13"
version = "0.38.0-beta.14"
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-beta.13",
"version": "0.38.0-beta.14",
"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-beta.13",
"version": "0.38.0-beta.14",
"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-beta.13",
"version": "0.38.0-beta.14",
"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-beta.13",
"version": "0.38.0-beta.14",
"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-beta.13",
"version": "0.38.0-beta.14",
"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-beta.13",
"version": "0.38.0-beta.14",
"os": [
"win32"
],
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-win32-x64-msvc",
"version": "0.38.0-beta.13",
"version": "0.38.0-beta.14",
"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-beta.13",
"version": "0.38.0-beta.14",
"main": "dist/index.js",
"exports": {
".": "./dist/index.js",
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "lancedb-python"
version = "0.38.0-beta.13"
version = "0.38.0-beta.14"
publish = false
edition.workspace = true
description = "Python bindings for LanceDB"
+57 -3
View File
@@ -54,6 +54,18 @@ _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")
@@ -239,6 +251,23 @@ 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):
@@ -247,18 +276,28 @@ 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.
Scheduling resources, priority, concurrency, and retry policy belong to
the submitting Job and are not part of this identity.
The GPU execution requirement is part of this identity. CPU and memory sizing,
priority, concurrency, and retry policy belong to the execution platform.
"""
name: str
@@ -996,6 +1035,7 @@ class UdfDefinition:
pip: tuple[str, ...],
env: Mapping[str, str],
python_version: Optional[str],
gpu: bool = False,
conda: tuple[str, ...] = (),
conda_channels: tuple[str, ...] = (),
):
@@ -1024,12 +1064,14 @@ 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",
kind="python_v2" if gpu_marker is not None else "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(
@@ -1075,6 +1117,7 @@ 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]: ...
@@ -1089,6 +1132,7 @@ 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] = (),
):
@@ -1121,6 +1165,10 @@ 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
@@ -1145,6 +1193,11 @@ 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:
@@ -1156,6 +1209,7 @@ 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,7 +19,7 @@ import pyarrow as pa
import pytest
import lancedb
from lancedb.functions import UdfDefinition, udf
from lancedb.functions import PythonRuntimeSpec, UdfDefinition, udf
THRESHOLD = 20
_CACHE = None
@@ -89,6 +89,58 @@ 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:
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "lancedb"
version = "0.38.0-beta.13"
version = "0.38.0-beta.14"
edition.workspace = true
description = "LanceDB: A serverless, low-latency vector database for AI applications"
license.workspace = true
+150 -27
View File
@@ -207,6 +207,33 @@ 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
@@ -219,22 +246,27 @@ impl PythonRuntimeSpec {
pub fn kind(&self) -> &str {
match self {
Self::Python { .. } => "python",
Self::PythonV2 { .. } => "python_v2",
Self::Unrecognized { kind } => kind,
}
}
/// The Python version for the V1 runtime, or `None` for an unknown kind.
/// The Python version for a known Python runtime, or `None` for an unknown kind.
pub fn python_version(&self) -> Option<&str> {
match self {
Self::Python { python_version, .. } => Some(python_version),
Self::Python { python_version, .. } | Self::PythonV2 { python_version, .. } => {
Some(python_version)
}
Self::Unrecognized { .. } => None,
}
}
/// The Python environment for the V1 runtime, or `None` for an unknown kind.
/// The Python environment for a known Python runtime, or `None` for an unknown kind.
pub fn environment(&self) -> Option<&PythonEnvironmentSpec> {
match self {
Self::Python { environment, .. } => Some(environment),
Self::Python { environment, .. } | Self::PythonV2 { environment, .. } => {
Some(environment)
}
Self::Unrecognized { .. } => None,
}
}
@@ -242,38 +274,73 @@ impl PythonRuntimeSpec {
/// Environment variables, or `None` for an unknown kind.
pub fn env(&self) -> Option<&BTreeMap<String, String>> {
match self {
Self::Python { env, .. } => Some(env),
Self::Python { env, .. } | Self::PythonV2 { 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 PythonRuntimeWire {
kind: String,
#[serde(default)]
python_version: Option<String>,
#[serde(default)]
environment: Option<PythonEnvironmentSpec>,
struct PythonRuntimeV1Wire {
python_version: String,
environment: 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 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 })
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 }),
}
}
}
@@ -287,6 +354,8 @@ 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)]
@@ -304,6 +373,19 @@ 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),
@@ -313,8 +395,8 @@ impl Serialize for PythonRuntimeSpec {
/// Immutable Function version returned by the Enterprise catalog.
///
/// Scheduling resources, priority, concurrency, and retry policy belong to
/// the submitting Job and are not part of this identity.
/// The GPU execution requirement is part of this identity. CPU and memory sizing,
/// priority, concurrency, and retry policy belong to the execution platform.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct FunctionVersion {
name: String,
@@ -589,7 +671,7 @@ impl_json!(RefreshColumnResult);
#[cfg(test)]
mod conda_environment_tests {
use super::PythonEnvironmentSpec;
use super::{PythonEnvironmentSpec, PythonRuntimeSpec};
#[test]
fn conda_channels_round_trip_and_pip_stays_bare() {
@@ -608,4 +690,45 @@ 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"}"#
);
}
}
}
+87
View File
@@ -7464,6 +7464,93 @@ 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| {
+215 -14
View File
@@ -547,7 +547,12 @@ fn ensure_known_binding_shape(value: &Value) -> Result<()> {
Ok(())
}
fn resolve_field_path<'a>(schema: &'a ArrowSchema, path: &str) -> Result<&'a ArrowField> {
struct ResolvedFieldPath<'a> {
root: &'a ArrowField,
leaf: &'a ArrowField,
}
fn resolve_field_path<'a>(schema: &'a ArrowSchema, path: &str) -> Result<ResolvedFieldPath<'a>> {
let parts = lance_core::datatypes::parse_field_path(path).map_err(|e| {
invalid_function(format!("invalid Function input field path '{path}': {e}"))
})?;
@@ -556,22 +561,23 @@ fn resolve_field_path<'a>(schema: &'a ArrowSchema, path: &str) -> Result<&'a Arr
"Function input field path cannot be empty",
));
};
let mut field = schema
let root = 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) = field.data_type() else {
let DataType::Struct(fields) = leaf.data_type() else {
return Err(invalid_function(format!(
"Function input field path '{path}' traverses a non-struct field"
)));
};
field = fields
leaf = fields
.iter()
.find(|field| field.name() == child)
.map(AsRef::as_ref)
.ok_or_else(|| invalid_function(format!("unknown Function input column '{path}'")))?;
}
Ok(field)
Ok(ResolvedFieldPath { root, leaf })
}
fn canonical_input_arrow_type(field: &JsonArrowField) -> Result<String> {
@@ -666,7 +672,8 @@ fn parse_output_arrow_type(raw: &str) -> Result<JsonArrowDataType> {
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 field = resolve_field_path(schema, &input.field_path)?;
let resolved = resolve_field_path(schema, &input.field_path)?;
let field = resolved.leaf;
if field
.metadata()
.get(COMPUTED_COLUMN_META_KEY)
@@ -721,6 +728,11 @@ 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(|_| {
@@ -747,6 +759,28 @@ fn ensure_binding_matches_schema(schema: &ArrowSchema, binding: &FunctionBinding
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()
)));
}
output_fields.push(ArrowField::new(
field.name().clone(),
field.data_type().clone(),
@@ -778,7 +812,7 @@ pub(crate) fn plan_function_application(
application: &FunctionApplication,
output_name: Option<&str>,
) -> Result<FunctionDeclarationPlan> {
ensure_no_function_bindings_for_mutation(schema, "Function binding declaration")?;
ensure_supported_function_metadata(schema)?;
if application.has_unknown_fields() {
return Err(Error::NotSupported {
message: "Function application contains fields from a newer contract".into(),
@@ -828,8 +862,9 @@ pub(crate) fn plan_function_application(
input.parameter
))
})?;
let field = resolve_field_path(schema, path)?;
if field
let resolved = resolve_field_path(schema, path)?;
if resolved
.root
.metadata()
.get(COMPUTED_COLUMN_META_KEY)
.map(String::as_str)
@@ -839,6 +874,7 @@ 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(),
@@ -2579,6 +2615,37 @@ 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!(
@@ -2586,7 +2653,11 @@ mod tests {
))
.unwrap();
ensure_binding_matches_schema(&function_binding_schema(false, false), &binding).unwrap();
ensure_binding_matches_schema(
&valid_function_binding_schema(false, false, &binding),
&binding,
)
.unwrap();
}
#[test]
@@ -2599,8 +2670,11 @@ 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(&function_binding_schema(true, false), &binding)
.unwrap_err();
let err = ensure_binding_matches_schema(
&valid_function_binding_schema(true, false, &binding),
&binding,
)
.unwrap_err();
assert!(
matches!(&err, Error::InvalidInput { message }
if message.contains("input column 'title' is nullable")
@@ -2611,6 +2685,73 @@ 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!(
@@ -2659,9 +2800,36 @@ mod tests {
output_ordinal: 1,
} if binding_id == "fb_01K3TEXT"
));
let err = plan_function_application(&reopened, &named_struct_application("{}"), None)
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"))
.unwrap_err();
assert!(matches!(err, Error::NotSupported { .. }));
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"]
);
}
#[test]
@@ -2817,5 +2985,38 @@ 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"))
);
}
}