feat(functions): address Functions by namespace path (#4341)

Adds namespace addressing to Functions, matching what Secrets already
have (#4150).

## API

`create_function` / `create_function_async`, `get_function`,
`list_functions`, `drop_function` / `drop_function_async` take
`namespace_path` — keyword-only in Python (sync, async and remote
connections) and a trailing `&[String]` in Rust — defaulting to the
root. The namespace is addressed in the route identifier
(`/v1/function/analytics$features$embed/describe`,
`/v1/namespace/analytics$features/function/list`); no body carries it.

`FunctionVersion` and `FunctionVersionRef` name their Function **by
parts** — a short `name` plus `namespace_path` — the same shape as
`SecretReference`, so a returned version reopens from its own fields:

```python
fv = db.create_function(embed, namespace_path=["analytics", "features"])
fv.name, fv.namespace_path          # "embed", ("analytics", "features")
db.get_function(fv.name, version=fv.version, namespace_path=list(fv.namespace_path))
```

An empty path is omitted from the wire, so root documents keep their
shape. `namespace_path` is passed straight to the binding, so a bare
string (`namespace_path="prod"`) is a `TypeError` rather than four
one-character segments; segments are checked for addressability before a
request is built.

The binding and application allow-lists now share one field list,
`FUNCTION_VERSION_REF_FIELDS`, which admits `namespace_path` — without
it a namespaced binding or application read as coming from a newer
contract.

## Service side

The Sophon half serves and persists FunctionVersion / FunctionVersionRef
documents in this split form, keeping the joined id as its internal
catalog key. It lands in sophon with the submodule bump.

## Tests

- Rust: every Function verb addresses a namespace in the path; an
unaddressable segment never reaches the transport; `FunctionVersion`
round-trips a namespace and `reference()` carries it; a namespaced
binding and application are known shapes.
- Python: sync and async verbs address the namespace; a created Function
reopens from its own `name`/`namespace_path`; a bare-string path is
refused before any request.

Draft: compiles, `cargo fmt --check` and clippy `-D warnings` clean;
lancedb lib Function tests (116) and `first_class_function_slice2` pass
on this base.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Jonathan Hsieh
2026-09-29 14:07:14 -07:00
committed by GitHub
co-authored by Claude Opus 5.5
parent 0a1c1a04c7
commit 82d39485a4
13 changed files with 602 additions and 111 deletions
+13 -5
View File
@@ -149,12 +149,20 @@ class Connection(object):
limit: Optional[int],
) -> list[str]: ... # Deprecated: Use list_tables instead
async def open_job(self, job_id: str) -> Job: ...
async def create_function_async(self, request_json: str) -> Job: ...
async def get_function(self, name: str, version: str) -> str: ...
async def list_functions(self) -> List[str]: ...
async def drop_function(self, name: str, version: str) -> bool: ...
async def create_function_async(
self, request_json: str, namespace_path: Optional[List[str]] = None
) -> Job: ...
async def get_function(
self, name: str, version: str, namespace_path: Optional[List[str]] = None
) -> str: ...
async def list_functions(
self, namespace_path: Optional[List[str]] = None
) -> List[str]: ...
async def drop_function(
self, name: str, version: str, namespace_path: Optional[List[str]] = None
) -> bool: ...
async def drop_function_async(
self, name: str, version: str
self, name: str, version: str, namespace_path: Optional[List[str]] = None
) -> Tuple[bool, Job]: ...
async def create_secret(
self, name: str, value: str, namespace_path: Optional[List[str]] = None
+123 -31
View File
@@ -784,6 +784,7 @@ class DBConnection(EnforceOverrides):
definition: UdfDefinition,
*,
secrets: Optional[Sequence[EnvVarSecret]] = None,
namespace_path: Optional[List[str]] = None,
) -> FunctionVersion:
"""Build and register a scalar Python UDF, then return its version.
@@ -800,6 +801,9 @@ class DBConnection(EnforceOverrides):
Function needs, each naming a Secret and the environment variable
its value arrives in. The Function's source is unchanged by this;
it reads the variable the way it already did.
namespace_path : list of str, optional
The namespace holding the Function. None or an empty list is the
root namespace.
Examples
--------
@@ -815,13 +819,16 @@ class DBConnection(EnforceOverrides):
)
```
"""
return self.create_function_async(definition, secrets=secrets).wait()
return self.create_function_async(
definition, secrets=secrets, namespace_path=namespace_path
).wait()
def create_function_async(
self,
definition: UdfDefinition,
*,
secrets: Optional[Sequence[EnvVarSecret]] = None,
namespace_path: Optional[List[str]] = None,
) -> Job[FunctionVersion]:
"""Submit a scalar Python UDF for building and registration.
@@ -833,17 +840,26 @@ class DBConnection(EnforceOverrides):
"Function catalog operations are not supported for this connection type"
)
def get_function(self, name: str, *, version: str) -> FunctionVersion:
def get_function(
self,
name: str,
*,
version: str,
namespace_path: Optional[List[str]] = None,
) -> FunctionVersion:
"""Open one exact immutable Function version from the remote catalog."""
raise NotImplementedError(
"Function catalog operations are not supported for this connection type"
)
def list_functions(self) -> List[FunctionVersion]:
"""List every published immutable Function version.
def list_functions(
self, *, namespace_path: Optional[List[str]] = None
) -> List[FunctionVersion]:
"""List every published immutable Function version in a namespace.
Results are ordered by Function name then version. Local connections
raise ``NotImplementedError``.
Functions in child namespaces are not included. Results are ordered by
Function name then version. Local connections raise
``NotImplementedError``.
Examples
--------
@@ -857,7 +873,13 @@ class DBConnection(EnforceOverrides):
"Function catalog operations are not supported for this connection type"
)
def drop_function(self, name: str, *, version: str) -> bool:
def drop_function(
self,
name: str,
*,
version: str,
namespace_path: Optional[List[str]] = None,
) -> bool:
"""Drop a Function name and the object it was bound to.
The requested version must exist in the currently named object. Returns
@@ -870,7 +892,13 @@ class DBConnection(EnforceOverrides):
"Function catalog operations are not supported for this connection type"
)
def drop_function_async(self, name: str, *, version: str) -> Tuple[bool, Job]:
def drop_function_async(
self,
name: str,
*,
version: str,
namespace_path: Optional[List[str]] = None,
) -> Tuple[bool, Job]:
"""Drop a Function name and return its cleanup job.
The name is unbound before this returns; the object's content may still
@@ -1808,25 +1836,62 @@ class LanceDBConnection(DBConnection):
definition: UdfDefinition,
*,
secrets: Optional[Sequence[EnvVarSecret]] = None,
namespace_path: Optional[List[str]] = None,
) -> Job[FunctionVersion]:
job = LOOP.run(self._conn.create_function_async(definition, secrets=secrets))
job = LOOP.run(
self._conn.create_function_async(
definition, secrets=secrets, namespace_path=namespace_path
)
)
return Job(job)
@override
def get_function(self, name: str, *, version: str) -> FunctionVersion:
return LOOP.run(self._conn.get_function(name, version=version))
def get_function(
self,
name: str,
*,
version: str,
namespace_path: Optional[List[str]] = None,
) -> FunctionVersion:
return LOOP.run(
self._conn.get_function(
name, version=version, namespace_path=namespace_path
)
)
@override
def list_functions(self) -> List[FunctionVersion]:
return LOOP.run(self._conn.list_functions())
def list_functions(
self, *, namespace_path: Optional[List[str]] = None
) -> List[FunctionVersion]:
return LOOP.run(self._conn.list_functions(namespace_path=namespace_path))
@override
def drop_function(self, name: str, *, version: str) -> bool:
return LOOP.run(self._conn.drop_function(name, version=version))
def drop_function(
self,
name: str,
*,
version: str,
namespace_path: Optional[List[str]] = None,
) -> bool:
return LOOP.run(
self._conn.drop_function(
name, version=version, namespace_path=namespace_path
)
)
@override
def drop_function_async(self, name: str, *, version: str) -> Tuple[bool, Job]:
dropped, job = LOOP.run(self._conn.drop_function_async(name, version=version))
def drop_function_async(
self,
name: str,
*,
version: str,
namespace_path: Optional[List[str]] = None,
) -> Tuple[bool, Job]:
dropped, job = LOOP.run(
self._conn.drop_function_async(
name, version=version, namespace_path=namespace_path
)
)
return dropped, Job(job)
@override
@@ -2787,6 +2852,7 @@ class AsyncConnection(object):
definition: UdfDefinition,
*,
secrets: Optional[Sequence[EnvVarSecret]] = None,
namespace_path: Optional[List[str]] = None,
) -> AsyncJob[FunctionVersion]:
"""Submit a scalar Python UDF for building and registration.
@@ -2794,36 +2860,60 @@ class AsyncConnection(object):
artifact. Waiting on the job returns the immutable Function version.
``secrets`` is a sequence of
[EnvVarSecret][lancedb.secrets.EnvVarSecret], each naming a Secret and
the environment variable its value arrives in. Local connections raise
``NotImplementedError``.
the environment variable its value arrives in. ``namespace_path`` is
the namespace the Function is registered in; None or an empty list is
the root. Local connections raise ``NotImplementedError``.
"""
if not isinstance(definition, UdfDefinition):
raise TypeError("create_function_async requires a @udf definition")
request = definition.bind_secrets(secrets)
inner = await self._inner.create_function_async(request.to_canonical_json())
inner = await self._inner.create_function_async(
request.to_canonical_json(), namespace_path
)
return _typed_job(inner, FunctionVersion.from_json)
async def get_function(self, name: str, *, version: str) -> FunctionVersion:
async def get_function(
self,
name: str,
*,
version: str,
namespace_path: Optional[List[str]] = None,
) -> FunctionVersion:
"""Open one exact immutable Function version from the remote catalog."""
return FunctionVersion.from_json(await self._inner.get_function(name, version))
return FunctionVersion.from_json(
await self._inner.get_function(name, version, namespace_path)
)
async def list_functions(self) -> List[FunctionVersion]:
"""List every published immutable Function version.
async def list_functions(
self, *, namespace_path: Optional[List[str]] = None
) -> List[FunctionVersion]:
"""List every published immutable Function version in a namespace.
Results are ordered by Function name then version. Local connections
raise ``NotImplementedError``.
Functions in child namespaces are not included. Results are ordered by
Function name then version. Local connections raise
``NotImplementedError``.
"""
return [
FunctionVersion.from_json(value)
for value in await self._inner.list_functions()
for value in await self._inner.list_functions(namespace_path)
]
async def drop_function(self, name: str, *, version: str) -> bool:
async def drop_function(
self,
name: str,
*,
version: str,
namespace_path: Optional[List[str]] = None,
) -> bool:
"""Drop a Function name and the object it was bound to."""
return await self._inner.drop_function(name, version)
return await self._inner.drop_function(name, version, namespace_path)
async def drop_function_async(
self, name: str, *, version: str
self,
name: str,
*,
version: str,
namespace_path: Optional[List[str]] = None,
) -> Tuple[bool, AsyncJob]:
"""Drop a Function name and return its cleanup job.
@@ -2831,7 +2921,9 @@ class AsyncConnection(object):
be being deleted. Await :meth:`AsyncJob.wait` to wait for that to
finish.
"""
dropped, job = await self._inner.drop_function_async(name, version)
dropped, job = await self._inner.drop_function_async(
name, version, namespace_path
)
return dropped, AsyncJob(job)
async def create_secret(
+10 -1
View File
@@ -356,9 +356,16 @@ _ObjectVersion = Annotated[
class FunctionVersion(_RemoteValue):
"""A pinned object revision, independent of its executable image digest."""
"""A pinned object revision, independent of its executable image digest.
``name`` and ``namespace_path`` are the Function's parts, as a
[SecretReference][lancedb.functions.SecretReference] carries a Secret's;
they are what [get_function][lancedb.db.DBConnection.get_function] takes.
An empty ``namespace_path`` is the root namespace.
"""
name: str
namespace_path: tuple[str, ...] = ()
object_id: str
location: str
version: _ObjectVersion
@@ -444,6 +451,7 @@ class FunctionVersion(_RemoteValue):
return FunctionApplication(
function=FunctionVersionRef(
name=self.name,
namespace_path=self.namespace_path,
object_id=self.object_id,
location=self.location,
version=self.version,
@@ -502,6 +510,7 @@ class FunctionRegistrationRequest(_RemoteValue):
class FunctionVersionRef(_OpenRemoteValue):
name: str
namespace_path: tuple[str, ...] = ()
object_id: str
location: str
version: _ObjectVersion
+46 -9
View File
@@ -862,25 +862,62 @@ class RemoteDBConnection(DBConnection):
definition: UdfDefinition,
*,
secrets: Optional[Sequence[EnvVarSecret]] = None,
namespace_path: Optional[List[str]] = None,
) -> Job[FunctionVersion]:
job = LOOP.run(self._conn.create_function_async(definition, secrets=secrets))
job = LOOP.run(
self._conn.create_function_async(
definition, secrets=secrets, namespace_path=namespace_path
)
)
return Job(job)
@override
def get_function(self, name: str, *, version: str) -> FunctionVersion:
return LOOP.run(self._conn.get_function(name, version=version))
def get_function(
self,
name: str,
*,
version: str,
namespace_path: Optional[List[str]] = None,
) -> FunctionVersion:
return LOOP.run(
self._conn.get_function(
name, version=version, namespace_path=namespace_path
)
)
@override
def list_functions(self) -> List[FunctionVersion]:
return LOOP.run(self._conn.list_functions())
def list_functions(
self, *, namespace_path: Optional[List[str]] = None
) -> List[FunctionVersion]:
return LOOP.run(self._conn.list_functions(namespace_path=namespace_path))
@override
def drop_function(self, name: str, *, version: str) -> bool:
return LOOP.run(self._conn.drop_function(name, version=version))
def drop_function(
self,
name: str,
*,
version: str,
namespace_path: Optional[List[str]] = None,
) -> bool:
return LOOP.run(
self._conn.drop_function(
name, version=version, namespace_path=namespace_path
)
)
@override
def drop_function_async(self, name: str, *, version: str) -> Tuple[bool, Job]:
dropped, job = LOOP.run(self._conn.drop_function_async(name, version=version))
def drop_function_async(
self,
name: str,
*,
version: str,
namespace_path: Optional[List[str]] = None,
) -> Tuple[bool, Job]:
dropped, job = LOOP.run(
self._conn.drop_function_async(
name, version=version, namespace_path=namespace_path
)
)
return dropped, Job(job)
@override
@@ -162,6 +162,29 @@ def test_function_version_binds_named_columns_as_one_immutable_application():
] == [("text", "column", "documents.body")]
def test_function_version_names_its_namespace_as_a_separate_field():
"""A Function is named by its parts, as a Secret is; a root one states no
namespace."""
value = job_result("remote_function_job.json")
root = FunctionVersion.from_json(json.dumps(value))
assert root.namespace_path == ()
assert "namespace_path" not in json.loads(root.to_canonical_json())
nested = FunctionVersion(**{**value, "namespace_path": ["analytics", "features"]})
assert nested.name == "embed"
assert nested.namespace_path == ("analytics", "features")
assert json.loads(nested.to_canonical_json())["namespace_path"] == [
"analytics",
"features",
]
application = nested(text=col("documents.body"))
assert application.function.name == "embed"
assert application.function.namespace_path == ("analytics", "features")
assert json.loads(application.to_canonical_json())["function"][
"namespace_path"
] == ["analytics", "features"]
def test_function_version_binding_validates_names_and_direct_columns():
version = FunctionVersion.from_json(
json.dumps(job_result("remote_function_job.json"))
@@ -1539,8 +1539,10 @@ def _mock_remote_function_catalog():
else (None, None)
)
if function_action[1] == "create":
*namespace_path, name = function_action[0].split("$")
state["version"] = {
"name": "normalize_score",
"name": name,
**({"namespace_path": namespace_path} if namespace_path else {}),
"version": FUNCTION_VERSION,
"object_id": "fixture",
"location": "memory:///fixture",
@@ -1565,10 +1567,18 @@ def _mock_remote_function_catalog():
"job_state": "DONE",
"result": state["version"],
}
elif self.path == "/v1/function/normalize_score/describe":
# A namespaced Function's id is its namespace path and name joined
# by `$`, so the name is the last segment of it.
elif (
function_action[1] == "describe"
and function_action[0].split("$")[-1] == "normalize_score"
):
assert body == {"version": FUNCTION_VERSION}
response = state["version"]
elif self.path == "/v1/function/normalize_score/drop":
elif (
function_action[1] == "drop"
and function_action[0].split("$")[-1] == "normalize_score"
):
assert body == {"version": FUNCTION_VERSION}
response = {"dropped": True}
elif secret_action[1] in ("create", "alter"):
@@ -1602,7 +1612,8 @@ def _mock_remote_function_catalog():
assert query["page_token"] == "next"
self._write_response(200, {"secrets": [{"name": "hf-prod"}]})
return
if url.path != "/v1/namespace/$/function/list":
parts = url.path.strip("/").split("/")
if parts[:2] != ["v1", "namespace"] or parts[3:] != ["function", "list"]:
self._write_response(404, {"error": "not found"})
return
assert query["include_definition"] == "true"
@@ -1840,3 +1851,102 @@ async def test_async_remote_drop_function_sends_exact_version():
{"version": FUNCTION_VERSION},
)
]
def test_remote_function_verbs_address_a_namespace_in_the_path():
"""A Function's namespace is part of the identifier each route addresses;
no body carries it, so a namespaced request differs only in its path."""
namespace = ["analytics", "features"]
with _mock_remote_function_catalog() as (host, state):
db = lancedb.connect(
"db://dev",
api_key="fake",
host_override=host,
client_config={"retry_config": {"retries": 0}},
)
created = db.create_function(normalize_score, namespace_path=namespace)
# The returned parts are what the other verbs take, so a created
# Function reopens from its own fields.
assert created.name == "normalize_score"
assert created.namespace_path == tuple(namespace)
reopened = db.get_function(
created.name,
version=created.version,
namespace_path=list(created.namespace_path),
)
assert reopened == created
assert reopened(value=lancedb.col("score")).function.namespace_path == tuple(
namespace
)
db.list_functions(namespace_path=namespace)
assert (
db.drop_function(
"normalize_score", version=FUNCTION_VERSION, namespace_path=namespace
)
is True
)
dropped, _ = db.drop_function_async(
"normalize_score", version=FUNCTION_VERSION, namespace_path=namespace
)
assert dropped is True
assert [path for path, _ in state["requests"]] == [
"/v1/function/analytics$features$normalize_score/create",
"/v1/jobs/describe",
"/v1/function/analytics$features$normalize_score/describe",
"/v1/namespace/analytics$features/function/list",
"/v1/namespace/analytics$features/function/list",
"/v1/function/analytics$features$normalize_score/drop",
"/v1/function/analytics$features$normalize_score/drop",
]
@pytest.mark.asyncio
async def test_async_remote_function_verbs_address_a_namespace_in_the_path():
namespace = ["analytics", "features"]
with _mock_remote_function_catalog() as (host, state):
db = await lancedb.connect_async(
"db://dev",
api_key="fake",
host_override=host,
client_config={"retry_config": {"retries": 0}},
)
registration = await db.create_function_async(
normalize_score, namespace_path=namespace
)
await registration.wait()
await db.get_function(
"normalize_score", version=FUNCTION_VERSION, namespace_path=namespace
)
await db.list_functions(namespace_path=namespace)
assert (
await db.drop_function(
"normalize_score", version=FUNCTION_VERSION, namespace_path=namespace
)
is True
)
assert [path for path, _ in state["requests"]] == [
"/v1/function/analytics$features$normalize_score/create",
"/v1/jobs/describe",
"/v1/function/analytics$features$normalize_score/describe",
"/v1/namespace/analytics$features/function/list",
"/v1/namespace/analytics$features/function/list",
"/v1/function/analytics$features$normalize_score/drop",
]
def test_a_function_namespace_path_must_be_a_list_of_segments():
"""A bare string is refused rather than read as a sequence of one-character
segments, and nothing reaches the service."""
with _mock_remote_function_catalog() as (host, state):
db = lancedb.connect(
"db://dev",
api_key="fake",
host_override=host,
client_config={"retry_config": {"retries": 0}},
)
with pytest.raises(TypeError):
db.list_functions(namespace_path="analytics")
assert state["requests"] == []
+26 -6
View File
@@ -754,31 +754,37 @@ impl Connection {
})
}
#[pyo3(signature = (request_json, namespace_path=None))]
pub fn create_function_async(
self_: PyRef<'_, Self>,
request_json: String,
namespace_path: Option<Vec<String>>,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.get_inner()?.clone();
let request = lancedb::function::FunctionRegistrationRequest::from_json(&request_json)
.infer_error()?;
let namespace_path = namespace_path.unwrap_or_default();
future_into_py(self_.py(), async move {
inner
.create_function_async(request)
.create_function_async(request, &namespace_path)
.await
.infer_error()
.map(crate::job::Job::new_typed)
})
}
#[pyo3(signature = (name, version, namespace_path=None))]
pub fn get_function(
self_: PyRef<'_, Self>,
name: String,
version: String,
namespace_path: Option<Vec<String>>,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.get_inner()?.clone();
let namespace_path = namespace_path.unwrap_or_default();
future_into_py(self_.py(), async move {
inner
.get_function(name, version)
.get_function(name, version, &namespace_path)
.await
.infer_error()?
.to_canonical_json()
@@ -786,11 +792,16 @@ impl Connection {
})
}
pub fn list_functions(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
#[pyo3(signature = (namespace_path=None))]
pub fn list_functions(
self_: PyRef<'_, Self>,
namespace_path: Option<Vec<String>>,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.get_inner()?.clone();
let namespace_path = namespace_path.unwrap_or_default();
future_into_py(self_.py(), async move {
inner
.list_functions()
.list_functions(&namespace_path)
.await
.infer_error()?
.into_iter()
@@ -799,26 +810,35 @@ impl Connection {
})
}
#[pyo3(signature = (name, version, namespace_path=None))]
pub fn drop_function(
self_: PyRef<'_, Self>,
name: String,
version: String,
namespace_path: Option<Vec<String>>,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.get_inner()?.clone();
let namespace_path = namespace_path.unwrap_or_default();
future_into_py(self_.py(), async move {
inner.drop_function(name, version).await.infer_error()
inner
.drop_function(name, version, &namespace_path)
.await
.infer_error()
})
}
#[pyo3(signature = (name, version, namespace_path=None))]
pub fn drop_function_async(
self_: PyRef<'_, Self>,
name: String,
version: String,
namespace_path: Option<Vec<String>>,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.get_inner()?.clone();
let namespace_path = namespace_path.unwrap_or_default();
future_into_py(self_.py(), async move {
inner
.drop_function_async(name, version)
.drop_function_async(name, version, &namespace_path)
.await
.infer_error()
.map(|(dropped, job)| (dropped, crate::job::Job::new(job)))
+20 -9
View File
@@ -587,7 +587,8 @@ impl Connection {
)
}
/// Build and register a Python callable as an immutable Function version.
/// Build and register a Python callable as an immutable Function version
/// in `namespace_path`, which is empty for the root namespace.
///
/// The server-side job builds the OCI image, then registers the completed
/// artifact. Waiting on the returned typed job yields the durable
@@ -597,8 +598,11 @@ impl Connection {
pub async fn create_function_async(
&self,
request: crate::function::FunctionRegistrationRequest,
namespace_path: &[String],
) -> Result<crate::job::Job<crate::function::FunctionVersion>> {
self.internal.create_function_async(request).await
self.internal
.create_function_async(request, namespace_path)
.await
}
/// Look up one exact immutable Function version in the remote catalog.
@@ -610,13 +614,15 @@ impl Connection {
&self,
name: impl AsRef<str>,
version: impl AsRef<str>,
namespace_path: &[String],
) -> Result<crate::function::FunctionVersion> {
self.internal
.get_function(name.as_ref(), version.as_ref())
.get_function(name.as_ref(), version.as_ref(), namespace_path)
.await
}
/// List every published immutable Function version in the remote catalog.
/// List every published immutable Function version in `namespace_path` of
/// the remote catalog. Functions in child namespaces are not included.
///
/// Results are ordered by Function name then version. The client walks all
/// server pages before returning. Local databases return
@@ -628,14 +634,17 @@ impl Connection {
/// # async fn list_functions(
/// # connection: &lancedb::Connection,
/// # ) -> Result<(), Box<dyn std::error::Error>> {
/// for function in connection.list_functions().await? {
/// for function in connection.list_functions(&[]).await? {
/// println!("{} {}", function.name(), function.version());
/// }
/// # Ok(())
/// # }
/// ```
pub async fn list_functions(&self) -> Result<Vec<crate::function::FunctionVersion>> {
self.internal.list_functions().await
pub async fn list_functions(
&self,
namespace_path: &[String],
) -> Result<Vec<crate::function::FunctionVersion>> {
self.internal.list_functions(namespace_path).await
}
/// Remove the current Function name binding, retaining the object history.
@@ -647,9 +656,10 @@ impl Connection {
&self,
name: impl AsRef<str>,
version: impl AsRef<str>,
namespace_path: &[String],
) -> Result<bool> {
self.internal
.drop_function(name.as_ref(), version.as_ref())
.drop_function(name.as_ref(), version.as_ref(), namespace_path)
.await
}
@@ -663,9 +673,10 @@ impl Connection {
&self,
name: impl AsRef<str>,
version: impl AsRef<str>,
namespace_path: &[String],
) -> Result<(bool, crate::job::Job)> {
self.internal
.drop_function_async(name.as_ref(), version.as_ref())
.drop_function_async(name.as_ref(), version.as_ref(), namespace_path)
.await
}
+18 -5
View File
@@ -333,10 +333,12 @@ pub trait Database:
///
/// See [`CloneTableRequest`] for detailed documentation and examples.
async fn clone_table(&self, request: CloneTableRequest) -> Result<Arc<dyn BaseTable>>;
/// Submit a Function creation job that builds an image and registers it.
/// Submit a Function creation job that builds an image and registers it
/// in `namespace_path`.
async fn create_function_async(
&self,
_request: crate::function::FunctionRegistrationRequest,
_namespace_path: &[String],
) -> Result<crate::job::Job<crate::function::FunctionVersion>> {
function_catalog_not_supported()
}
@@ -412,15 +414,25 @@ pub trait Database:
&self,
_name: &str,
_version: &str,
_namespace_path: &[String],
) -> Result<crate::function::FunctionVersion> {
function_catalog_not_supported()
}
/// List every published immutable Function version in the remote catalog.
async fn list_functions(&self) -> Result<Vec<crate::function::FunctionVersion>> {
/// List every published immutable Function version in one namespace of the
/// remote catalog.
async fn list_functions(
&self,
_namespace_path: &[String],
) -> Result<Vec<crate::function::FunctionVersion>> {
function_catalog_not_supported()
}
/// Remove the current Function name binding, retaining the object history.
async fn drop_function(&self, _name: &str, _version: &str) -> Result<bool> {
async fn drop_function(
&self,
_name: &str,
_version: &str,
_namespace_path: &[String],
) -> Result<bool> {
function_catalog_not_supported()
}
/// Start dropping a Function and return a handle to the cleanup job.
@@ -431,8 +443,9 @@ pub trait Database:
&self,
name: &str,
version: &str,
namespace_path: &[String],
) -> Result<(bool, crate::job::Job)> {
let dropped = self.drop_function(name, version).await?;
let dropped = self.drop_function(name, version, namespace_path).await?;
Ok((dropped, crate::job::Job::new_done()))
}
/// Create a named Secret in this database. Fails if the name is taken, so
+71 -12
View File
@@ -85,22 +85,25 @@ fn has_unknown_keys(value: &Value, allowed: &[&str]) -> bool {
.is_some_and(|object| object.keys().any(|key| !allowed.contains(&key.as_str())))
}
/// Every field a [`FunctionVersionRef`] document may carry. An application or
/// binding with any other key comes from a newer contract.
pub(crate) const FUNCTION_VERSION_REF_FIELDS: &[&str] = &[
"name",
"namespace_path",
"object_id",
"location",
"version",
"manifest_digest",
];
fn application_has_unknown_nested_fields(value: &Value) -> bool {
let Some(application) = value.as_object() else {
return false;
};
if application.get("function").is_some_and(|value| {
has_unknown_keys(
value,
&[
"name",
"object_id",
"location",
"version",
"manifest_digest",
],
)
}) {
if application
.get("function")
.is_some_and(|value| has_unknown_keys(value, FUNCTION_VERSION_REF_FIELDS))
{
return true;
}
if application
@@ -431,9 +434,17 @@ fn deserialize_object_version<'de, D: Deserializer<'de>>(
}
/// One immutable Function object revision and its executable artifact.
///
/// The Function is named by its parts, `name` and `namespace_path`, the way a
/// [`SecretReference`](crate::secrets::SecretReference) names a Secret.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct FunctionVersion {
name: String,
/// The namespace holding the Function. Empty is the root, and is absent
/// from the wire, as a [`SecretReference`](crate::secrets::SecretReference)
/// omits it.
#[serde(default, skip_serializing_if = "Vec::is_empty")]
namespace_path: Vec<String>,
object_id: String,
location: String,
#[serde(deserialize_with = "deserialize_object_version")]
@@ -463,6 +474,7 @@ impl FunctionVersion {
pub fn reference(&self) -> FunctionVersionRef {
FunctionVersionRef {
name: self.name.clone(),
namespace_path: self.namespace_path.clone(),
object_id: self.object_id.clone(),
location: self.location.clone(),
version: self.version.clone(),
@@ -472,6 +484,9 @@ impl FunctionVersion {
pub fn name(&self) -> &str {
&self.name
}
pub fn namespace_path(&self) -> &[String] {
&self.namespace_path
}
pub fn version(&self) -> &str {
&self.version
}
@@ -556,6 +571,10 @@ impl_json!(FunctionRegistrationRequest);
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct FunctionVersionRef {
pub name: String,
/// The namespace holding the Function; empty is the root and is absent
/// from the wire.
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub namespace_path: Vec<String>,
pub object_id: String,
pub location: String,
#[serde(deserialize_with = "deserialize_object_version")]
@@ -954,6 +973,46 @@ mod tests {
);
}
/// A Function is named by its parts, and a root one states no namespace.
#[test]
fn function_version_carries_its_namespace_as_a_separate_field() {
const VERSION: &str = include_str!(
"../tests/fixtures/first_class_functions/v1/remote_function_version.canonical.json"
);
let root = FunctionVersion::from_json(VERSION).unwrap();
assert!(root.namespace_path().is_empty());
assert_eq!(root.to_canonical_json().unwrap(), VERSION.trim_end());
let mut document: Value = serde_json::from_str(VERSION).unwrap();
document["namespace_path"] = serde_json::json!(["analytics", "features"]);
let nested = FunctionVersion::from_json(&document.to_string()).unwrap();
assert_eq!(nested.name(), "embed");
assert_eq!(nested.namespace_path(), ["analytics", "features"]);
let reference = nested.reference();
assert_eq!(reference.name, "embed");
assert_eq!(reference.namespace_path, ["analytics", "features"]);
let reference: Value = serde_json::to_value(&reference).unwrap();
assert_eq!(
reference["namespace_path"],
serde_json::json!(["analytics", "features"])
);
let root_reference = serde_json::to_value(root.reference()).unwrap();
assert!(root_reference.get("namespace_path").is_none());
// An application carrying the namespace is not from a newer contract.
let application = serde_json::json!({
"function": reference,
"inputs": [],
"output": serde_json::to_value(nested.signature().output.clone()).unwrap(),
});
let application = FunctionApplication::from_json(&application.to_string()).unwrap();
assert!(!application.has_unknown_fields());
assert_eq!(
application.function().namespace_path,
["analytics", "features"]
);
}
/// Arrays are ordered by the caller, so canonicalization must leave them
/// alone -- sorting them would change what a signature means.
#[test]
+117 -19
View File
@@ -979,8 +979,9 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
async fn create_function_async(
&self,
request: FunctionRegistrationRequest,
namespace_path: &[String],
) -> Result<Job<FunctionVersion>> {
let function_id = build_object_identifier("Function name", &request.name, &[])?;
let function_id = build_object_identifier("Function name", &request.name, namespace_path)?;
let req = self
.client
.post(&format!("/v1/function/{function_id}/create"))
@@ -1005,8 +1006,13 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
))))
}
async fn get_function(&self, name: &str, version: &str) -> Result<FunctionVersion> {
let function_id = build_object_identifier("Function name", name, &[])?;
async fn get_function(
&self,
name: &str,
version: &str,
namespace_path: &[String],
) -> Result<FunctionVersion> {
let function_id = build_object_identifier("Function name", name, namespace_path)?;
let req = self
.client
.post(&format!("/v1/function/{function_id}/describe"))
@@ -1018,8 +1024,8 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
response.json().await.err_to_http(request_id)
}
async fn list_functions(&self) -> Result<Vec<FunctionVersion>> {
let namespace_id = build_namespace_identifier(&[])?;
async fn list_functions(&self, namespace_path: &[String]) -> Result<Vec<FunctionVersion>> {
let namespace_id = build_namespace_identifier(namespace_path)?;
let path = format!("/v1/namespace/{namespace_id}/function/list");
let mut functions = Vec::new();
let mut page_token: Option<String> = None;
@@ -1059,12 +1065,25 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
Ok(functions)
}
async fn drop_function(&self, name: &str, version: &str) -> Result<bool> {
Ok(self.drop_function_async(name, version).await?.0)
async fn drop_function(
&self,
name: &str,
version: &str,
namespace_path: &[String],
) -> Result<bool> {
Ok(self
.drop_function_async(name, version, namespace_path)
.await?
.0)
}
async fn drop_function_async(&self, name: &str, version: &str) -> Result<(bool, Job)> {
let function_id = build_object_identifier("Function name", name, &[])?;
async fn drop_function_async(
&self,
name: &str,
version: &str,
namespace_path: &[String],
) -> Result<(bool, Job)> {
let function_id = build_object_identifier("Function name", name, namespace_path)?;
let req = self
.client
.post(&format!("/v1/function/{function_id}/drop"))
@@ -2004,7 +2023,7 @@ mod tests {
.body(serde_json::json!({"dropped": true, "job_id": "j1-fn-drop"}).to_string())
.unwrap()
});
let (dropped, job) = db.drop_function_async("embed", "1").await.unwrap();
let (dropped, job) = db.drop_function_async("embed", "1", &[]).await.unwrap();
assert!(dropped);
assert_eq!(job.id(), Some("j1-fn-drop"));
}
@@ -2019,7 +2038,7 @@ mod tests {
.body(serde_json::json!({"dropped": false}).to_string())
.unwrap()
});
let (dropped, job) = db.drop_function_async("embed", "1").await.unwrap();
let (dropped, job) = db.drop_function_async("embed", "1", &[]).await.unwrap();
assert!(!dropped);
assert_eq!(job.id(), None);
assert_eq!(job.status().await.unwrap(), "finished");
@@ -2036,7 +2055,11 @@ mod tests {
let db = super::RemoteDatabase::new_mock(move |_| {
http::Response::builder().status(202).body(body).unwrap()
});
let error = db.drop_function_async("embed", "1").await.err().unwrap();
let error = db
.drop_function_async("embed", "1", &[])
.await
.err()
.unwrap();
assert!(error.to_string().contains("valid job_id"));
}
}
@@ -3905,7 +3928,7 @@ mod tests {
http::Response::builder().status(200).body("{}").unwrap()
});
let error = conn
.drop_function(name, "fv_1")
.drop_function(name, "fv_1", &[])
.await
.expect_err("a dot-only Function name must be refused");
assert!(
@@ -4307,7 +4330,7 @@ mod tests {
path => panic!("unexpected path: {path}"),
});
let request = crate::function::FunctionRegistrationRequest::from_json(REQUEST).unwrap();
let job = conn.create_function_async(request).await.unwrap();
let job = conn.create_function_async(request, &[]).await.unwrap();
assert_eq!(job.id(), Some("job-function-1"));
let version = job.wait().await.unwrap();
assert_eq!(version.name(), "embed");
@@ -4327,7 +4350,7 @@ mod tests {
assert_eq!(body, serde_json::json!({"version": "1"}));
http::Response::builder().status(200).body(VERSION).unwrap()
});
let version = conn.get_function("embed", "1").await.unwrap();
let version = conn.get_function("embed", "1", &[]).await.unwrap();
assert_eq!(version.name(), "embed");
assert_eq!(version.version(), "1");
}
@@ -4370,7 +4393,7 @@ mod tests {
}
}
});
let functions = conn.list_functions().await.unwrap();
let functions = conn.list_functions(&[]).await.unwrap();
assert_eq!(functions.len(), 1);
assert_eq!(functions[0].name(), "embed");
assert_eq!(functions[0].version(), "1");
@@ -4393,7 +4416,7 @@ mod tests {
.unwrap()
});
let functions = conn.list_functions().await.unwrap();
let functions = conn.list_functions(&[]).await.unwrap();
assert!(functions.is_empty());
assert_eq!(requests.load(Ordering::SeqCst), 1);
}
@@ -4434,7 +4457,7 @@ mod tests {
.unwrap()
});
let error = conn.list_functions().await.unwrap_err();
let error = conn.list_functions(&[]).await.unwrap_err();
assert!(
matches!(
&error,
@@ -4461,7 +4484,82 @@ mod tests {
.body(r#"{"dropped":false}"#)
.unwrap()
});
assert!(!conn.drop_function("embed", "1").await.unwrap());
assert!(!conn.drop_function("embed", "1", &[]).await.unwrap());
}
/// A Function's namespace is addressed in the path the way a Secret's is:
/// every route takes the joined identifier and no body carries the
/// namespace, so a namespaced request differs from a root one only in its
/// path.
#[tokio::test]
async fn test_a_function_namespace_path_is_addressed_in_the_path() {
const REQUEST: &str = include_str!(
"../../tests/fixtures/first_class_functions/v1/remote_function_registration_request.json"
);
const VERSION: &str = include_str!(
"../../tests/fixtures/first_class_functions/v1/remote_function_version.canonical.json"
);
let namespace = ["analytics".to_string(), "features".to_string()];
let mut expected_create: serde_json::Value = serde_json::from_str(REQUEST).unwrap();
expected_create.as_object_mut().unwrap().remove("name");
let conn = Connection::new_with_handler(move |request| {
let body = request
.body()
.and_then(|body| body.as_bytes())
.map(|bytes| serde_json::from_slice::<serde_json::Value>(bytes).unwrap());
match request.url().path() {
"/v1/function/analytics$features$normalize_score/create" => {
assert_eq!(body.unwrap(), expected_create);
http::Response::builder()
.status(202)
.body(r#"{"job_id":"job-function-1"}"#.to_string())
.unwrap()
}
"/v1/function/analytics$features$embed/describe" => {
assert_eq!(body.unwrap(), serde_json::json!({"version": "1"}));
http::Response::builder()
.status(200)
.body(VERSION.to_string())
.unwrap()
}
"/v1/function/analytics$features$embed/drop" => {
assert_eq!(body.unwrap(), serde_json::json!({"version": "1"}));
http::Response::builder()
.status(200)
.body(r#"{"dropped":true}"#.to_string())
.unwrap()
}
// Listing is namespace-scoped, so the namespace is the whole
// identifier.
"/v1/namespace/analytics$features/function/list" => http::Response::builder()
.status(200)
.body(r#"{"functions":[]}"#.to_string())
.unwrap(),
path => panic!("unexpected path: {path}"),
}
});
let request = crate::function::FunctionRegistrationRequest::from_json(REQUEST).unwrap();
let job = conn
.create_function_async(request, &namespace)
.await
.unwrap();
assert_eq!(job.id(), Some("job-function-1"));
conn.get_function("embed", "1", &namespace).await.unwrap();
assert!(conn.list_functions(&namespace).await.unwrap().is_empty());
assert!(conn.drop_function("embed", "1", &namespace).await.unwrap());
}
/// Each namespace segment is checked before a request is built, so an
/// empty segment cannot collapse the identifier onto the parent namespace.
#[tokio::test]
async fn test_an_unaddressable_function_namespace_segment_is_refused() {
let conn = Connection::new_with_handler(|request| -> http::Response<String> {
panic!("reached the transport: {}", request.url().path())
});
let namespace = ["analytics".to_string(), String::new()];
assert!(conn.get_function("embed", "1", &namespace).await.is_err());
assert!(conn.list_functions(&namespace).await.is_err());
assert!(conn.drop_function("embed", "1", &namespace).await.is_err());
}
#[tokio::test]
+15 -7
View File
@@ -704,13 +704,7 @@ fn ensure_known_binding_shape(value: &Value) -> Result<()> {
object
.get("function")
.ok_or_else(|| invalid_function("Function binding is missing its exact version"))?,
&[
"name",
"object_id",
"location",
"version",
"manifest_digest",
],
crate::function::FUNCTION_VERSION_REF_FIELDS,
"version reference",
)?;
for input in object
@@ -3760,6 +3754,20 @@ mod tests {
ensure_known_binding_shape(&raw_binding).unwrap();
}
/// A binding names its Function by parts, so one declared against a
/// namespaced Function is part of the known contract too.
#[test]
fn test_namespaced_bindings_are_a_known_shape() {
let mut raw_binding: Value = serde_json::from_str(include_str!(
"../../tests/fixtures/first_class_functions/v1/remote_initialized_function_binding.json"
))
.unwrap();
raw_binding["function"]["namespace_path"] = serde_json::json!(["analytics", "features"]);
ensure_known_binding_shape(&raw_binding).unwrap();
let binding = FunctionBinding::from_json(&raw_binding.to_string()).unwrap();
assert_eq!(binding.function().namespace_path, ["analytics", "features"]);
}
#[test]
fn test_named_struct_can_be_kept_as_one_nullable_physical_column() {
let application = named_struct_application("{}");
@@ -97,13 +97,16 @@ async fn local_function_catalog_operations_return_stable_not_supported() {
))
.unwrap();
let create_error = connection.create_function_async(request).await.unwrap_err();
let create_error = connection
.create_function_async(request, &[])
.await
.unwrap_err();
let lookup_error = connection
.get_function("normalize_score", "1")
.get_function("normalize_score", "1", &[])
.await
.unwrap_err();
let drop_error = connection
.drop_function("normalize_score", "1")
.drop_function("normalize_score", "1", &[])
.await
.unwrap_err();
for error in [create_error, lookup_error, drop_error] {