feat(python): bind function versions to columns (#4012)

A registered `FunctionVersion` has an exact identity and grouped output
contract, but the Python SDK cannot currently bind it to table columns
without manually constructing wire models.

Calling a `FunctionVersion` with named `col(...)` references now returns
one immutable `FunctionApplication` pinned to that exact version. The
application preserves named-struct outputs as one sibling group, while
`rename(columns=...)` defines the result-field to table-column mapping
consumed by `Table.add_columns`. Derived expressions and incomplete or
unknown input names fail before declaration.

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:

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, ...]

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"}