adapt typed job decoding to current errors

python/python/lancedb/_lancedb.pyi

@@ -235,6 +235,12 @@ class FunctionJob:
     async def wait(self) -> str: ...
     async def cancel(self) -> None: ...
 
+class RefreshJob:
+    id: Optional[str]
+    async def status(self) -> str: ...
+    async def wait(self) -> str: ...
+    async def cancel(self) -> None: ...
+
 class JobInfo:
     @property
     def job_id(self) -> str: ...
@@ -354,7 +360,7 @@ class Table:
         self, application_json: str, output_name: Optional[str]
     ) -> AddColumnsResult: ...
     async def refresh_column(self, column: str) -> RefreshColumnResult: ...
-    async def refresh_column_async(self, column: str) -> Job: ...
+    async def refresh_column_async(self, column: str) -> RefreshJob: ...
     async def add_columns_with_schema(self, schema: pa.Schema) -> AddColumnsResult: ...
     async def alter_columns(
         self, columns: list[dict[str, Any]]

python/python/lancedb/expr.py

@@ -85,8 +85,9 @@ class Expr:
     # for dict keys / set membership.
     __hash__ = None  # type: ignore[assignment]
 
-    def __init__(self, inner: PyExpr) -> None:
+    def __init__(self, inner: PyExpr, *, column_path: str | None = None) -> None:
         self._inner = inner
+        self._column_path = column_path
 
     # ── comparisons ──────────────────────────────────────────────────────────
 
@@ -273,7 +274,7 @@ def col(name: str) -> Expr:
     >>> col("age") > lit(18)
     Expr((age > 18))
     """
-    return Expr(expr_col(name))
+    return Expr(expr_col(name), column_path=name)
 
 
 def lit(value: Union[bool, int, float, str, bytes, date, datetime, Decimal]) -> Expr:

python/python/lancedb/functions.py

@@ -20,6 +20,7 @@ import re
 import sys
 import textwrap
 import types
+import uuid
 from collections.abc import Mapping
 from datetime import date, datetime
 from typing import (
@@ -265,6 +266,42 @@ class FunctionVersion(_RemoteValue):
     required_secrets: tuple[str, ...] = ()
     created_at: str
 
+    def __call__(self, **inputs: Any) -> FunctionApplication:
+        """Bind this exact version to table columns as one grouped application."""
+        from lancedb.expr import Expr
+
+        parameters = tuple(parameter.name for parameter in self.signature.inputs)
+        missing = [parameter for parameter in parameters if parameter not in inputs]
+        unknown = sorted(set(inputs) - set(parameters))
+        if missing or unknown:
+            details = []
+            if missing:
+                details.append(f"missing inputs: {missing!r}")
+            if unknown:
+                details.append(f"unknown inputs: {unknown!r}")
+            raise TypeError("invalid Function inputs (" + "; ".join(details) + ")")
+
+        bindings = []
+        for parameter in parameters:
+            value = inputs[parameter]
+            if not isinstance(value, Expr) or value._column_path is None:
+                raise TypeError(
+                    f"Function input {parameter!r} must be a direct col(...) reference"
+                )
+            bindings.append(
+                ApplicationInput(
+                    parameter=parameter,
+                    kind="column",
+                    value={"path": value._column_path},
+                )
+            )
+        return FunctionApplication(
+            function=FunctionVersionRef(name=self.name, version=self.version),
+            inputs=tuple(bindings),
+            output=self.signature.output,
+            group_id=f"fg_{uuid.uuid4().hex}",
+        )
+
 
 class FunctionRegistrationRequest(_RemoteValue):
     """Stable remote registration envelope produced by :func:`udf`.

python/python/lancedb/job.py

@@ -130,3 +130,27 @@ class _FunctionJobAdapter:
 
 def _function_job(inner: "_lancedb.FunctionJob") -> AsyncJob[FunctionVersion]:
     return AsyncJob(_FunctionJobAdapter(inner))
+
+
+class _RefreshJobAdapter:
+    def __init__(self, inner: "_lancedb.RefreshJob"):
+        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):
+        from .functions import RefreshColumnResult
+
+        return RefreshColumnResult.from_json(await self._inner.wait())
+
+    async def cancel(self):
+        await self._inner.cancel()
+
+
+def _refresh_job(inner: "_lancedb.RefreshJob") -> AsyncJob:
+    return AsyncJob(_RefreshJobAdapter(inner))

python/python/lancedb/table.py

@@ -6148,7 +6148,9 @@ class AsyncTable:
         >>> asyncio.run(refresh_in_background())
         'finished'
         """
-        return AsyncJob(await self._inner.refresh_column_async(column))
+        from .job import _refresh_job
+
+        return _refresh_job(await self._inner.refresh_column_async(column))
 
     async def alter_columns(
         self, *alterations: Iterable[dict[str, Any]]

python/src/job.rs

@@ -22,6 +22,12 @@ pub struct FunctionJob {
     inner: Arc<lancedb::Job<lancedb::function::FunctionVersion>>,
 }
 
+/// Python bridge for a typed remote Function-column refresh job.
+#[pyclass]
+pub struct RefreshJob {
+    inner: Arc<lancedb::Job>,
+}
+
 impl FunctionJob {
     pub(crate) fn new(inner: lancedb::Job<lancedb::function::FunctionVersion>) -> Self {
         Self {
@@ -38,6 +44,56 @@ impl Job {
     }
 }
 
+impl RefreshJob {
+    pub(crate) fn new(inner: lancedb::Job) -> Self {
+        Self {
+            inner: Arc::new(inner),
+        }
+    }
+}
+
+#[pymethods]
+impl RefreshJob {
+    #[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 {
+            let value = inner.wait_raw_json().await.infer_error()?.ok_or_else(|| {
+                pyo3::exceptions::PyValueError::new_err(
+                    "successful refresh job did not contain a result",
+                )
+            })?;
+            let result: lancedb::function::RefreshColumnResult = serde_json::from_value(value)
+                .map_err(|error| {
+                    pyo3::exceptions::PyValueError::new_err(format!(
+                        "invalid refresh job result: {error}"
+                    ))
+                })?;
+            result.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(())
+        })
+    }
+}
+
 #[pymethods]
 impl FunctionJob {
     #[getter]

python/src/lib.rs

@@ -48,6 +48,7 @@ pub fn _lancedb(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_class::<Table>()?;
     m.add_class::<crate::job::Job>()?;
     m.add_class::<crate::job::FunctionJob>()?;
+    m.add_class::<crate::job::RefreshJob>()?;
     m.add_class::<crate::job::JobInfo>()?;
     m.add_class::<crate::job::JobDescription>()?;
     m.add_class::<crate::job::JobFailureInfo>()?;

python/src/table.rs

@@ -1584,7 +1584,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::RefreshJob::new(job))
         })
     }
 

rust/lancedb/src/job.rs

@@ -54,15 +54,15 @@ 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(|| Error::Other {
+            message: format!(
+                "successful typed job response did not contain a result (request_id={request_id})"
+            ),
+            source: None,
         })?;
-        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| Error::Other {
+            message: format!("failed to parse typed job result (request_id={request_id})"),
+            source: Some(Box::new(error)),
         })
     }
 }
@@ -120,6 +120,15 @@ impl Job<()> {
     pub(crate) fn spawned(task: JoinHandle<Result<()>>) -> Self {
         Self::new(Box::new(SpawnedJob::new(task)))
     }
+
+    /// Wait for an untyped job while preserving a remote terminal payload.
+    #[doc(hidden)]
+    pub async fn wait_raw_json(&self) -> Result<Option<Value>> {
+        match &self.inner {
+            JobInner::Handle { handle, .. } => Ok(handle.wait().await?.value),
+            JobInner::Completed(()) => Ok(None),
+        }
+    }
 }
 
 impl<T> Job<T>