mirror of
https://github.com/lancedb/lancedb.git
synced 2026-10-13 23:32:27 +00:00
docs: apply a registered Function version in the Function guide
The guide applied the local decorated callable to `table["value"]`, which
raises TypeError: a table is not subscriptable, and add_columns takes an
application of a registered version, not the callable's result.
The guide now walks one example from definition to filled column: define
`double`, call it locally, connect to Enterprise, create a table whose
`value` column matches the parameter's type and nullability, register the
function, apply `version(value=col("value"))`, refresh, wait, and read the
values back. Its snippets come from docs/web-tests/py/test_function.py.
The local tests in that file check what `@udf` sends for registration, how
a registered version binds a column, and that a local connection refuses
both registration and Function columns. The Enterprise walkthrough runs
only when LANCEDB_URI, LANCEDB_API_KEY and LANCEDB_HOST_OVERRIDE are set.
The page also stops promising that a retry after a failed refresh costs
only the unfinished rows.
This commit is contained in:
@@ -0,0 +1,193 @@
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
# SPDX-FileCopyrightText: Copyright The LanceDB Authors
|
||||
|
||||
"""Examples for docs/web/management/function.mdx.
|
||||
|
||||
Registering a Function and filling its column need a LanceDB Enterprise
|
||||
deployment, so `test_function_enterprise` runs only when the connection
|
||||
variables the Enterprise examples already use are set: LANCEDB_URI,
|
||||
LANCEDB_API_KEY and LANCEDB_HOST_OVERRIDE, plus LANCEDB_REGION if the
|
||||
deployment needs one. It creates the table `function_demo` and a version of
|
||||
the Function `double` in that database, and drops both when it finishes.
|
||||
|
||||
The other tests run locally and check the client side of the same code: what
|
||||
`@udf` sends for registration, how a registered version binds a column, and
|
||||
where a local connection stops. They do not run a Function remotely.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import lancedb
|
||||
import pytest
|
||||
from lancedb import col
|
||||
from lancedb.functions import FunctionApplication, FunctionVersion
|
||||
|
||||
# --8<-- [start:function_define]
|
||||
from lancedb import udf
|
||||
|
||||
|
||||
@udf
|
||||
def double(value: float) -> float:
|
||||
return value * 2
|
||||
|
||||
|
||||
# --8<-- [end:function_define]
|
||||
|
||||
# A Function version as the service returns it after registration: the SDK's
|
||||
# shared golden reply, with the name and signature of the definition above.
|
||||
SERVICE_REPLY = (
|
||||
Path(__file__).parents[3]
|
||||
/ "rust"
|
||||
/ "lancedb"
|
||||
/ "tests"
|
||||
/ "fixtures"
|
||||
/ "first_class_functions"
|
||||
/ "v1"
|
||||
/ "remote_function_job.json"
|
||||
)
|
||||
|
||||
|
||||
def registered_version(definition) -> FunctionVersion:
|
||||
request = definition.registration_request
|
||||
reply = json.loads(SERVICE_REPLY.read_text())["result"]
|
||||
reply["name"] = request.name
|
||||
reply["signature"] = json.loads(request.signature.to_canonical_json())
|
||||
return FunctionVersion.from_json(json.dumps(reply))
|
||||
|
||||
|
||||
def require_env(var_name: str) -> str:
|
||||
"""Skip the test unless the required environment variable is provided."""
|
||||
value = os.environ.get(var_name)
|
||||
if not value:
|
||||
pytest.skip(f"Set {var_name} to run this Enterprise example")
|
||||
return value
|
||||
|
||||
|
||||
def test_function_runs_locally_as_python():
|
||||
# --8<-- [start:function_local_call]
|
||||
print(double(1.5))
|
||||
# 3.0
|
||||
# --8<-- [end:function_local_call]
|
||||
assert double(1.5) == 3.0
|
||||
|
||||
|
||||
def test_function_registration_request():
|
||||
request = double.registration_request
|
||||
assert request.name == "double"
|
||||
assert [
|
||||
(parameter.name, parameter.arrow_type, parameter.nullable)
|
||||
for parameter in request.signature.inputs
|
||||
] == [("value", "float64", False)]
|
||||
output = request.signature.output
|
||||
assert (output.kind, output.arrow_type, output.nullable) == (
|
||||
"scalar",
|
||||
"float64",
|
||||
False,
|
||||
)
|
||||
version = request.runtime.python_version
|
||||
assert version == f"{sys.version_info.major}.{sys.version_info.minor}"
|
||||
assert request.runtime.environment.packages == ()
|
||||
|
||||
|
||||
def test_function_registration_is_remote(tmp_path):
|
||||
db = lancedb.connect(tmp_path)
|
||||
with pytest.raises(NotImplementedError):
|
||||
db.create_function(double)
|
||||
|
||||
|
||||
def test_function_version_binds_a_column(tmp_path):
|
||||
version = registered_version(double)
|
||||
|
||||
application = version(value=col("value"))
|
||||
|
||||
assert isinstance(application, FunctionApplication)
|
||||
assert application.function.name == "double"
|
||||
assert application.function.version == version.version
|
||||
assert [
|
||||
(binding.parameter, binding.kind, binding.value["path"])
|
||||
for binding in application.inputs
|
||||
] == [("value", "column", "value")]
|
||||
with pytest.raises(TypeError, match="direct col"):
|
||||
version(value=1.5)
|
||||
with pytest.raises(TypeError, match="missing"):
|
||||
version(x=col("value"))
|
||||
|
||||
# The binding is accepted by add_columns, which a local table refuses:
|
||||
# a Function column needs LanceDB Cloud or Enterprise.
|
||||
table = lancedb.connect(tmp_path).create_table("t", [{"value": 1.0}])
|
||||
with pytest.raises(NotImplementedError, match="Cloud and Enterprise"):
|
||||
table.add_columns({"doubled": application})
|
||||
|
||||
|
||||
def test_function_enterprise():
|
||||
require_env("LANCEDB_URI")
|
||||
require_env("LANCEDB_API_KEY")
|
||||
require_env("LANCEDB_HOST_OVERRIDE")
|
||||
|
||||
# --8<-- [start:function_connect]
|
||||
import os
|
||||
|
||||
import lancedb
|
||||
|
||||
db = lancedb.connect(
|
||||
uri=os.environ["LANCEDB_URI"], # db://your-database
|
||||
api_key=os.environ["LANCEDB_API_KEY"],
|
||||
region=os.environ.get("LANCEDB_REGION", "us-east-1"),
|
||||
host_override=os.environ["LANCEDB_HOST_OVERRIDE"],
|
||||
)
|
||||
# --8<-- [end:function_connect]
|
||||
|
||||
table = None
|
||||
version = None
|
||||
try:
|
||||
# --8<-- [start:function_create_table]
|
||||
import pyarrow as pa
|
||||
|
||||
schema = pa.schema([pa.field("value", pa.float64(), nullable=False)])
|
||||
table = db.create_table(
|
||||
"function_demo",
|
||||
[{"value": 1.0}, {"value": 2.0}, {"value": 3.0}],
|
||||
schema=schema,
|
||||
mode="overwrite",
|
||||
)
|
||||
# --8<-- [end:function_create_table]
|
||||
|
||||
# --8<-- [start:function_register]
|
||||
job = db.create_function_async(double)
|
||||
version = job.wait()
|
||||
# --8<-- [end:function_register]
|
||||
assert version.name == "double"
|
||||
|
||||
# --8<-- [start:function_apply]
|
||||
from lancedb import col
|
||||
|
||||
table.add_columns({"doubled": version(value=col("value"))})
|
||||
# --8<-- [end:function_apply]
|
||||
|
||||
# --8<-- [start:function_refresh]
|
||||
refresh = table.refresh_column_async("doubled")
|
||||
result = refresh.wait()
|
||||
print(result.rows_assigned)
|
||||
# 3
|
||||
# --8<-- [end:function_refresh]
|
||||
assert result.rows_assigned == 3
|
||||
assert result.rows_failed == 0
|
||||
|
||||
# --8<-- [start:function_inspect]
|
||||
rows = table.search().select(["value", "doubled"]).to_list()
|
||||
print(sorted((row["value"], row["doubled"]) for row in rows))
|
||||
# [(1.0, 2.0), (2.0, 4.0), (3.0, 6.0)]
|
||||
# --8<-- [end:function_inspect]
|
||||
assert sorted((row["value"], row["doubled"]) for row in rows) == [
|
||||
(1.0, 2.0),
|
||||
(2.0, 4.0),
|
||||
(3.0, 6.0),
|
||||
]
|
||||
finally:
|
||||
if table is not None:
|
||||
db.drop_table("function_demo")
|
||||
if version is not None:
|
||||
db.drop_function(version.name, version=version.version)
|
||||
@@ -6,6 +6,17 @@ icon: code
|
||||
keywords: ["udf", "functions", "create_function", "embedding", "feature engineering"]
|
||||
---
|
||||
|
||||
import {
|
||||
PyFunctionDefine as FunctionDefine,
|
||||
PyFunctionLocalCall as FunctionLocalCall,
|
||||
PyFunctionConnect as FunctionConnect,
|
||||
PyFunctionCreateTable as FunctionCreateTable,
|
||||
PyFunctionRegister as FunctionRegister,
|
||||
PyFunctionApply as FunctionApply,
|
||||
PyFunctionRefresh as FunctionRefresh,
|
||||
PyFunctionInspect as FunctionInspect,
|
||||
} from '/snippets/function.mdx';
|
||||
|
||||
<Badge color="red">Enterprise</Badge>
|
||||
|
||||
Some columns cannot be written as a SQL expression: an embedding, a model call,
|
||||
@@ -13,35 +24,109 @@ an image decode. A function is a Python callable registered with the database,
|
||||
versioned, and then applied as a column — the code runs where the data is
|
||||
rather than pulling rows to your process.
|
||||
|
||||
Registration is a remote operation. On a local connection `create_function`
|
||||
raises `NotImplementedError`; use a
|
||||
[computed column](/feature-engineering/refresh-column) instead.
|
||||
Registering and running a function needs a LanceDB Enterprise connection. On a
|
||||
local connection `create_function` raises `NotImplementedError`; to derive a
|
||||
column locally, declare a
|
||||
[SQL computed column](/feature-engineering/add-computed-column) instead.
|
||||
|
||||
<Note>
|
||||
In SQL, a function is registered with [`CREATE FUNCTION`](/sql/ddl/function#create) and listed with [`SHOW FUNCTIONS`](/sql/ddl/function#show).
|
||||
</Note>
|
||||
|
||||
The steps below take a column of numbers to a filled column of their doubles.
|
||||
|
||||
## Define a function {#define}
|
||||
|
||||
`@udf` packages a scalar Python callable. Input and output types are inferred
|
||||
from annotations:
|
||||
from annotations, so `double` takes one non-nullable `float64` parameter,
|
||||
`value`, and returns a non-nullable `float64`:
|
||||
|
||||
```python
|
||||
from lancedb import udf
|
||||
<CodeBlock filename="Python" language="Python" icon="python">
|
||||
{FunctionDefine}
|
||||
</CodeBlock>
|
||||
|
||||
@udf(pip=["numpy==2.2.0"])
|
||||
def score(value: float) -> float:
|
||||
return value * 2
|
||||
```
|
||||
The decorated object stays callable. Calling it runs the original function in
|
||||
your process, without the database, so unit tests stay ordinary tests:
|
||||
|
||||
The decorated object stays callable, so `score(1.5)` runs the original function
|
||||
locally and unit tests stay ordinary tests.
|
||||
<CodeBlock filename="Python" language="Python" icon="python">
|
||||
{FunctionLocalCall}
|
||||
</CodeBlock>
|
||||
|
||||
Where annotations cannot express the Arrow type precisely, pass `input_schema`
|
||||
and `output_schema` together — one without the other is rejected. Outputs must
|
||||
be non-nullable, because a physical NULL is what marks a computed-column row as
|
||||
unfilled.
|
||||
|
||||
## Connect and create a table {#table}
|
||||
|
||||
The remaining steps run against your Enterprise deployment. This connection
|
||||
reads its settings from environment variables, so no credential appears in the
|
||||
code:
|
||||
|
||||
<CodeBlock filename="Python" language="Python" icon="python">
|
||||
{FunctionConnect}
|
||||
</CodeBlock>
|
||||
|
||||
A column binds to a parameter only when their Arrow types and nullability
|
||||
match. `value: float` is non-nullable, so the table declares `value`
|
||||
non-nullable; annotate the parameter `Optional[float]` to bind a nullable
|
||||
column instead.
|
||||
|
||||
<CodeBlock filename="Python" language="Python" icon="python">
|
||||
{FunctionCreateTable}
|
||||
</CodeBlock>
|
||||
|
||||
## Register it {#register}
|
||||
|
||||
Registration returns a job; the immutable version exists once the job succeeds.
|
||||
|
||||
<CodeBlock filename="Python" language="Python" icon="python">
|
||||
{FunctionRegister}
|
||||
</CodeBlock>
|
||||
|
||||
`create_function` is the blocking equivalent. A registered version is fetched
|
||||
with `db.get_function(name, version=...)`.
|
||||
|
||||
<Note>
|
||||
**Enterprise job type:** `create_function`. Registration returns a job because
|
||||
the function is baked into an image before the version becomes usable.
|
||||
</Note>
|
||||
|
||||
## Apply it as a column {#apply}
|
||||
|
||||
Calling the registered version binds each parameter, by name, to a column with
|
||||
`col(...)`. `add_columns` declares the result as a new column:
|
||||
|
||||
<CodeBlock filename="Python" language="Python" icon="python">
|
||||
{FunctionApply}
|
||||
</CodeBlock>
|
||||
|
||||
The column is declared empty; the function has not run yet. A mapping with one
|
||||
function application keeps its scalar or named-struct result in the named
|
||||
column. A bare named-struct application expands its result fields as one atomic
|
||||
binding, with aliases from `rename(columns=...)`.
|
||||
|
||||
## Fill the column {#refresh}
|
||||
|
||||
A refresh runs the function on the cluster as a [job](/basics/jobs). `wait()`
|
||||
blocks until the job finishes and returns its result, including how many rows
|
||||
it filled:
|
||||
|
||||
<CodeBlock filename="Python" language="Python" icon="python">
|
||||
{FunctionRefresh}
|
||||
</CodeBlock>
|
||||
|
||||
The filled values are read like any other column:
|
||||
|
||||
<CodeBlock filename="Python" language="Python" icon="python">
|
||||
{FunctionInspect}
|
||||
</CodeBlock>
|
||||
|
||||
Filling then works as it does for a SQL computed column: a later refresh fills
|
||||
rows added since the last one and recomputes rows whose inputs changed, and
|
||||
leaves filled rows with unchanged inputs alone. See
|
||||
[Refresh a column](/feature-engineering/refresh-column).
|
||||
|
||||
## Declare the environment {#environment}
|
||||
|
||||
The function runs remotely, so its dependencies are part of its definition:
|
||||
@@ -90,9 +175,10 @@ in the function definition: `@udf` takes no error-handling parameter, so a
|
||||
function that calls a flaky external service should handle that itself, inside
|
||||
the function body.
|
||||
|
||||
Letting the job fail is the other option. Since
|
||||
[a refresh never revisits a filled row](/feature-engineering/refresh-column#idempotent),
|
||||
re-running after a partial failure costs only the rows that did not complete.
|
||||
Letting the job fail is the other option: fix the cause and refresh again. The
|
||||
next refresh [skips rows that are filled and current](/feature-engineering/refresh-column#idempotent),
|
||||
but how much of the failed job's work was kept is not guaranteed, so do not
|
||||
count on the retry costing only the rows that did not complete.
|
||||
|
||||
## What gets packaged {#packaging}
|
||||
|
||||
@@ -110,41 +196,6 @@ In practice this means a function should be self-contained: take its inputs as
|
||||
arguments, import what it needs at module level, and avoid capturing state from
|
||||
the surrounding scope.
|
||||
|
||||
<Note>
|
||||
**Enterprise job type:** `create_function`. Registration returns a job because
|
||||
the function is baked into an image before the version becomes usable -- see
|
||||
below.
|
||||
</Note>
|
||||
|
||||
## Register it {#register}
|
||||
|
||||
Registration returns a job; the immutable version exists once the job succeeds.
|
||||
|
||||
```python
|
||||
job = db.create_function_async(score)
|
||||
version = job.wait()
|
||||
```
|
||||
|
||||
`create_function` is the blocking equivalent. A registered version is fetched
|
||||
with `db.get_function(name, version=...)`.
|
||||
|
||||
## Apply it as a column {#apply}
|
||||
|
||||
A registered function is applied through `add_columns`, in the `transforms`
|
||||
mapping:
|
||||
|
||||
```python
|
||||
table.add_columns(transforms={"doubled": score(table["value"])})
|
||||
```
|
||||
|
||||
A mapping with one function application keeps its scalar or named-struct result
|
||||
in the named column. A bare named-struct application expands its result fields
|
||||
as one atomic binding, with aliases from `rename(columns=...)`.
|
||||
|
||||
Filling then works exactly as it does for any computed column — see
|
||||
[Computed columns](/feature-engineering/refresh-column) for refresh
|
||||
semantics, including the rule that filled rows are never revisited.
|
||||
|
||||
## What to read next {#next}
|
||||
|
||||
<CardGroup cols={2}>
|
||||
|
||||
@@ -0,0 +1,18 @@
|
||||
{/* Auto-generated by scripts/mdx_snippets_gen.py. Do not edit manually. */}
|
||||
|
||||
export const PyFunctionApply = "from lancedb import col\n\ntable.add_columns({\"doubled\": version(value=col(\"value\"))})\n";
|
||||
|
||||
export const PyFunctionConnect = "import os\n\nimport lancedb\n\ndb = lancedb.connect(\n uri=os.environ[\"LANCEDB_URI\"], # db://your-database\n api_key=os.environ[\"LANCEDB_API_KEY\"],\n region=os.environ.get(\"LANCEDB_REGION\", \"us-east-1\"),\n host_override=os.environ[\"LANCEDB_HOST_OVERRIDE\"],\n)\n";
|
||||
|
||||
export const PyFunctionCreateTable = "import pyarrow as pa\n\nschema = pa.schema([pa.field(\"value\", pa.float64(), nullable=False)])\ntable = db.create_table(\n \"function_demo\",\n [{\"value\": 1.0}, {\"value\": 2.0}, {\"value\": 3.0}],\n schema=schema,\n mode=\"overwrite\",\n)\n";
|
||||
|
||||
export const PyFunctionDefine = "from lancedb import udf\n\n\n@udf\ndef double(value: float) -> float:\n return value * 2\n";
|
||||
|
||||
export const PyFunctionInspect = "rows = table.search().select([\"value\", \"doubled\"]).to_list()\nprint(sorted((row[\"value\"], row[\"doubled\"]) for row in rows))\n# [(1.0, 2.0), (2.0, 4.0), (3.0, 6.0)]\n";
|
||||
|
||||
export const PyFunctionLocalCall = "print(double(1.5))\n# 3.0\n";
|
||||
|
||||
export const PyFunctionRefresh = "refresh = table.refresh_column_async(\"doubled\")\nresult = refresh.wait()\nprint(result.rows_assigned)\n# 3\n";
|
||||
|
||||
export const PyFunctionRegister = "job = db.create_function_async(double)\nversion = job.wait()\n";
|
||||
|
||||
Reference in New Issue
Block a user