fence remote function refresh submissions

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`.
@@ -704,7 +741,11 @@ def _package_source(function: Callable[..., Any]) -> bytes:
     if globals_source:
         parts.extend(["", *globals_source])
     parts.extend(["", function_source, ""])
-    return "\n".join(parts).encode("utf-8")
+    try:
+        import cloudpickle
+    except ImportError as error:
+        raise RuntimeError("@udf remote registration requires cloudpickle") from error
+    return cloudpickle.dumps(function)
 
 
 class UdfDefinition:

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

@@ -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>

rust/lancedb/src/remote/db.rs

@@ -494,7 +494,7 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
         &self,
         request: FunctionRegistrationRequest,
     ) -> Result<Job<FunctionVersion>> {
-        let req = self.client.post("/v1/function/create").json(&request);
+        let req = self.client.post("/v1/functions/create").json(&request);
         let (request_id, response) = self.client.send(req).await?;
         let response = self.client.check_response(&request_id, response).await?;
         let status = response.status();
@@ -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/function/describe")
+            .post("/v1/functions/get")
             .json(&serde_json::json!({
                 "name": name,
                 "version": version,
@@ -2489,7 +2489,7 @@ mod tests {
             include_str!("../../tests/fixtures/first_class_functions/v1/remote_function_job.json");
         let expected: serde_json::Value = serde_json::from_str(REQUEST).unwrap();
         let conn = Connection::new_with_handler(move |request| match request.url().path() {
-            "/v1/function/create" => {
+            "/v1/functions/create" => {
                 assert_eq!(request.method(), &reqwest::Method::POST);
                 let body: serde_json::Value =
                     serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()).unwrap();
@@ -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/function/describe");
+            assert_eq!(request.url().path(), "/v1/functions/get");
             let body: serde_json::Value =
                 serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()).unwrap();
             assert_eq!(

rust/lancedb/src/remote/table.rs

@@ -2866,8 +2866,7 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
         let mut body = serde_json::json!({ "column": column });
         self.apply_branch_body(&mut body);
         let request = self
-            .client
-            .post(&format!("/v1/table/{}/backfill_column", self.identifier))
+            .post_read(&format!("/v1/table/{}/backfill_column", self.identifier))
             .json(&body);
         let (request_id, response) = self.send(request, true).await?;
         let response = self.check_table_response(&request_id, response).await?;