mirror of
https://github.com/lancedb/lancedb.git
synced 2026-11-13 22:26:53 +00:00
feat: resolve latest Function versions (#4949)
## Problem Function catalog lookup required an exact version string even though callers commonly need the latest currently visible immutable revision. On the current SDK, resolving that version required an out-of-band catalog query before calling `get_function`. ## Change - Make Python `get_function(name)` default to latest while accepting `int` or `str` for exact lookup. - Normalize integer versions to their decimal string form and reject `bool` explicitly. Existing string validation and errors remain unchanged. - Thread `None` through the rebuilt PyO3 binding and route it to the new Rust `Connection::get_function_latest`. - Resolve latest through the existing catalog `DESCRIBE FUNCTION` path without `WITH (version = ...)`. The returned value is one exact immutable version, not a mutable alias. - Preserve the existing Rust exact-version API. ## Validation On the designated EC2 host with the rebuilt extension: - `test_local_function_catalog_operations_are_not_supported`: **1 passed**. This crosses the Python connection, rebuilt PyO3, and embedded Rust `NotSupported` boundary with the omitted-version call. - Async and sync catalog-path tests: **2 passed**, 117 deselected. - Remote SQL Function creation/description integration test: **1 passed**. - An earlier Rust invocation with the wrong module path matched **0 tests**; it is retained as harness evidence and is not counted as a pass. ## Scope This is the Function lookup SDK slice. It does not include the separate atomic rebind/publication server work tracked under ENT-3418.
This commit is contained in:
@@ -107,7 +107,9 @@ single `id` is the SQL query UUID. Repeated waits reuse the completed result.
|
||||
</CodeBlock>
|
||||
|
||||
`create_function` is the blocking equivalent. A registered version is fetched
|
||||
with `db.get_function(name, version=...)`.
|
||||
with `db.get_function(name, version=...)`. The version may be an integer or
|
||||
its string form. Omit `version` to resolve and return the latest currently
|
||||
visible revision.
|
||||
|
||||
<Note>
|
||||
The SQL command submits an internal `create_function` image-build job and
|
||||
@@ -118,9 +120,10 @@ have no registry history and cannot be reopened with `open_job()`.
|
||||
## List and describe {#list-describe}
|
||||
|
||||
`list_functions` pages through every published version in a namespace,
|
||||
ordered by name then version; `get_function` opens one exact version by
|
||||
name and `version`. Both take `namespace_path`, defaulting to the root
|
||||
namespace, the same as `create_function`.
|
||||
ordered by name then version. `get_function` accepts an integer or string
|
||||
version, or resolves the latest visible revision when `version` is omitted.
|
||||
Both take `namespace_path`, defaulting to the root namespace, the same as
|
||||
`create_function`.
|
||||
|
||||
<CodeGroup>
|
||||
```python Python icon="python"
|
||||
@@ -128,6 +131,7 @@ for version in db.list_functions():
|
||||
print(version.name, version.version, version.created_at_millis)
|
||||
|
||||
version = db.get_function("double", version="1")
|
||||
latest = db.get_function("double")
|
||||
```
|
||||
|
||||
```rust
|
||||
@@ -140,6 +144,7 @@ while let Some(version) = versions.try_next().await? {
|
||||
}
|
||||
|
||||
let version = connection.get_function("double", "1", &[]).await?;
|
||||
let latest = connection.get_function_latest("double", &[]).await?;
|
||||
# Ok(())
|
||||
# }
|
||||
```
|
||||
@@ -151,6 +156,7 @@ SHOW FUNCTIONS
|
||||
SHOW FUNCTIONS IN "analytics" WITH (name = 'double', limit = 100)
|
||||
|
||||
DESCRIBE FUNCTION double WITH (version = '1')
|
||||
DESCRIBE FUNCTION double
|
||||
```
|
||||
</CodeGroup>
|
||||
|
||||
|
||||
@@ -94,9 +94,12 @@ registration handle has type `query` and uses the SQL query UUID as its single
|
||||
|
||||
```sql
|
||||
DESCRIBE FUNCTION embed WITH (version = '1', include_source = true)
|
||||
DESCRIBE FUNCTION embed
|
||||
```
|
||||
|
||||
DESCRIBE returns one `definition` JSON column for the exact version.
|
||||
DESCRIBE returns one `definition` JSON column for the exact version. When
|
||||
`version` is omitted, it resolves and returns the latest currently visible
|
||||
immutable revision.
|
||||
`include_source` defaults to false; source-inclusive reads report an omission
|
||||
when the stored artifact exceeds the inline source limit.
|
||||
|
||||
|
||||
@@ -190,7 +190,10 @@ class Connection(object):
|
||||
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
|
||||
self,
|
||||
name: str,
|
||||
version: Optional[str] = None,
|
||||
namespace_path: Optional[List[str]] = None,
|
||||
) -> str: ...
|
||||
def list_functions(
|
||||
self,
|
||||
|
||||
@@ -140,6 +140,18 @@ def _view_description(
|
||||
)
|
||||
|
||||
|
||||
def _normalize_function_version(version: Optional[Union[int, str]]) -> Optional[str]:
|
||||
if version is None:
|
||||
return None
|
||||
if isinstance(version, bool):
|
||||
raise TypeError("version must be an int, str, or None")
|
||||
if isinstance(version, int):
|
||||
return str(version)
|
||||
if isinstance(version, str):
|
||||
return version
|
||||
raise TypeError("version must be an int, str, or None")
|
||||
|
||||
|
||||
class DBConnection(EnforceOverrides):
|
||||
"""An active LanceDB connection interface.
|
||||
|
||||
@@ -1027,10 +1039,15 @@ class DBConnection(EnforceOverrides):
|
||||
self,
|
||||
name: str,
|
||||
*,
|
||||
version: str,
|
||||
version: Optional[Union[int, str]] = None,
|
||||
namespace_path: Optional[List[str]] = None,
|
||||
) -> FunctionVersion:
|
||||
"""Open one exact immutable Function version from the remote catalog."""
|
||||
"""Open an immutable Function version from the remote catalog.
|
||||
|
||||
``version`` may be an integer or string. When omitted, the server
|
||||
resolves the latest currently visible revision and returns that exact
|
||||
immutable version.
|
||||
"""
|
||||
raise NotImplementedError(
|
||||
_append_doc_url(
|
||||
"Function catalog operations are not supported "
|
||||
@@ -2310,9 +2327,10 @@ class LanceDBConnection(DBConnection):
|
||||
self,
|
||||
name: str,
|
||||
*,
|
||||
version: str,
|
||||
version: Optional[Union[int, str]] = None,
|
||||
namespace_path: Optional[List[str]] = None,
|
||||
) -> FunctionVersion:
|
||||
version = _normalize_function_version(version)
|
||||
return LOOP.run(
|
||||
self._conn.get_function(
|
||||
name, version=version, namespace_path=namespace_path
|
||||
@@ -3610,10 +3628,16 @@ class AsyncConnection(object):
|
||||
self,
|
||||
name: str,
|
||||
*,
|
||||
version: str,
|
||||
version: Optional[Union[int, str]] = None,
|
||||
namespace_path: Optional[List[str]] = None,
|
||||
) -> FunctionVersion:
|
||||
"""Open one exact immutable Function version from the remote catalog."""
|
||||
"""Open an immutable Function version from the remote catalog.
|
||||
|
||||
``version`` may be an integer or string. When omitted, the server
|
||||
resolves the latest currently visible revision and returns that exact
|
||||
immutable version.
|
||||
"""
|
||||
version = _normalize_function_version(version)
|
||||
return FunctionVersion.from_json(
|
||||
await self._inner.get_function(name, version, namespace_path)
|
||||
)
|
||||
|
||||
@@ -37,7 +37,7 @@ import pyarrow as pa
|
||||
|
||||
from ..common import DATA
|
||||
from lancedb.listing import Listing, NamespaceListing, TableListing
|
||||
from ..db import DBConnection, LOOP
|
||||
from ..db import DBConnection, LOOP, _normalize_function_version
|
||||
from ..functions import FunctionVersion, FunctionDefinition
|
||||
from ..graph import EdgeTable, NodeTable, PropertyGraph
|
||||
from ..job import AsyncJob, Job
|
||||
@@ -985,9 +985,10 @@ class RemoteDBConnection(DBConnection):
|
||||
self,
|
||||
name: str,
|
||||
*,
|
||||
version: str,
|
||||
version: Optional[Union[int, str]] = None,
|
||||
namespace_path: Optional[List[str]] = None,
|
||||
) -> FunctionVersion:
|
||||
version = _normalize_function_version(version)
|
||||
return LOOP.run(
|
||||
self._conn.get_function(
|
||||
name, version=version, namespace_path=namespace_path
|
||||
|
||||
@@ -1894,6 +1894,8 @@ def test_local_function_catalog_operations_are_not_supported(tmp_path):
|
||||
db.create_function_async(normalize_score)
|
||||
with pytest.raises(NotImplementedError, match=message):
|
||||
db.get_function("normalize_score", version=FUNCTION_VERSION)
|
||||
with pytest.raises(NotImplementedError, match=message):
|
||||
db.get_function("normalize_score")
|
||||
with pytest.raises(NotImplementedError, match=message):
|
||||
list(db.list_functions())
|
||||
with pytest.raises(NotImplementedError, match=message):
|
||||
@@ -1903,8 +1905,10 @@ def test_local_function_catalog_operations_are_not_supported(tmp_path):
|
||||
class _RawFunctionCatalog:
|
||||
def __init__(self, definition, *, encode=False):
|
||||
self.definition = json.dumps(definition) if encode else definition
|
||||
self.requested_versions = []
|
||||
|
||||
async def get_function(self, _name, version, namespace_path):
|
||||
async def get_function(self, _name, version=None, namespace_path=None):
|
||||
self.requested_versions.append(version)
|
||||
return self.definition
|
||||
|
||||
def list_functions(self, *, namespace_path=None, page_token=None, page_limit=None):
|
||||
@@ -1944,13 +1948,22 @@ async def test_async_catalog_path_preserves_execution_controls():
|
||||
row_timeout_seconds=8.0,
|
||||
result_stability="input_and_context",
|
||||
)
|
||||
connection = AsyncConnection(_RawFunctionCatalog(definition, encode=True))
|
||||
catalog = _RawFunctionCatalog(definition, encode=True)
|
||||
connection = AsyncConnection(catalog)
|
||||
|
||||
latest = await connection.get_function("embed")
|
||||
integer = await connection.get_function("embed", version=1)
|
||||
exact = await connection.get_function("embed", version="1")
|
||||
await connection.get_function("embed", version="001")
|
||||
listed = [item async for item in connection.list_functions(page_limit=32)]
|
||||
assert len(listed) == 1
|
||||
for version in [exact, *listed]:
|
||||
for version in [latest, integer, exact, *listed]:
|
||||
assert _assert_execution_control_encoding(version) == definition
|
||||
assert catalog.requested_versions == [None, "1", "1", "001"]
|
||||
with pytest.raises(TypeError, match="int, str, or None"):
|
||||
await connection.get_function("embed", version=True)
|
||||
with pytest.raises(TypeError, match="int, str, or None"):
|
||||
await connection.get_function("embed", version=1.5)
|
||||
|
||||
|
||||
def test_sync_catalog_path_preserves_execution_controls():
|
||||
@@ -1963,13 +1976,22 @@ def test_sync_catalog_path_preserves_execution_controls():
|
||||
result_stability="input_and_context",
|
||||
)
|
||||
connection = object.__new__(RemoteDBConnection)
|
||||
connection._conn = _RawFunctionCatalog(FunctionVersion(**definition))
|
||||
catalog = _RawFunctionCatalog(FunctionVersion(**definition))
|
||||
connection._conn = catalog
|
||||
|
||||
latest = connection.get_function("embed")
|
||||
integer = connection.get_function("embed", version=1)
|
||||
exact = connection.get_function("embed", version="1")
|
||||
connection.get_function("embed", version="001")
|
||||
listed = list(connection.list_functions(page_limit=32))
|
||||
assert len(listed) == 1
|
||||
for version in [exact, *listed]:
|
||||
for version in [latest, integer, exact, *listed]:
|
||||
assert _assert_execution_control_encoding(version) == definition
|
||||
assert catalog.requested_versions == [None, "1", "1", "001"]
|
||||
with pytest.raises(TypeError, match="int, str, or None"):
|
||||
connection.get_function("embed", version=True)
|
||||
with pytest.raises(TypeError, match="int, str, or None"):
|
||||
connection.get_function("embed", version=1.5)
|
||||
|
||||
|
||||
def test_sync_drop_function_emits_server_warnings_without_changing_bool_api():
|
||||
|
||||
@@ -877,22 +877,22 @@ impl Connection {
|
||||
})
|
||||
}
|
||||
|
||||
#[pyo3(signature = (name, version, namespace_path=None))]
|
||||
#[pyo3(signature = (name, version=None, namespace_path=None))]
|
||||
pub fn get_function(
|
||||
self_: PyRef<'_, Self>,
|
||||
name: String,
|
||||
version: String,
|
||||
version: Option<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, &namespace_path)
|
||||
.await
|
||||
.infer_error()?
|
||||
.to_canonical_json()
|
||||
.infer_error()
|
||||
let function = match version {
|
||||
Some(version) => inner.get_function(name, version, &namespace_path).await,
|
||||
None => inner.get_function_latest(name, &namespace_path).await,
|
||||
}
|
||||
.infer_error()?;
|
||||
function.to_canonical_json().infer_error()
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -668,9 +668,10 @@ impl Connection {
|
||||
|
||||
/// Look up one exact immutable Function version in the remote catalog.
|
||||
///
|
||||
/// Both the logical name and server-assigned version id are required;
|
||||
/// mutable aliases and latest-version lookup are intentionally absent.
|
||||
/// LanceDB Embedded databases return [`Error::NotSupported`].
|
||||
/// Both the logical name and server-assigned version id are required.
|
||||
/// Use [`Self::get_function_latest`] when the caller wants the latest
|
||||
/// currently visible revision. LanceDB Embedded databases return
|
||||
/// [`Error::NotSupported`].
|
||||
pub async fn get_function(
|
||||
&self,
|
||||
name: impl AsRef<str>,
|
||||
@@ -682,6 +683,22 @@ impl Connection {
|
||||
.await
|
||||
}
|
||||
|
||||
/// Look up the latest currently visible immutable Function version.
|
||||
///
|
||||
/// The server resolves and returns one exact version document. This is a
|
||||
/// resolution point, not a mutable alias: later registrations do not
|
||||
/// change the returned object. LanceDB Embedded databases return
|
||||
/// [`Error::NotSupported`].
|
||||
pub async fn get_function_latest(
|
||||
&self,
|
||||
name: impl AsRef<str>,
|
||||
namespace_path: &[String],
|
||||
) -> Result<crate::function::FunctionVersion> {
|
||||
self.internal
|
||||
.get_function_latest(name.as_ref(), namespace_path)
|
||||
.await
|
||||
}
|
||||
|
||||
/// List every published immutable Function version in `namespace_path` of
|
||||
/// the remote catalog. Functions in child namespaces are not included.
|
||||
///
|
||||
|
||||
@@ -552,6 +552,14 @@ pub trait Database:
|
||||
) -> Result<crate::function::FunctionVersion> {
|
||||
function_catalog_not_supported()
|
||||
}
|
||||
/// Look up the latest currently visible immutable Function version.
|
||||
async fn get_function_latest(
|
||||
&self,
|
||||
_name: &str,
|
||||
_namespace_path: &[String],
|
||||
) -> Result<crate::function::FunctionVersion> {
|
||||
function_catalog_not_supported()
|
||||
}
|
||||
/// Fetch one page of published immutable Function versions in one namespace.
|
||||
async fn list_functions(
|
||||
&self,
|
||||
|
||||
@@ -887,7 +887,20 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
|
||||
validate_function_name(name)?;
|
||||
validate_namespace(namespace_path)?;
|
||||
self.sql_client()?
|
||||
.describe_function(name, version, namespace_path)
|
||||
.describe_function(name, Some(version), namespace_path)
|
||||
.await
|
||||
.map_err(direct_catalog_resource_error)
|
||||
}
|
||||
|
||||
async fn get_function_latest(
|
||||
&self,
|
||||
name: &str,
|
||||
namespace_path: &[String],
|
||||
) -> Result<FunctionVersion> {
|
||||
validate_function_name(name)?;
|
||||
validate_namespace(namespace_path)?;
|
||||
self.sql_client()?
|
||||
.describe_function(name, None, namespace_path)
|
||||
.await
|
||||
.map_err(direct_catalog_resource_error)
|
||||
}
|
||||
|
||||
@@ -51,17 +51,20 @@ impl SqlClient {
|
||||
pub(in crate::remote) async fn describe_function(
|
||||
&self,
|
||||
name: &str,
|
||||
version: &str,
|
||||
version: Option<&str>,
|
||||
namespace: &[String],
|
||||
) -> Result<FunctionVersion> {
|
||||
validate_function_name(name)?;
|
||||
validate_namespace(namespace)?;
|
||||
let version_clause = match version {
|
||||
Some(version) => format!(" WITH (version = {})", literal(version)),
|
||||
None => String::new(),
|
||||
};
|
||||
self.object_job(
|
||||
ExecuteQueryRequest {
|
||||
query: format!(
|
||||
"DESCRIBE FUNCTION {} WITH (version = {})",
|
||||
quote_identifier(name),
|
||||
literal(version)
|
||||
"DESCRIBE FUNCTION {}{version_clause}",
|
||||
quote_identifier(name)
|
||||
),
|
||||
default_namespace_path: namespace.to_vec(),
|
||||
parameters: None,
|
||||
|
||||
@@ -3049,6 +3049,13 @@ async fn function_creation_and_description_use_sql_query_handles() {
|
||||
assert_eq!(described_json["batch_size"], 12);
|
||||
assert_eq!(described_json["row_timeout_seconds"], 8.0);
|
||||
assert_eq!(described_json["result_stability"], "input_and_context");
|
||||
let latest = db
|
||||
.get_function_latest("embed", &["team".into()])
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(latest.name(), "embed");
|
||||
assert_eq!(latest.version(), "1");
|
||||
assert_eq!(latest.batch_size(), Some(12));
|
||||
let listed = db
|
||||
.list_functions(&["team".into()], Default::default())
|
||||
.await
|
||||
@@ -3068,6 +3075,7 @@ async fn function_creation_and_description_use_sql_query_handles() {
|
||||
assert_eq!(listed_json["result_stability"], "input_and_context");
|
||||
let seen = statements.lock().unwrap();
|
||||
assert_eq!(seen[1], "DESCRIBE FUNCTION \"embed\" WITH (version = '1')");
|
||||
assert_eq!(seen[2], "DESCRIBE FUNCTION \"embed\"");
|
||||
shutdown.send(()).unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -156,11 +156,15 @@ async fn local_function_catalog_operations_return_stable_not_supported() {
|
||||
.get_function("normalize_score", "1", &[])
|
||||
.await
|
||||
.unwrap_err();
|
||||
let latest_lookup_error = connection
|
||||
.get_function_latest("normalize_score", &[])
|
||||
.await
|
||||
.unwrap_err();
|
||||
let drop_error = connection
|
||||
.drop_function("normalize_score", "1", &[])
|
||||
.await
|
||||
.unwrap_err();
|
||||
for error in [create_error, lookup_error, drop_error] {
|
||||
for error in [create_error, lookup_error, latest_lookup_error, drop_error] {
|
||||
assert!(matches!(
|
||||
error,
|
||||
Error::NotSupported { message }
|
||||
|
||||
Reference in New Issue
Block a user