feat: return a cleanup job from drop_view (#4262)

Dropping a view unbinds its name and leaves the definition dataset to a
server-side cleanup job, so the two are separate events a caller may
want to wait on.

`drop_view_async` returns that job — the same shape
`drop_materialized_view_async` and `drop_function_async` already use: a
`202` carries the job id, a `200` (nothing was bound to the name) yields
an already-finished job with no id, and any other success status is an
error rather than a silent no-op.

## `drop_view` waits

`drop_view` now awaits the job before returning, so a caller who does
not want to think about cleanup gets the stronger guarantee: when it
returns, the definition really is deleted.

That is deliberately **different** from `drop_materialized_view` and
`drop_function`, which return as soon as the name is unbound and
document that content may still be deleting. The view API is the newer
one, and waiting is the semantic worth having; the other two are left
alone rather than changing behaviour already released.

## Surfaces

`Database` trait, the remote client, `Connection`, and the Python and
Node bindings — matching where `drop_materialized_view_async` is already
exposed.

Four client tests cover the accepted case reporting its job id, the
nothing-bound case reporting a finished job, a `202` without a usable
`job_id`, and an unexpected success status.

python/python/lancedb/_lancedb.pyi

@@ -180,6 +180,9 @@ class Connection(object):
     async def drop_view(
         self, name: str, namespace_path: Optional[List[str]] = None
     ) -> None: ...
+    async def drop_view_async(
+        self, name: str, namespace_path: Optional[List[str]] = None
+    ) -> Job: ...
     async def list_views(
         self, namespace_path: Optional[List[str]] = None
     ) -> List[str]: ...

python/python/lancedb/db.py

@@ -983,15 +983,30 @@ class DBConnection(EnforceOverrides):
     def drop_view(
         self, name: str, *, namespace_path: Optional[List[str]] = None
     ) -> None:
-        """Drop a view.
+        """Drop a view and wait for its definition to be deleted.
 
-        The tables it reads are untouched: a view holds no rows of its own.
+        The tables it reads are untouched: a view holds no rows of its own. Use
+        :meth:`drop_view_async` to get the cleanup job instead of waiting on it.
         Local connections raise ``NotImplementedError``.
         """
         raise NotImplementedError(
             "View operations are not supported for this connection type"
         )
 
+    def drop_view_async(
+        self, name: str, *, namespace_path: Optional[List[str]] = None
+    ) -> "Job[None]":
+        """Start dropping a view and return the job deleting its definition.
+
+        The name is free before this returns. Call :meth:`Job.wait` to wait for
+        the definition dataset to be deleted. When nothing was bound to the
+        name, the returned job is already finished and has no id. Local
+        connections raise ``NotImplementedError``.
+        """
+        raise NotImplementedError(
+            "View operations are not supported for this connection type"
+        )
+
     def list_views(self, *, namespace_path: Optional[List[str]] = None) -> List[str]:
         """The names of the views in one namespace.
 
@@ -1836,6 +1851,14 @@ class LanceDBConnection(DBConnection):
     ) -> None:
         LOOP.run(self._conn.drop_view(name, namespace_path=namespace_path))
 
+    @override
+    def drop_view_async(
+        self, name: str, *, namespace_path: Optional[List[str]] = None
+    ) -> "Job[None]":
+        return Job(
+            LOOP.run(self._conn.drop_view_async(name, namespace_path=namespace_path))
+        )
+
     @override
     def list_views(self, *, namespace_path: Optional[List[str]] = None) -> List[str]:
         return LOOP.run(self._conn.list_views(namespace_path=namespace_path))
@@ -2821,9 +2844,30 @@ class AsyncConnection(object):
     async def drop_view(
         self, name: str, *, namespace_path: Optional[List[str]] = None
     ) -> None:
-        """Drop a view. The tables it reads are untouched."""
+        """Drop a view and wait for its definition to be deleted.
+
+        The tables it reads are untouched. Use :meth:`drop_view_async` to get
+        the cleanup job instead of waiting on it.
+        """
         await self._inner.drop_view(name, list(namespace_path or []))
 
+    async def drop_view_async(
+        self,
+        name: str,
+        *,
+        namespace_path: Optional[List[str]] = None,
+    ) -> AsyncJob[None]:
+        """Start dropping a view and return the job deleting its definition.
+
+        The name is free before this returns. Await :meth:`AsyncJob.wait` before
+        assuming the definition dataset is gone.
+        """
+        if namespace_path is None:
+            namespace_path = []
+        return AsyncJob(
+            await self._inner.drop_view_async(name, namespace_path=namespace_path)
+        )
+
     async def list_views(
         self, *, namespace_path: Optional[List[str]] = None
     ) -> List[str]:

python/python/lancedb/remote/db.py

@@ -931,6 +931,13 @@ class RemoteDBConnection(DBConnection):
     ) -> None:
         LOOP.run(self._conn.drop_view(name, namespace_path=namespace_path))
 
+    @override
+    def drop_view_async(
+        self, name: str, *, namespace_path: Optional[List[str]] = None
+    ) -> Job[None]:
+        job = LOOP.run(self._conn.drop_view_async(name, namespace_path=namespace_path))
+        return Job(job)
+
     @override
     def list_views(self, *, namespace_path: Optional[List[str]] = None) -> List[str]:
         return LOOP.run(self._conn.list_views(namespace_path=namespace_path))

python/src/connection.rs

@@ -936,6 +936,23 @@ impl Connection {
         })
     }
 
+    #[pyo3(signature = (name, namespace_path=None))]
+    pub fn drop_view_async(
+        self_: PyRef<'_, Self>,
+        name: 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
+                .drop_view_async(name, &namespace_path)
+                .await
+                .infer_error()
+                .map(crate::job::Job::new)
+        })
+    }
+
     #[pyo3(signature = (namespace_path=None))]
     pub fn list_views(
         self_: PyRef<'_, Self>,

python/src/runtime.rs

@@ -208,13 +208,12 @@ where
         let _guard = guard;
         fut.await;
     };
+    // Detaching is the point: dropping the `JoinHandle` leaves the task running, and the
+    // `OutstandingGuard` it carries is what keeps it visible to `shutdown`. Written as `drop`
+    // rather than `let _ =`, which reads as discarding an unpolled future.
     match runtime::Handle::try_current() {
-        Ok(handle) => {
-            let _ = handle.spawn(task);
-        }
-        Err(_) => {
-            let _ = get_runtime().spawn(task);
-        }
+        Ok(handle) => drop(handle.spawn(task)),
+        Err(_) => drop(get_runtime().spawn(task)),
     }
 }