diff --git a/python/python/lancedb/expr.py b/python/python/lancedb/expr.py index e8b2d63a4..d16ba95d7 100644 --- a/python/python/lancedb/expr.py +++ b/python/python/lancedb/expr.py @@ -85,8 +85,9 @@ class Expr: # for dict keys / set membership. __hash__ = None # type: ignore[assignment] - def __init__(self, inner: PyExpr) -> None: + def __init__(self, inner: PyExpr, *, column_path: str | None = None) -> None: self._inner = inner + self._column_path = column_path # ── comparisons ────────────────────────────────────────────────────────── @@ -273,7 +274,7 @@ def col(name: str) -> Expr: >>> col("age") > lit(18) Expr((age > 18)) """ - return Expr(expr_col(name)) + return Expr(expr_col(name), column_path=name) def lit(value: Union[bool, int, float, str, bytes, date, datetime, Decimal]) -> Expr: diff --git a/python/python/lancedb/functions.py b/python/python/lancedb/functions.py index 1613e03b4..3b2d05951 100644 --- a/python/python/lancedb/functions.py +++ b/python/python/lancedb/functions.py @@ -20,6 +20,7 @@ import re import sys import textwrap import types +import uuid from collections.abc import Mapping from datetime import date, datetime from typing import ( @@ -265,6 +266,64 @@ class FunctionVersion(_RemoteValue): required_secrets: tuple[str, ...] = () created_at: str + def __call__(self, **inputs: Any) -> FunctionApplication: + """Bind this exact version to named table columns. + + Every input must be a direct [lancedb.col][lancedb.expr.col] + reference. The returned application is immutable and retains a + named-struct output as one sibling group, so every row's sibling values + come from one logical Function evaluation. Map result fields to table + columns with + [FunctionApplication.rename][lancedb.functions.FunctionApplication.rename], + then pass the application to + [Table.add_columns][lancedb.table.Table.add_columns]. + + Examples + -------- + >>> from lancedb import col + >>> application = function( + ... title=col("title"), + ... body=col("body"), + ... ).rename(columns={ + ... "normalized_text": "search_text", + ... "token_count": "search_token_count", + ... }) + >>> table.add_columns(application) # doctest: +SKIP + """ + from lancedb.expr import Expr + + parameters = tuple(parameter.name for parameter in self.signature.inputs) + missing = [parameter for parameter in parameters if parameter not in inputs] + unknown = sorted(set(inputs) - set(parameters)) + if missing or unknown: + details = [] + if missing: + details.append(f"missing inputs: {missing!r}") + if unknown: + details.append(f"unknown inputs: {unknown!r}") + raise TypeError("invalid Function inputs (" + "; ".join(details) + ")") + + bindings = [] + for parameter in parameters: + value = inputs[parameter] + if not isinstance(value, Expr) or value._column_path is None: + raise TypeError( + f"Function input {parameter!r} must be a direct col(...) reference" + ) + bindings.append( + ApplicationInput( + parameter=parameter, + kind="column", + value={"path": value._column_path}, + ) + ) + return FunctionApplication( + function=FunctionVersionRef(name=self.name, version=self.version), + inputs=tuple(bindings), + output=self.signature.output, + group_id=f"fg_{uuid.uuid4().hex}", + ) + class FunctionRegistrationRequest(_RemoteValue): """Stable remote registration envelope produced by :func:`udf`. @@ -304,7 +363,14 @@ class ApplicationInput(_OpenRemoteValue): class FunctionApplication(_OpenRemoteValue): - """Immutable pre-declaration application of an exact Function version.""" + """Immutable pre-declaration application of an exact Function version. + + A named-struct output remains one grouped application through table + declaration and execution. + [FunctionApplication.rename][lancedb.functions.FunctionApplication.rename] + records the result-field to table-column mapping without splitting sibling + outputs into separate UDF calls. + """ function: FunctionVersionRef inputs: tuple[ApplicationInput, ...] diff --git a/python/python/tests/test_first_class_function_slice1.py b/python/python/tests/test_first_class_function_slice1.py index fead28bc8..1bb8feaf4 100644 --- a/python/python/tests/test_first_class_function_slice1.py +++ b/python/python/tests/test_first_class_function_slice1.py @@ -6,6 +6,7 @@ from pathlib import Path import pytest +from lancedb import col import lancedb.functions as functions from lancedb.functions import ( FunctionApplication, @@ -120,6 +121,83 @@ def test_function_version_identity_is_immutable_and_exact(): assert FunctionVersion(**changed) != version +def test_function_version_binds_named_columns_as_one_immutable_group(): + version = FunctionVersion.from_json( + json.dumps(job_result("remote_function_job.json")) + ) + + application = version(text=col("documents.body")) + + assert application.function.name == version.name + assert application.function.version == version.version + assert application.output is version.signature.output + assert application.group_id.startswith("fg_") + assert [ + (value.parameter, value.kind, value.value["path"]) + for value in application.inputs + ] == [("text", "column", "documents.body")] + with pytest.raises((TypeError, ValueError)): + application.group_id = "fg_changed" + + +def test_function_version_binding_validates_names_and_direct_columns(): + version = FunctionVersion.from_json( + json.dumps(job_result("remote_function_job.json")) + ) + + with pytest.raises(TypeError, match=r"missing inputs: \['text'\]"): + version() + with pytest.raises(TypeError, match=r"unknown inputs: \['body'\]"): + version(text=col("text"), body=col("body")) + with pytest.raises(TypeError, match="direct col"): + version(text=col("text").lower()) + + +def test_function_version_keeps_named_struct_outputs_in_one_application(): + value = job_result("remote_function_job.json") + value["name"] = "text_features" + value["version"] = "fv_grouped" + value["signature"] = { + "inputs": [ + {"name": "title", "arrow_type": "utf8", "nullable": True}, + {"name": "body", "arrow_type": "utf8", "nullable": True}, + ], + "output": { + "kind": "named_struct", + "fields": [ + { + "name": "normalized_text", + "arrow_type": "utf8", + "nullable": False, + }, + { + "name": "token_count", + "arrow_type": "int64", + "nullable": False, + }, + ], + }, + } + version = FunctionVersion(**value) + + application = version(body=col("body"), title=col("title")).rename( + columns={ + "normalized_text": "search_text", + "token_count": "search_token_count", + } + ) + + assert [value.parameter for value in application.inputs] == ["title", "body"] + assert [field.name for field in application.output.fields] == [ + "normalized_text", + "token_count", + ] + assert dict(application.columns) == { + "normalized_text": "search_text", + "token_count": "search_token_count", + } + + def test_unknown_fields_and_discriminators_are_forward_decodable(): value = job_result("remote_function_job.json") value["future_version_metadata"] = {"retention_class": "catalog"}