rust/lancedb
1
0
Fork 0
mirror of https://github.com/lancedb/lancedb.git synced 2026-08-25 15:38:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: return typed refresh job results (#4013)

Browse Source 
## Problem

`refresh_column_async` returned a unit-result job even though durable
refresh jobs carry a canonical terminal result. Python callers could not
obtain row counts or source and published versions through the public
`Job` API, and local and remote refresh jobs exposed different result
semantics.

## Behavior

`refresh_column_async` now returns `Job[RefreshColumnResult]` for local
and remote tables. The general typed-job bridge binds each endpoint to
its public result model while preserving unit-result jobs and existing
status, wait, cancel, and timeout behavior. A local no-op refresh
reports no published version.

The Node.js API continues to resolve `wait()` as `void`; its binding
erases the Rust result type internally to preserve the existing public
contract.

## Ownership and integration boundary

LanceDB owns the language-neutral `Job<T>` contract and language-binding
decode. Sophon owns production and durable persistence of terminal
payloads. Sophon #7348 and #7378 now publish the canonical refresh
result for Function-backed and expression-backed refresh jobs,
respectively. The remote client fixture matches the merged server
schema; live deployment and end-to-end demo acceptance remain separate
rollout checks.
This commit is contained in:
Xuanwo
2026-08-25 00:01:32 +08:00
committed by GitHub
parent 242ade8017
commit b0dae5eb0b
23 changed files with 429 additions and 187 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nodejs/src/job.rs
+5 -2
View File
@@ -14,9 +14,12 @@ pub struct Job {
}
impl Job {
pub(crate) fn new(inner: lancedb::Job) -> Self {
pub(crate) fn new<T>(inner: lancedb::Job<T>) -> Self
where
T: Clone + Send + Sync + 'static,
{
Self {
inner: Arc::new(inner),
inner: Arc::new(inner.map(|_| ())),
}
}
}
python/python/lancedb/__init__.py
+1
View File
@@ -29,6 +29,7 @@ from .functions import (
FunctionRegistrationRequest as FunctionRegistrationRequest,
FunctionVersion as FunctionVersion,
PythonRuntimeSpec as PythonRuntimeSpec,
RefreshColumnResult as RefreshColumnResult,
UdfDefinition as UdfDefinition,
udf as udf,
)
python/python/lancedb/_lancedb.pyi
+2 -9
View File
@@ -147,7 +147,7 @@ class Connection(object):
limit: Optional[int],
) -> list[str]: ... # Deprecated: Use list_tables instead
def job(self, job_id: str) -> Job: ...
async def create_function_async(self, request_json: str) -> FunctionJob: ...
async def create_function_async(self, request_json: str) -> Job: ...
async def get_function(self, name: str, version: str) -> str: ...
async def list_jobs(self) -> List[JobInfo]: ...
async def get_job(self, job_id: str) -> Optional[JobDescription]: ...
@@ -234,14 +234,7 @@ class Job:
@property
def id(self) -> Optional[str]: ...
async def status(self) -> str: ...
async def wait(self) -> None: ...
async def cancel(self) -> None: ...
class FunctionJob:
@property
def id(self) -> Optional[str]: ...
async def status(self) -> str: ...
async def wait(self) -> str: ...
async def wait(self) -> Optional[str]: ...
async def cancel(self) -> None: ...
class JobInfo:
python/python/lancedb/db.py
+2 -2
View File
@@ -46,7 +46,7 @@ from lance_namespace.errors import NamespaceNotEmptyError, TableNotFoundError
from . import __version__
from ._lancedb import connect as lancedb_connect # type: ignore
from .functions import FunctionVersion, UdfDefinition
from .job import AsyncJob, Job, _function_job
from .job import AsyncJob, Job, _typed_job
from .materialized_view import (
AsyncMaterializedView,
MaterializedView,
@@ -2237,7 +2237,7 @@ class AsyncConnection(object):
inner = await self._inner.create_function_async(
definition.registration_request.to_canonical_json()
)
return _function_job(inner)
return _typed_job(inner, FunctionVersion.from_json)
async def get_function(self, name: str, *, version: str) -> FunctionVersion:
"""Open one exact immutable Function version from the remote catalog."""
python/python/lancedb/functions.py
+8 -2
View File
@@ -1,10 +1,12 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright The LanceDB Authors
"""Canonical values exchanged with LanceDB Enterprise Function services.
"""Canonical Function values exchanged with LanceDB Enterprise services.
These immutable models contain client/wire state only. Catalog persistence,
environment bake, secret resolution, and execution are owned by Sophon.
``RefreshColumnResult`` is also the backend-neutral result of a local
expression-backed refresh job.
"""
from __future__ import annotations
@@ -460,7 +462,11 @@ class FunctionBinding(_RemoteValue):
class RefreshColumnResult(_RemoteValue):
"""Terminal result of a remote Function-column refresh Job."""
"""Terminal result of an expression-backed or Function-backed refresh Job.
Local jobs produce this value in process. LanceDB Cloud and Enterprise
decode the same value from the durable server-job terminal payload.
"""
rows_assigned: _UInt64
rows_failed: _UInt64
python/python/lancedb/job.py
+28 -30
View File
@@ -5,12 +5,11 @@
import asyncio
from datetime import timedelta
from typing import Any, Generic, Optional, TypeVar, cast
from typing import Any, Callable, Generic, Optional, TypeVar, cast
from lancedb.background_loop import LOOP
from . import _lancedb
from .functions import FunctionVersion
T = TypeVar("T")
@@ -18,11 +17,18 @@ T = TypeVar("T")
class AsyncJob(Generic[T]):
"""A handle to an operation that may still be running.
The operation may already be complete when the handle is created.
The operation may already be complete when the handle is created. ``T``
is the endpoint's terminal result type; unit-result jobs resolve to
``None``.
"""
def __init__(self, inner: Optional[Any]):
def __init__(
self,
inner: Optional[Any],
result_decoder: Optional[Callable[[Any], T]] = None,
):
self._inner = inner
self._result_decoder = result_decoder
@property
def id(self) -> Optional[str]:
@@ -50,17 +56,21 @@ class AsyncJob(Generic[T]):
async def wait(self, timeout: Optional[timedelta] = None) -> T:
"""Wait until the operation reaches a terminal state.
Returns the endpoint's typed result, or ``None`` for a unit-result
job.
Raises `JobFailedError` if the operation failed, `JobCancelledError`
if it was cancelled, and `TimeoutError` if `timeout` elapses first.
"""
if self._inner is None:
return cast(T, None)
if timeout is None:
return cast(T, await self._inner.wait())
return cast(
T,
await asyncio.wait_for(self._inner.wait(), timeout.total_seconds()),
)
result = await self._inner.wait()
else:
result = await asyncio.wait_for(self._inner.wait(), timeout.total_seconds())
if self._result_decoder is not None:
return self._result_decoder(result)
return cast(T, result)
async def cancel(self):
"""Request cancellation. Cancelling a finished operation is a no-op."""
@@ -70,7 +80,7 @@ class AsyncJob(Generic[T]):
class Job(Generic[T]):
"""Synchronous counterpart of `AsyncJob`."""
"""Synchronous counterpart of `AsyncJob` with the same result type."""
def __init__(self, inner: Optional[AsyncJob[T]]):
self._inner = inner
@@ -96,6 +106,9 @@ class Job(Generic[T]):
def wait(self, timeout: Optional[timedelta] = None) -> T:
"""Block until the operation reaches a terminal state.
Returns the endpoint's typed result, or ``None`` for a unit-result
job.
Raises `JobFailedError` if the operation failed, `JobCancelledError`
if it was cancelled, and `TimeoutError` if `timeout` elapses first.
"""
@@ -110,23 +123,8 @@ class Job(Generic[T]):
LOOP.run(self._inner.cancel())
class _FunctionJobAdapter:
def __init__(self, inner: "_lancedb.FunctionJob"):
self._inner = inner
@property
def id(self) -> Optional[str]:
return self._inner.id
async def status(self) -> str:
return await self._inner.status()
async def wait(self) -> FunctionVersion:
return FunctionVersion.from_json(await self._inner.wait())
async def cancel(self):
await self._inner.cancel()
def _function_job(inner: "_lancedb.FunctionJob") -> AsyncJob[FunctionVersion]:
return AsyncJob(_FunctionJobAdapter(inner))
def _typed_job(
inner: "_lancedb.Job", result_decoder: Callable[[str], T]
) -> AsyncJob[T]:
"""Bind an internal JSON-producing job to its public result model."""
return AsyncJob(inner, result_decoder)
python/python/lancedb/remote/table.py
+2 -2
View File
@@ -49,7 +49,7 @@ from lancedb.index import (
LabelList,
)
from lancedb.job import Job
from lancedb.functions import FunctionApplication
from lancedb.functions import FunctionApplication, RefreshColumnResult
from lancedb.remote.db import LOOP
from lancedb.table import IndexConfigType, KNOWN_METRICS
import pyarrow as pa
@@ -972,7 +972,7 @@ class RemoteTable(Table):
def refresh_column(self, column: str):
return LOOP.run(self._table.refresh_column(column))
def refresh_column_async(self, column: str) -> Job:
def refresh_column_async(self, column: str) -> Job[RefreshColumnResult]:
return Job(LOOP.run(self._table.refresh_column_async(column)))
def alter_columns(
python/python/lancedb/table.py
+31 -8
View File
@@ -40,7 +40,7 @@ from ._blob import (
from .types import BlobMode
from lancedb.arrow import peek_reader
from lancedb.background_loop import LOOP, embedding_executor
from lancedb.job import AsyncJob, Job
from lancedb.job import AsyncJob, Job, _typed_job
from .dependencies import (
_check_for_hugging_face,
_check_for_lance,
@@ -72,7 +72,10 @@ from .index import (
FTS,
)
from .expr import Expr
from .functions import FunctionApplication
from .functions import (
FunctionApplication,
RefreshColumnResult as RefreshColumnJobResult,
)
from .merge import LanceMergeInsertBuilder
from .pydantic import LanceModel, model_to_dict
from .query import (
@@ -2039,7 +2042,7 @@ class Table(ABC):
"""
@abstractmethod
def refresh_column_async(self, column: str) -> Job:
def refresh_column_async(self, column: str) -> Job[RefreshColumnJobResult]:
"""
Like :meth:`refresh_column`, but returns a handle to the refresh job
instead of blocking until it completes.
@@ -2050,6 +2053,12 @@ class Table(ABC):
than failing the job. On local tables the job runs in-process; on
LanceDB Cloud and Enterprise it is the server's backfill job.
Returns
-------
Job[RefreshColumnResult]
A job whose successful ``wait`` returns row counts plus the source
and published table versions.
Examples
--------
>>> import lancedb
@@ -2058,7 +2067,9 @@ class Table(ABC):
>>> table.add_columns(computed={"doubled": "x * 2"})
AddColumnsResult(version=2)
>>> job = table.refresh_column_async("doubled")
>>> job.wait()
>>> result = job.wait()
>>> result.rows_assigned
2
>>> job.status()
'finished'
"""
@@ -4082,7 +4093,7 @@ class LanceTable(Table):
[`AsyncTable.refresh_column`][lancedb.AsyncTable.refresh_column]."""
return LOOP.run(self._table.refresh_column(column))
def refresh_column_async(self, column: str) -> Job:
def refresh_column_async(self, column: str) -> Job[RefreshColumnJobResult]:
"""Fill a computed column's unfilled rows, returning a handle to the
refresh job. See
[`Table.refresh_column_async`][lancedb.table.Table.refresh_column_async].
@@ -6122,7 +6133,9 @@ class AsyncTable:
"""
return await self._inner.refresh_column(column)
async def refresh_column_async(self, column: str) -> AsyncJob:
async def refresh_column_async(
self, column: str
) -> AsyncJob[RefreshColumnJobResult]:
"""
Like :meth:`refresh_column`, but returns a handle to the refresh job
instead of blocking until it completes.
@@ -6134,6 +6147,12 @@ class AsyncTable:
in-process; on LanceDB Cloud and Enterprise it is the server's
backfill job.
Returns
-------
AsyncJob[RefreshColumnResult]
A job whose successful ``wait`` returns row counts plus the source
and published table versions.
Examples
--------
>>> import asyncio
@@ -6143,12 +6162,16 @@ class AsyncTable:
... table = await db.create_table("computed_job_async_demo", [{"x": 1}])
... await table.add_columns(computed={"doubled": "x * 2"})
... job = await table.refresh_column_async("doubled")
... await job.wait()
... result = await job.wait()
... assert result.rows_assigned == 1
... return await job.status()
>>> asyncio.run(refresh_in_background())
'finished'
"""
return AsyncJob(await self._inner.refresh_column_async(column))
return _typed_job(
await self._inner.refresh_column_async(column),
RefreshColumnJobResult.from_json,
)
async def alter_columns(
self, *alterations: Iterable[dict[str, Any]]
python/python/tests/test_db.py
+2 -2
View File
@@ -774,7 +774,7 @@ def test_drop_table_async(tmp_db: lancedb.DBConnection):
job = tmp_db.drop_table_async("test")
assert job.id is None
assert job.status() == "finished"
job.wait()
assert job.wait() is None
assert tmp_db.table_names() == []
tmp_db.create_table("test", data=data)
@@ -790,7 +790,7 @@ async def test_drop_table_async_connection(tmp_db_async: lancedb.AsyncConnection
job = await tmp_db_async.drop_table_async("test")
assert job.id is None
assert await job.status() == "finished"
await job.wait()
assert await job.wait() is None
assert await tmp_db_async.table_names() == []
python/python/tests/test_first_class_function_slice2.py
+1 -1
View File
@@ -187,7 +187,7 @@ def _mock_remote_function_catalog():
"job_state": "DONE",
"result": state["version"],
}
elif self.path == "/v1/functions/get":
elif self.path == "/v1/functions/describe":
assert body == {
"name": "normalize_score",
"version": "fv_exact",
python/python/tests/test_index.py
+1 -1
View File
@@ -88,7 +88,7 @@ async def binary_table(db_async):
async def test_create_index_async_returns_done_job(some_table: AsyncTable):
job = await some_table.create_index_async("id", config=BTree())
assert job.id is None
await job.wait()
assert await job.wait() is None
assert len(await some_table.list_indices()) == 1
await job.cancel()
python/python/tests/test_remote_db.py
+75 -1
View File
@@ -875,11 +875,85 @@ def test_remote_create_index_async_returns_job():
table = db.create_table("test", [{"id": 1}])
job = table.create_index_async("id", config=BTree())
assert job.id == "job-1"
job.wait(timeout=timedelta(seconds=30))
assert job.wait(timeout=timedelta(seconds=30)) is None
assert len(describe_calls) == 2
job.cancel()
def test_remote_refresh_async_returns_typed_terminal_result():
terminal_result = {
"rows_assigned": 12,
"rows_failed": 0,
"rows_remaining": 0,
"source_version": 7,
"published_version": 8,
}
def handler(request):
content_len = int(request.headers.get("Content-Length", 0))
body = request.rfile.read(content_len) if content_len > 0 else b""
if request.path == "/v1/table/test/backfill_column":
assert json.loads(body)["column"] == "derived"
request.send_response(202)
request.send_header("Content-Type", "application/json")
request.end_headers()
request.wfile.write(b'{"job_id": "refresh-1"}')
elif request.path == "/v1/jobs/describe":
assert json.loads(body)["job_id"] == "refresh-1"
request.send_response(200)
request.send_header("Content-Type", "application/json")
request.end_headers()
request.wfile.write(
json.dumps(
{
"job_id": "refresh-1",
"job_type": "function_refresh",
"job_state": "DONE",
"result": terminal_result,
}
).encode()
)
elif request.path == "/v1/table/test/create/?mode=create":
request.send_response(200)
request.send_header("Content-Type", "application/json")
request.end_headers()
request.wfile.write(b"{}")
elif request.path == "/v1/table/test/describe/":
request.send_response(200)
request.send_header("Content-Type", "application/json")
request.end_headers()
request.wfile.write(
json.dumps(
{
"version": 1,
"schema": {
"fields": [
{
"name": "id",
"type": {"type": "int64"},
"nullable": False,
}
]
},
}
).encode()
)
else:
request.send_response(404)
request.end_headers()
with mock_lancedb_connection(handler) as db:
table = db.create_table("test", [{"id": 1}])
job = table.refresh_column_async("derived")
assert job.id == "refresh-1"
result = job.wait(timeout=timedelta(seconds=30))
assert isinstance(result, lancedb.RefreshColumnResult)
assert result.model_dump() == terminal_result
assert result.rows_filled == 12
assert result.version == 8
def test_remote_job_wait_raises_on_failure():
from lancedb.exceptions import JobFailedError
from lancedb.index import BTree
python/python/tests/test_table.py
+18 -3
View File
@@ -1467,7 +1467,7 @@ def test_create_index_async_returns_done_job(mem_db: DBConnection):
table = mem_db.create_table("job_test", [{"id": i} for i in range(10)])
job = table.create_index_async("id", config=BTree())
assert job.id is None
job.wait()
assert job.wait() is None
assert len(table.list_indices()) == 1
job.cancel()
@@ -3947,10 +3947,21 @@ def test_refresh_column_async_returns_job(tmp_path):
job = table.refresh_column_async("doubled")
assert job.id is None # in-process jobs have no server id
assert job.wait() is None
result = job.wait()
assert isinstance(result, lancedb.RefreshColumnResult)
assert result.rows_assigned == 2
assert result.rows_failed == 0
assert result.rows_remaining == 0
assert result.source_version == 2
assert result.published_version == 3
assert job.status() == "finished"
assert sorted(table.to_arrow()["doubled"].to_pylist()) == [2, 4]
no_op = table.refresh_column_async("doubled").wait()
assert no_op.rows_assigned == 0
assert no_op.source_version == 3
assert no_op.published_version is None
# Bad input raises at the call, not through the job.
with pytest.raises(Exception, match="not a computed column"):
table.refresh_column_async("x")
@@ -3963,6 +3974,10 @@ async def test_refresh_column_async_job_async_table(tmp_path):
await table.add_columns(computed={"tripled": "x * 3"})
job = await table.refresh_column_async("tripled")
assert await job.wait() is None
result = await job.wait()
assert isinstance(result, lancedb.RefreshColumnResult)
assert result.rows_assigned == 1
assert result.source_version == 2
assert result.published_version == 3
assert await job.status() == "finished"
assert (await table.to_arrow())["tripled"].to_pylist() == [9]
python/src/connection.rs
+1 -1
View File
@@ -609,7 +609,7 @@ impl Connection {
.create_function_async(request)
.await
.infer_error()
.map(crate::job::FunctionJob::new)
.map(crate::job::Job::new_typed)
})
}
python/src/job.rs
+18 -55
View File
@@ -5,72 +5,33 @@ use std::sync::Arc;
use crate::runtime::future_into_py;
use pyo3::{Bound, PyAny, PyRef, PyResult, pyclass, pymethods};
use serde::Serialize;
use crate::error::PythonErrorExt;
#[pyclass]
pub struct Job {
inner: Arc<lancedb::Job>,
}
/// Python bridge for a typed remote Function registration job.
///
/// The public Python layer decodes the canonical JSON returned by `wait`
/// into its immutable `FunctionVersion` model.
#[pyclass]
pub struct FunctionJob {
inner: Arc<lancedb::Job<lancedb::function::FunctionVersion>>,
}
impl FunctionJob {
pub(crate) fn new(inner: lancedb::Job<lancedb::function::FunctionVersion>) -> Self {
Self {
inner: Arc::new(inner),
}
}
inner: Arc<lancedb::Job<std::result::Result<Option<String>, String>>>,
}
impl Job {
pub(crate) fn new(inner: lancedb::Job) -> Self {
Self {
inner: Arc::new(inner),
inner: Arc::new(inner.map(|()| Ok(None))),
}
}
}
#[pymethods]
impl FunctionJob {
#[getter]
pub fn id(&self) -> Option<String> {
self.inner.id().map(str::to_string)
}
pub fn status(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner.clone();
future_into_py(
self_.py(),
async move { inner.status().await.infer_error() },
)
}
pub fn wait(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner.clone();
future_into_py(self_.py(), async move {
inner
.wait()
.await
.infer_error()?
.to_canonical_json()
.infer_error()
})
}
pub fn cancel(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner.clone();
future_into_py(self_.py(), async move {
inner.cancel().await.infer_error()?;
Ok(())
})
pub(crate) fn new_typed<T>(inner: lancedb::Job<T>) -> Self
where
T: Clone + Serialize + Send + Sync + 'static,
{
Self {
inner: Arc::new(inner.map(|result| {
serde_json::to_string(&result)
.map(Some)
.map_err(|error| format!("failed to serialize typed job result: {error}"))
})),
}
}
}
@@ -92,8 +53,10 @@ impl Job {
pub fn wait(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner.clone();
future_into_py(self_.py(), async move {
inner.wait().await.infer_error()?;
Ok(None::<()>)
let result = inner.wait().await.infer_error()?;
result
.map_err(|message| lancedb::Error::Runtime { message })
.infer_error()
})
}
python/src/lib.rs
-1
View File
@@ -47,7 +47,6 @@ pub fn _lancedb(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
m.add_class::<Session>()?;
m.add_class::<Table>()?;
m.add_class::<crate::job::Job>()?;
m.add_class::<crate::job::FunctionJob>()?;
m.add_class::<crate::job::JobInfo>()?;
m.add_class::<crate::job::JobDescription>()?;
m.add_class::<crate::job::JobFailureInfo>()?;
python/src/table.rs
+1 -1
View File
@@ -1619,7 +1619,7 @@ impl Table {
let inner = self_.inner_ref()?.clone();
future_into_py(self_.py(), async move {
let job = inner.refresh_column_async(column).await.infer_error()?;
Ok(crate::job::Job::new(job))
Ok(crate::job::Job::new_typed(job))
})
}
rust/lancedb/src/function.rs
+12 -2
View File
@@ -1,7 +1,8 @@
// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
//! Canonical values exchanged with the Enterprise Function service.
//! Canonical Function values exchanged with the Enterprise service, plus the
//! backend-neutral terminal result of a computed-column refresh.
//!
//! This module contains client/wire values only. Catalog persistence,
//! environment bake, secret resolution, and execution are owned by Sophon.
@@ -580,13 +581,22 @@ impl FunctionBinding {
impl_json!(FunctionBinding);
/// Stable terminal result of a remote Function-column refresh Job.
/// Stable terminal result of an expression-backed or Function-backed column
/// refresh [`crate::Job`].
///
/// Local refresh jobs produce this value in process. LanceDB Cloud and
/// Enterprise decode the same value from the durable job's terminal payload.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RefreshColumnResult {
/// Rows assigned a value by this refresh.
pub rows_assigned: u64,
/// Rows whose computation failed.
pub rows_failed: u64,
/// Rows that still need a value when the job completes.
pub rows_remaining: u64,
/// Exact table version the refresh read.
pub source_version: u64,
/// Table version made visible by the refresh, when one was published.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub published_version: Option<u64>,
}
rust/lancedb/src/job.rs
+115 -37
View File
@@ -6,7 +6,7 @@
use std::sync::Arc;
use async_trait::async_trait;
use serde::de::DeserializeOwned;
use serde::{Serialize, de::DeserializeOwned};
use serde_json::Value;
use tokio::sync::watch;
use tokio::task::{AbortHandle, JoinHandle};
@@ -26,20 +26,16 @@ pub(crate) trait JobHandle: Send + Sync {
}
/// A backend-neutral successful terminal result.
///
/// Local operations do not carry a value. Remote operations may carry JSON
/// that the public [`Job`] decodes according to its result type.
#[derive(Clone)]
pub(crate) struct TerminalResult {
#[allow(dead_code)] // Typed remote submit endpoints consume this after Slice 1.
value: Option<Value>,
#[allow(dead_code)] // Preserved so typed decode errors retain request correlation.
request_id: Option<String>,
}
impl TerminalResult {
pub(crate) fn local() -> Self {
fn local(value: Value) -> Self {
Self {
value: None,
value: Some(value),
request_id: None,
}
}
@@ -51,23 +47,31 @@ impl TerminalResult {
}
}
#[allow(dead_code)] // Exercised by the remote typed-result fixtures in Slice 1.
fn decode<T: DeserializeOwned>(self) -> Result<T> {
let request_id = self.request_id.unwrap_or_default();
let value = self.value.ok_or_else(|| Error::Http {
source: "successful typed job response did not contain a result".into(),
request_id: request_id.clone(),
status_code: None,
let value = self.value.ok_or_else(|| match &self.request_id {
Some(request_id) => Error::Http {
source: "successful typed job response did not contain a result".into(),
request_id: request_id.clone(),
status_code: None,
},
None => Error::Runtime {
message: "successful typed job did not contain a result".to_string(),
},
})?;
serde_json::from_value(value).map_err(|error| Error::Http {
source: format!("failed to parse typed job result: {error}").into(),
request_id,
status_code: None,
serde_json::from_value(value).map_err(|error| match self.request_id {
Some(request_id) => Error::Http {
source: format!("failed to parse typed job result: {error}").into(),
request_id,
status_code: None,
},
None => Error::Runtime {
message: format!("failed to parse typed job result: {error}"),
},
})
}
}
type ResultDecoder<T> = fn(TerminalResult) -> Result<T>;
type ResultDecoder<T> = Arc<dyn Fn(TerminalResult) -> Result<T> + Send + Sync>;
enum JobInner<T> {
Handle {
@@ -79,7 +83,9 @@ enum JobInner<T> {
/// A handle to an operation that may still be running.
///
/// The operation may already be complete when the handle is created.
/// The operation may already be complete when the handle is created. `T` is
/// the endpoint's successful terminal result; unit-result operations use the
/// default `Job<()>`.
pub struct Job<T = ()>
where
T: Clone + Send + Sync + 'static,
@@ -111,15 +117,10 @@ impl Job<()> {
Self {
inner: JobInner::Handle {
handle,
decode: |_| Ok(()),
decode: Arc::new(|_| Ok(())),
},
}
}
/// A unit-result job running as a task in this process.
pub(crate) fn spawned(task: JoinHandle<Result<()>>) -> Self {
Self::new(Box::new(SpawnedJob::new(task)))
}
}
impl<T> Job<T>
@@ -131,12 +132,22 @@ where
Self {
inner: JobInner::Handle {
handle,
decode: TerminalResult::decode::<T>,
decode: Arc::new(TerminalResult::decode::<T>),
},
}
}
}
impl<T> Job<T>
where
T: Clone + Serialize + DeserializeOwned + Send + Sync + 'static,
{
/// A typed job running as a task in this process.
pub(crate) fn spawned(task: JoinHandle<Result<T>>) -> Self {
Self::new_typed(Box::new(SpawnedJob::new(task)))
}
}
impl<T> Job<T>
where
T: Clone + Send + Sync + 'static,
@@ -169,11 +180,13 @@ where
/// Waits until the operation reaches a terminal state.
///
/// Returns the endpoint's typed result. Unit-result jobs return `()`.
///
/// Returns [`crate::Error::JobFailed`] if the operation failed and
/// [`crate::Error::JobCancelled`] if it was cancelled.
pub async fn wait(&self) -> Result<T> {
match &self.inner {
JobInner::Handle { handle, decode } => decode(handle.wait().await?),
JobInner::Handle { handle, decode } => (decode)(handle.wait().await?),
JobInner::Completed(result) => Ok(result.clone()),
}
}
@@ -187,21 +200,53 @@ where
JobInner::Completed(_) => Ok(()),
}
}
/// Maps a successful terminal result without changing the job lifecycle.
/// The mapping may run once for each call to [`Job::wait`], so it should
/// be deterministic and free of externally visible side effects.
///
/// ```
/// use lancedb::{Job, function::RefreshColumnResult};
///
/// # async fn rows_assigned(
/// # job: Job<RefreshColumnResult>,
/// # ) -> lancedb::Result<u64> {
/// let job = job.map(|result| result.rows_assigned);
/// job.wait().await
/// # }
/// ```
pub fn map<U, F>(self, map: F) -> Job<U>
where
U: Clone + Send + Sync + 'static,
F: Fn(T) -> U + Send + Sync + 'static,
{
match self.inner {
JobInner::Handle { handle, decode } => Job {
inner: JobInner::Handle {
handle,
decode: Arc::new(move |result| Ok(map((decode)(result)?))),
},
},
JobInner::Completed(result) => Job {
inner: JobInner::Completed(map(result)),
},
}
}
}
/// How an in-process operation ended. Cloneable so every waiter can be given
/// the outcome; [`Error`] is not, so failures share one behind an [`Arc`].
#[derive(Clone)]
enum Outcome {
Succeeded,
Succeeded(TerminalResult),
Failed(Arc<Error>),
Cancelled,
}
impl Outcome {
fn into_result(self) -> Result<()> {
fn into_result(self) -> Result<TerminalResult> {
match self {
Self::Succeeded => Ok(()),
Self::Succeeded(result) => Ok(result),
Self::Failed(source) => Err(Error::JobFailed {
job_id: None,
failure: JobFailure::from_source(source),
@@ -220,12 +265,20 @@ struct SpawnedJob {
}
impl SpawnedJob {
fn new(task: JoinHandle<Result<()>>) -> Self {
fn new<T>(task: JoinHandle<Result<T>>) -> Self
where
T: Serialize + Send + 'static,
{
let abort = task.abort_handle();
let (tx, outcome) = watch::channel(None);
tokio::spawn(async move {
let outcome = match task.await {
Ok(Ok(())) => Outcome::Succeeded,
Ok(Ok(result)) => match serde_json::to_value(result) {
Ok(value) => Outcome::Succeeded(TerminalResult::local(value)),
Err(err) => Outcome::Failed(Arc::new(Error::Runtime {
message: format!("failed to serialize job result: {err}"),
})),
},
Ok(Err(err)) => Outcome::Failed(Arc::new(err)),
Err(err) if err.is_cancelled() => Outcome::Cancelled,
Err(err) => Outcome::Failed(Arc::new(Error::Runtime {
@@ -243,7 +296,7 @@ impl JobHandle for SpawnedJob {
async fn status(&self) -> Result<String> {
let label = match &*self.outcome.borrow() {
None => "running",
Some(Outcome::Succeeded) => "finished",
Some(Outcome::Succeeded(_)) => "finished",
Some(Outcome::Failed(_)) => "failed",
Some(Outcome::Cancelled) => "cancelled",
};
@@ -256,12 +309,11 @@ impl JobHandle for SpawnedJob {
.wait_for(|outcome| outcome.is_some())
.await
.map_err(|_| Error::Runtime {
message: "index job outcome was dropped before it completed".to_string(),
message: "job outcome was dropped before it completed".to_string(),
})?
.clone()
.expect("wait_for returns once an outcome is set");
settled.into_result()?;
Ok(TerminalResult::local())
settled.into_result()
}
async fn cancel(&self) -> Result<()> {
@@ -269,3 +321,29 @@ impl JobHandle for SpawnedJob {
Ok(())
}
}
#[cfg(test)]
mod tests {
use std::future::pending;
use super::*;
#[tokio::test]
async fn mapped_spawned_job_reuses_outcome() {
let job = Job::spawned(tokio::spawn(async { Ok(41_u64) })).map(|value| value + 1);
assert_eq!(job.wait().await.unwrap(), 42);
assert_eq!(job.wait().await.unwrap(), 42);
assert_eq!(job.status().await.unwrap(), "finished");
}
#[tokio::test]
async fn mapped_spawned_job_preserves_cancellation() {
let job = Job::spawned(tokio::spawn(async { pending::<Result<u64>>().await }))
.map(|value| value.to_string());
job.cancel().await.unwrap();
assert!(matches!(job.wait().await, Err(Error::JobCancelled { .. })));
assert_eq!(job.status().await.unwrap(), "cancelled");
}
}
rust/lancedb/src/remote/db.rs
+2 -2
View File
@@ -513,7 +513,7 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
async fn get_function(&self, name: &str, version: &str) -> Result<FunctionVersion> {
let req = self
.client
.post("/v1/functions/get")
.post("/v1/functions/describe")
.json(&serde_json::json!({
"name": name,
"version": version,
@@ -2520,7 +2520,7 @@ mod tests {
);
let conn = Connection::new_with_handler(|request| {
assert_eq!(request.method(), &reqwest::Method::POST);
assert_eq!(request.url().path(), "/v1/functions/get");
assert_eq!(request.url().path(), "/v1/functions/describe");
let body: serde_json::Value =
serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()).unwrap();
assert_eq!(
rust/lancedb/src/remote/table.rs
+28 -8
View File
@@ -2864,7 +2864,10 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
})
}
async fn refresh_column_async(&self, column: &str) -> Result<Job> {
async fn refresh_column_async(
&self,
column: &str,
) -> Result<Job<crate::function::RefreshColumnResult>> {
self.check_mutable().await?;
let mut body = serde_json::json!({ "column": column });
self.apply_branch_body(&mut body);
@@ -2885,7 +2888,7 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
status_code: None,
})?;
Ok(Job::new(Box::new(FreshnessJob {
Ok(Job::new_typed(Box::new(FreshnessJob {
inner: RemoteJob::new(self.client.clone(), response.job_id),
freshness: self.freshness.clone(),
version: self.version.clone(),
@@ -3280,6 +3283,21 @@ mod tests {
},
};
fn refresh_done(job_id: &str) -> String {
json!({
"job_id": job_id,
"job_state": "DONE",
"result": {
"rows_assigned": 12,
"rows_failed": 0,
"rows_remaining": 0,
"source_version": 7,
"published_version": 8,
}
})
.to_string()
}
#[tokio::test]
async fn test_not_found() {
let table = Table::new_with_handler("my_table", |_| {
@@ -6892,7 +6910,7 @@ mod tests {
.unwrap(),
"/v1/jobs/describe" => http::Response::builder()
.status(200)
.body(r#"{"job_id": "j-7", "job_state": "DONE"}"#.to_string())
.body(refresh_done("j-7"))
.unwrap(),
"/v1/table/my_table/count_rows/" => {
saw.store(
@@ -6908,7 +6926,9 @@ mod tests {
});
let job = table.refresh_column_async("doubled").await.unwrap();
job.wait().await.unwrap();
let result = job.wait().await.unwrap();
assert_eq!(result.rows_assigned, 12);
assert_eq!(result.published_version, Some(8));
table.count_rows(None).await.unwrap();
assert!(
saw_min_timestamp.load(std::sync::atomic::Ordering::SeqCst),
@@ -6930,7 +6950,7 @@ mod tests {
.unwrap(),
"/v1/jobs/describe" => http::Response::builder()
.status(200)
.body(r#"{"job_id": "j-8", "job_state": "DONE"}"#.to_string())
.body(refresh_done("j-8"))
.unwrap(),
"/v1/table/my_table/describe/" => {
let schema = Schema::new(vec![Field::new("x", DataType::Int32, true)]);
@@ -6976,7 +6996,7 @@ mod tests {
.unwrap(),
"/v1/jobs/describe" => http::Response::builder()
.status(200)
.body(r#"{"job_id": "j-9", "job_state": "DONE"}"#.to_string())
.body(refresh_done("j-9"))
.unwrap(),
"/v1/table/my_table/tags/version/" => http::Response::builder()
.status(200)
@@ -7040,7 +7060,7 @@ mod tests {
}
"/v1/jobs/describe" => http::Response::builder()
.status(200)
.body(r#"{"job_id": "j-10", "job_state": "DONE"}"#.to_string())
.body(refresh_done("j-10"))
.unwrap(),
"/v1/table/my_table/describe/" => {
let schema = Schema::new(vec![Field::new("x", DataType::Int32, true)]);
@@ -7113,7 +7133,7 @@ mod tests {
}
"/v1/jobs/describe" => http::Response::builder()
.status(200)
.body(r#"{"job_id": "j-11", "job_state": "DONE"}"#.to_string())
.body(refresh_done("j-11"))
.unwrap(),
"/v1/table/my_table/count_rows/" => {
*saw.lock().unwrap() = request
rust/lancedb/src/table.rs
+17 -5
View File
@@ -771,7 +771,10 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
}
/// Fill a computed column's unfilled rows, returning a [`Job`] tracking
/// the operation.
async fn refresh_column_async(&self, _column: &str) -> Result<Job> {
async fn refresh_column_async(
&self,
_column: &str,
) -> Result<Job<crate::function::RefreshColumnResult>> {
Err(Error::NotSupported {
message: "computed columns are supported only on local tables".into(),
})
@@ -1708,7 +1711,9 @@ impl Table {
/// operation instead of blocking until it completes.
///
/// The job may already be complete when returned, and callers must not
/// assume the column is filled until [`Job::wait`] returns. Invalid input
/// assume the column is filled until [`Job::wait`] returns. A successful
/// wait returns the durable [`crate::function::RefreshColumnResult`] for
/// both expression-backed and Function-backed columns. Invalid input
/// -- an unknown column, or one that is not computed -- is reported by
/// this call rather than by the job. On local tables the job runs as an
/// in-process task; on LanceDB Cloud and Enterprise it is the server's
@@ -1719,11 +1724,15 @@ impl Table {
/// # async fn refresh_in_background(table: &Table) -> Result<(), Box<dyn std::error::Error>> {
/// let job = table.refresh_column_async("doubled").await?;
/// println!("refresh running: {:?}", job.status().await?);
/// job.wait().await?;
/// let result = job.wait().await?;
/// println!("assigned {} rows", result.rows_assigned);
/// # Ok(())
/// # }
/// ```
pub async fn refresh_column_async(&self, column: impl AsRef<str>) -> Result<Job> {
pub async fn refresh_column_async(
&self,
column: impl AsRef<str>,
) -> Result<Job<crate::function::RefreshColumnResult>> {
self.inner.refresh_column_async(column.as_ref()).await
}
@@ -3425,7 +3434,10 @@ impl BaseTable for NativeTable {
Ok(result)
}
async fn refresh_column_async(&self, column: &str) -> Result<Job> {
async fn refresh_column_async(
&self,
column: &str,
) -> Result<Job<crate::function::RefreshColumnResult>> {
refresh::execute_refresh_column_async(self, column).await
}
rust/lancedb/src/table/refresh.rs
+59 -12
View File
@@ -49,11 +49,25 @@ pub struct RefreshColumnResult {
pub version: u64,
}
struct RefreshExecution {
result: RefreshColumnResult,
source_version: u64,
}
/// Internal implementation of the refresh logic.
pub(crate) async fn execute_refresh_column(
table: &NativeTable,
column: &str,
) -> Result<RefreshColumnResult> {
Ok(execute_refresh_column_with_source(table, column)
.await?
.result)
}
async fn execute_refresh_column_with_source(
table: &NativeTable,
column: &str,
) -> Result<RefreshExecution> {
table.dataset.ensure_mutable()?;
ensure_no_lsm_write_spec(table).await?;
let dataset = table.dataset.get().await?;
@@ -87,9 +101,13 @@ pub(crate) async fn execute_refresh_column(
}
if replacements.is_empty() {
return Ok(RefreshColumnResult {
rows_filled: 0,
version: dataset.version().version,
let source_version = dataset.version().version;
return Ok(RefreshExecution {
result: RefreshColumnResult {
rows_filled: 0,
version: source_version,
},
source_version,
});
}
@@ -110,14 +128,20 @@ pub(crate) async fn execute_refresh_column(
let version = new_dataset.version().version;
table.dataset.update(new_dataset);
Ok(RefreshColumnResult {
rows_filled,
version,
Ok(RefreshExecution {
result: RefreshColumnResult {
rows_filled,
version,
},
source_version: read_version,
})
}
/// Run the refresh as a [`Job`] in this process.
pub(crate) async fn execute_refresh_column_async(table: &NativeTable, column: &str) -> Result<Job> {
pub(crate) async fn execute_refresh_column_async(
table: &NativeTable,
column: &str,
) -> Result<Job<crate::function::RefreshColumnResult>> {
// Validate before spawning so bad input is reported by this call rather
// than only by the job.
table.dataset.ensure_mutable()?;
@@ -129,9 +153,16 @@ pub(crate) async fn execute_refresh_column_async(table: &NativeTable, column: &s
let table = table.clone();
let column = column.to_string();
Ok(Job::spawned(tokio::spawn(async move {
execute_refresh_column(&table, &column).await?;
let execution = execute_refresh_column_with_source(&table, &column).await?;
table.bump_freshness();
Ok(())
Ok(crate::function::RefreshColumnResult {
rows_assigned: execution.result.rows_filled,
rows_failed: 0,
rows_remaining: 0,
source_version: execution.source_version,
published_version: (execution.result.rows_filled > 0)
.then_some(execution.result.version),
})
})))
}
@@ -396,6 +427,17 @@ mod tests {
read(&table, "doubled").await,
vec![Some(2), Some(4), Some(6)]
);
let no_op = table
.refresh_column_async("doubled")
.await
.unwrap()
.wait()
.await
.unwrap();
assert_eq!(no_op.rows_assigned, 0);
assert_eq!(no_op.source_version, 3);
assert_eq!(no_op.published_version, None);
}
/// Values written after the last refresh must be reachable by another one.
@@ -646,7 +688,12 @@ mod tests {
let job = table.refresh_column_async("doubled").await.unwrap();
assert!(job.id().is_none(), "in-process jobs have no server id");
job.wait().await.unwrap();
let result = job.wait().await.unwrap();
assert_eq!(result.rows_assigned, 3);
assert_eq!(result.rows_failed, 0);
assert_eq!(result.rows_remaining, 0);
assert_eq!(result.source_version, 2);
assert_eq!(result.published_version, Some(3));
assert_eq!(job.status().await.unwrap(), "finished");
assert_eq!(
read(&table, "doubled").await,
@@ -672,9 +719,9 @@ mod tests {
declare_doubled(&table).await.unwrap();
let job = table.refresh_column_async("doubled").await.unwrap();
job.wait().await.unwrap();
let first = job.wait().await.unwrap();
// A second wait after completion observes the same outcome.
job.wait().await.unwrap();
assert_eq!(job.wait().await.unwrap(), first);
assert_eq!(job.status().await.unwrap(), "finished");
}