feat: return a cleanup job from drop_view

Dropping a view unbinds its name and leaves the definition dataset to a cleanup
job, so the two are separate events a caller may want to wait on.
`drop_view_async` returns the job, the same shape `drop_materialized_view_async`
and `drop_function_async` already use: a 202 carries the job id, and a 200 --
nothing bound to the name -- yields an already-finished job.

`drop_view` waits on that job before returning, so a caller that does not want
to think about cleanup gets the stronger guarantee: when it returns, the
definition really is deleted. That differs from `drop_materialized_view` and
`drop_function`, which return as soon as the name is unbound; the view API is
the newer one and this is the semantic worth having.

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