rust/lancedb
1
0
Fork 0
mirror of https://github.com/lancedb/lancedb.git synced 2026-08-22 05:58:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add grouped function column bindings (#3994)

Browse Source 
Function applications from the canonical remote contract cannot
currently declare scalar or grouped computed-column outputs atomically.

This adds the remote-only declaration contract for scalar,
struct-as-one-column, and expanded named-struct outputs. It validates
result mappings, fixes exact input/output Arrow schemas in the request,
persists grouped sibling metadata, and keeps local Function execution
unsupported. Unknown newer application or binding metadata remains
readable, while schema-changing mutations fail closed instead of
rewriting it.

Stable Lance field IDs are deliberately not a declaration prerequisite
in this slice. Inputs bind by parameter name and field path; Sophon
remains responsible for exact-version validation, atomic all-NULL
sibling creation, binding identity and revision allocation, and
persisted output identities.
This commit is contained in:
Xuanwo
2026-08-21 17:01:04 +08:00
committed by GitHub
parent 4ba2421254
commit 685cb01d6d
20 changed files with 1799 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/python/lancedb/_lancedb.pyi
+3
View File
@@ -341,6 +341,9 @@ class Table:
async def add_computed_columns(
self, columns: list[tuple[str, str]]
) -> AddColumnsResult: ...
async def add_function_columns(
self, application_json: str, output_name: Optional[str]
) -> AddColumnsResult: ...
async def refresh_column(self, column: str) -> RefreshColumnResult: ...
async def refresh_column_async(self, column: str) -> Job: ...
async def add_columns_with_schema(self, schema: pa.Schema) -> AddColumnsResult: ...
python/python/lancedb/functions.py
+51 -5
View File
@@ -123,6 +123,23 @@ class _RemoteValue(BaseModel):
)
class _OpenRemoteValue(_RemoteValue):
"""Forward-readable value whose extras stay out of canonical encoding."""
if _PYDANTIC_V2:
model_config = {"extra": "allow", "frozen": True}
else:
class Config:
allow_mutation = False
extra = "allow"
def _unknown_field_names(self) -> set[str]:
if _PYDANTIC_V2:
return set((self.__pydantic_extra__ or {}).keys())
return set(self.__dict__) - set(self.__fields__)
class FunctionArtifact(_RemoteValue):
"""Content-addressed Python artifact identity."""
@@ -137,13 +154,13 @@ class FunctionParameter(_RemoteValue):
nullable: bool
class FunctionResultField(_RemoteValue):
class FunctionResultField(_OpenRemoteValue):
name: str
arrow_type: str
nullable: bool
class FunctionOutput(_RemoteValue):
class FunctionOutput(_OpenRemoteValue):
"""Scalar or ordered named-struct output; unknown kinds remain decodable."""
kind: str
@@ -211,12 +228,12 @@ class FunctionVersion(_RemoteValue):
created_at: str
class FunctionVersionRef(_RemoteValue):
class FunctionVersionRef(_OpenRemoteValue):
name: str
version: str
class ApplicationInput(_RemoteValue):
class ApplicationInput(_OpenRemoteValue):
"""One parameter value.
Slice 1 freezes integers, strings, booleans, nulls, arrays, and objects.
@@ -234,7 +251,7 @@ class ApplicationInput(_RemoteValue):
return _validate_literal(value)
class FunctionApplication(_RemoteValue):
class FunctionApplication(_OpenRemoteValue):
"""Immutable pre-declaration application of an exact Function version."""
function: FunctionVersionRef
@@ -243,6 +260,33 @@ class FunctionApplication(_RemoteValue):
group_id: str
columns: Mapping[str, str] = Field(default_factory=dict)
def _known_dict(self) -> dict[str, Any]:
value = super()._known_dict()
for name in self._unknown_field_names():
value.pop(name, None)
return value
def _ensure_declarable(self) -> None:
unknown = {f"application.{name}" for name in self._unknown_field_names()}
unknown.update(
f"function.{name}" for name in self.function._unknown_field_names()
)
for index, input_value in enumerate(self.inputs):
unknown.update(
f"inputs[{index}].{name}" for name in input_value._unknown_field_names()
)
unknown.update(f"output.{name}" for name in self.output._unknown_field_names())
for index, field in enumerate(self.output.fields):
unknown.update(
f"output.fields[{index}].{name}"
for name in field._unknown_field_names()
)
if unknown:
raise ValueError(
"Function application contains fields from a newer contract: "
f"{sorted(unknown)!r}"
)
def rename(self, *, columns: Mapping[str, str]) -> FunctionApplication:
"""Return a copy with result-field to table-column aliases."""
if self.output.kind != "named_struct":
@@ -293,6 +337,8 @@ class FunctionBinding(_RemoteValue):
group_id: str
inputs: tuple[InputBinding, ...]
outputs: tuple[OutputMapping, ...]
input_schema: Optional[Mapping[str, Any]] = None
output_schema: Optional[Mapping[str, Any]] = None
class RefreshColumnResult(_RemoteValue):
python/python/lancedb/remote/table.py
+4 -1
View File
@@ -49,6 +49,7 @@ from lancedb.index import (
LabelList,
)
from lancedb.job import Job
from lancedb.functions import FunctionApplication
from lancedb.remote.db import LOOP
from lancedb.table import IndexConfigType, KNOWN_METRICS
import pyarrow as pa
@@ -960,7 +961,9 @@ class RemoteTable(Table):
def add_columns(
self,
transforms: Dict[str, str] | None = None,
transforms: Dict[str, str | FunctionApplication]
| FunctionApplication
| None = None,
*,
computed: Dict[str, str] | None = None,
) -> AddColumnsResult:
python/python/lancedb/table.py
+54 -9
View File
@@ -72,6 +72,7 @@ from .index import (
FTS,
)
from .expr import Expr
from .functions import FunctionApplication
from .merge import LanceMergeInsertBuilder
from .pydantic import LanceModel, model_to_dict
from .query import (
@@ -1942,7 +1943,8 @@ class Table(ABC):
@abstractmethod
def add_columns(
self,
transforms: Dict[str, str]
transforms: Dict[str, str | FunctionApplication]
| FunctionApplication
| pa.Field
| List[pa.Field]
| pa.Schema
@@ -1955,13 +1957,21 @@ class Table(ABC):
Parameters
----------
transforms: Dict[str, str], pa.Field, List[pa.Field], pa.Schema
transforms: Dict[str, str | FunctionApplication], FunctionApplication,
pa.Field, List[pa.Field], pa.Schema
A map of column name to a SQL expression to use to calculate the
value of the new column. These expressions will be evaluated for
each row in the table, and can reference existing columns.
Alternatively, a pyarrow Field or Schema can be provided to add
new columns with the specified data types. The new columns will
be initialized with null values.
A mapping with one ``FunctionApplication`` value keeps its scalar
or named-struct result in the named table column. A bare
named-struct application expands its ordered result fields as one
atomic sibling group; aliases come from ``rename(columns=...)``.
Function columns are supported only on LanceDB Cloud and
Enterprise.
computed: Dict[str, str], optional
A map of column name to a SQL expression defining the column. The
column's type and inputs are derived from the expression, so no
@@ -4056,9 +4066,10 @@ class LanceTable(Table):
def add_columns(
self,
transforms: Dict[str, str]
| pa.field
| List[pa.field]
transforms: Dict[str, str | FunctionApplication]
| FunctionApplication
| pa.Field
| List[pa.Field]
| pa.Schema
| None = None,
*,
@@ -5992,9 +6003,10 @@ class AsyncTable:
async def add_columns(
self,
transforms: dict[str, str]
| pa.field
| List[pa.field]
transforms: dict[str, str | FunctionApplication]
| FunctionApplication
| pa.Field
| List[pa.Field]
| pa.Schema
| None = None,
*,
@@ -6005,12 +6017,19 @@ class AsyncTable:
Parameters
----------
transforms: Dict[str, str]
transforms: Dict[str, str | FunctionApplication] or FunctionApplication
A map of column name to a SQL expression to use to calculate the
value of the new column. These expressions will be evaluated for
each row in the table, and can reference existing columns.
Alternatively, you can pass a pyarrow field or schema to add
new columns with NULLs.
A mapping with one ``FunctionApplication`` value keeps its scalar
or named-struct result in the named table column. A bare
named-struct application expands its ordered result fields as one
atomic sibling group; aliases come from ``rename(columns=...)``.
Function columns are supported only on LanceDB Cloud and
Enterprise.
computed: Dict[str, str], optional
A map of column name to a SQL expression defining the column. The
column's type and inputs are derived from the expression.
@@ -6034,6 +6053,32 @@ class AsyncTable:
version: the new version number of the table after adding columns.
"""
function_application = None
function_output_name = None
if isinstance(transforms, FunctionApplication):
function_application = transforms
elif isinstance(transforms, dict) and any(
isinstance(value, FunctionApplication) for value in transforms.values()
):
if len(transforms) != 1 or not all(
isinstance(value, FunctionApplication) for value in transforms.values()
):
raise ValueError(
"one add_columns call declares exactly one Function sibling group"
)
function_output_name, function_application = next(iter(transforms.items()))
if function_application is not None:
if computed:
raise ValueError(
"add_columns cannot mix a Function application with SQL "
"computed columns"
)
function_application._ensure_declarable()
return await self._inner.add_function_columns(
function_application.to_canonical_json(), function_output_name
)
if isinstance(transforms, pa.Field):
transforms = [transforms]
if isinstance(transforms, list) and all(
python/python/tests/test_first_class_function_slice1.py
+86
View File
@@ -14,6 +14,7 @@ from lancedb.functions import (
PythonRuntimeSpec,
RefreshColumnResult,
)
from lancedb.table import AsyncTable
FIXTURES = (
@@ -166,6 +167,8 @@ def test_binding_and_refresh_result_keep_stable_remote_fields():
assert binding.revision == 3
assert binding.function.version == "fv_01K3TEXT"
assert [output.output_ordinal for output in binding.outputs] == [0, 1]
assert binding.input_schema is not None
assert binding.output_schema is not None
result = RefreshColumnResult.from_json(
json.dumps(job_result("remote_refresh_job.json"))
@@ -223,3 +226,86 @@ def test_canonical_client_values_contain_secret_names_only():
canonical = json.loads(version.to_canonical_json())
assert canonical["required_secrets"] == ["HF_TOKEN"]
assert_no_secret_values(canonical)
class _FunctionDeclarationInner:
def __init__(self):
self.calls = []
async def add_function_columns(self, application_json, output_name):
self.calls.append((json.loads(application_json), output_name))
return "declared"
def known_application() -> FunctionApplication:
value = json.loads(fixture("remote_function_application.json"))
value.pop("future_application")
return FunctionApplication(**value)
@pytest.mark.asyncio
async def test_add_columns_routes_struct_as_one_and_grouped_expansion_atomically():
inner = _FunctionDeclarationInner()
table = AsyncTable(inner)
application = known_application()
result = await table.add_columns(
{"features": application._copy(update={"columns": {}})}
)
assert result == "declared"
assert inner.calls[-1][1] == "features"
bare = application._copy(update={"columns": {}}).rename(
columns={"normalized_text": "search_text"}
)
result = await table.add_columns(bare)
assert result == "declared"
assert inner.calls[-1][1] is None
assert inner.calls[-1][0]["columns"] == {"normalized_text": "search_text"}
@pytest.mark.asyncio
async def test_add_columns_rejects_mixed_groups_and_unknown_newer_application():
inner = _FunctionDeclarationInner()
table = AsyncTable(inner)
application = known_application()
with pytest.raises(ValueError, match="exactly one Function sibling group"):
await table.add_columns({"a": application, "b": application})
future = json.loads(fixture("remote_function_application.json"))
application = FunctionApplication(**future)
with pytest.raises(ValueError, match="newer contract"):
await table.add_columns(application)
future.pop("future_application")
future["output"]["assignment"] = "cell_flag"
application = FunctionApplication(**future)
assert "assignment" not in json.loads(application.to_canonical_json())["output"]
with pytest.raises(ValueError, match="output.assignment"):
await table.add_columns(application)
assert inner.calls == []
def test_rename_requires_named_struct_and_keeps_partial_mapping_immutable():
scalar = FunctionApplication.from_json(
json.dumps(
{
"function": {"name": "embed", "version": "fv_exact"},
"inputs": [],
"output": {
"kind": "scalar",
"arrow_type": "list<float32>",
"nullable": False,
},
"group_id": "fg_scalar",
}
)
)
with pytest.raises(ValueError, match="named-struct"):
scalar.rename(columns={"value": "embedding"})
application = known_application()._copy(update={"columns": {}})
renamed = application.rename(columns={"normalized_text": "search_text"})
assert dict(application.columns) == {}
assert dict(renamed.columns) == {"normalized_text": "search_text"}