feat: return a cleanup job from drop_function

Dropping a Function now deletes the object it was bound to, so the name drop
and the content deletion are separate events the caller may want to wait on.
`drop_function_async` returns the unbind result alongside a job for the
cleanup, the same shape `drop_table_async` and
`drop_materialized_view_async` already use: a 202 carries the job id, and a
200 -- an inline deletion, or a name that was not bound -- yields an
already-finished job.

python/python/lancedb/_lancedb.pyi

@@ -153,6 +153,9 @@ class Connection(object):
     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 drop_function_async(
+        self, name: str, version: str
+    ) -> Tuple[bool, Job]: ...
     async def create_secret(
         self, name: str, value: str, namespace_path: Optional[List[str]] = None
     ) -> None: ...

python/python/lancedb/db.py

@@ -18,6 +18,7 @@ from typing import (
     Literal,
     Optional,
     Sequence,
+    Tuple,
     Union,
 )
 from uuid import UUID
@@ -828,12 +829,26 @@ class DBConnection(EnforceOverrides):
         )
 
     def drop_function(self, name: str, *, version: str) -> bool:
-        """Remove the current Function name binding from the remote catalog.
+        """Drop a Function name and the object it was bound to.
 
-        The requested version must exist in the currently named object.
-        Object history and existing computed-column references are retained.
-        Returns True when the name was removed and False when it was absent.
-        Local connections raise NotImplementedError.
+        The requested version must exist in the currently named object. Returns
+        True when the name was removed and False when it was absent. The
+        object's content is deleted with it, which may finish after this
+        returns; use :meth:`drop_function_async` to wait for that. Local
+        connections raise NotImplementedError.
+        """
+        raise NotImplementedError(
+            "Function catalog operations are not supported for this connection type"
+        )
+
+    def drop_function_async(self, name: str, *, version: str) -> 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
+        be being deleted. Call :meth:`Job.wait` to wait for that to finish. When
+        the server deletes inline, or when nothing was bound, the returned job
+        is already finished and has no id. Local connections raise
+        NotImplementedError.
         """
         raise NotImplementedError(
             "Function catalog operations are not supported for this connection type"
@@ -1683,6 +1698,11 @@ class LanceDBConnection(DBConnection):
     def drop_function(self, name: str, *, version: str) -> bool:
         return LOOP.run(self._conn.drop_function(name, version=version))
 
+    @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))
+        return dropped, Job(job)
+
     @override
     def create_secret(
         self, name: str, value: str, *, namespace_path: Optional[List[str]] = None
@@ -2596,9 +2616,21 @@ class AsyncConnection(object):
         ]
 
     async def drop_function(self, name: str, *, version: str) -> bool:
-        """Remove the current name binding, retaining the object and its history."""
+        """Drop a Function name and the object it was bound to."""
         return await self._inner.drop_function(name, version)
 
+    async def drop_function_async(
+        self, name: str, *, version: str
+    ) -> Tuple[bool, AsyncJob]:
+        """Drop a Function name and return its cleanup job.
+
+        The name is unbound before this returns; the object's content may still
+        be being deleted. Await :meth:`AsyncJob.wait` to wait for that to
+        finish.
+        """
+        dropped, job = await self._inner.drop_function_async(name, version)
+        return dropped, AsyncJob(job)
+
     async def create_secret(
         self, name: str, value: str, *, namespace_path: Optional[List[str]] = None
     ) -> None:

python/python/lancedb/remote/db.py

@@ -16,6 +16,7 @@ from typing import (
     List,
     Optional,
     Sequence,
+    Tuple,
     Union,
 )
 from urllib.parse import urlparse
@@ -876,6 +877,11 @@ class RemoteDBConnection(DBConnection):
     def drop_function(self, name: str, *, version: str) -> bool:
         return LOOP.run(self._conn.drop_function(name, version=version))
 
+    @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))
+        return dropped, Job(job)
+
     @override
     def create_secret(
         self, name: str, value: str, *, namespace_path: Optional[List[str]] = None

python/src/connection.rs

@@ -768,6 +768,21 @@ impl Connection {
         })
     }
 
+    pub fn drop_function_async(
+        self_: PyRef<'_, Self>,
+        name: String,
+        version: String,
+    ) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.get_inner()?.clone();
+        future_into_py(self_.py(), async move {
+            inner
+                .drop_function_async(name, version)
+                .await
+                .infer_error()
+                .map(|(dropped, job)| (dropped, crate::job::Job::new(job)))
+        })
+    }
+
     #[pyo3(signature = (name, value, namespace_path=None))]
     pub fn create_secret(
         self_: PyRef<'_, Self>,