diff --git a/docs/src/js/classes/Connection.md b/docs/src/js/classes/Connection.md index fa4e0748a..e4cbc1e96 100644 --- a/docs/src/js/classes/Connection.md +++ b/docs/src/js/classes/Connection.md @@ -386,6 +386,29 @@ Drop an existing table. *** +### dropTableAsync() + +```ts +abstract dropTableAsync(name, namespacePath?): Promise +``` + +Start dropping a table and return its cleanup job. + +The table may become unavailable before its data files are removed. Wait +on the returned job to know when cleanup has finished. + +#### Parameters + +* **name**: `string` + +* **namespacePath?**: `string`[] + +#### Returns + +`Promise`<[`Job`](Job.md)> + +*** + ### getJob() ```ts diff --git a/nodejs/__test__/connection.test.ts b/nodejs/__test__/connection.test.ts index 68180471a..af471b478 100644 --- a/nodejs/__test__/connection.test.ts +++ b/nodejs/__test__/connection.test.ts @@ -89,6 +89,16 @@ describe("given a connection", () => { await db.createTable("test4", [{ id: 1 }, { id: 2 }]); }); + it("should return a completed job when dropping a local table", async () => { + await db.createTable("async-drop", [{ id: 1 }]); + + const job = await db.dropTableAsync("async-drop"); + expect(job.id).toBeNull(); + await expect(job.status()).resolves.toBe("finished"); + await job.wait(); + await expect(db.tableNames()).resolves.toEqual([]); + }); + it("should fail if creating table twice, unless overwrite is true", async () => { let tbl = await db.createTable("test", [{ id: 1 }, { id: 2 }]); await expect(tbl.countRows()).resolves.toBe(2); diff --git a/nodejs/lancedb/connection.ts b/nodejs/lancedb/connection.ts index e63a7ae65..a81dc0442 100644 --- a/nodejs/lancedb/connection.ts +++ b/nodejs/lancedb/connection.ts @@ -327,6 +327,14 @@ export abstract class Connection { */ abstract dropTable(name: string, namespacePath?: string[]): Promise; + /** + * Start dropping a table and return its cleanup job. + * + * The table may become unavailable before its data files are removed. Wait + * on the returned job to know when cleanup has finished. + */ + abstract dropTableAsync(name: string, namespacePath?: string[]): Promise; + /** * Drop all tables in the database. * @param {string[]} namespacePath The namespace path to drop tables from (defaults to root namespace). @@ -705,6 +713,10 @@ export class LocalConnection extends Connection { return this.inner.dropTable(name, namespacePath ?? []); } + async dropTableAsync(name: string, namespacePath?: string[]): Promise { + return this.inner.dropTableAsync(name, namespacePath ?? []); + } + async dropAllTables(namespacePath?: string[]): Promise { return this.inner.dropAllTables(namespacePath ?? []); } diff --git a/nodejs/src/connection.rs b/nodejs/src/connection.rs index c45321aba..c9f5e10ea 100644 --- a/nodejs/src/connection.rs +++ b/nodejs/src/connection.rs @@ -334,6 +334,22 @@ impl Connection { .default_error() } + /// Start dropping a table and return its cleanup job. + #[napi(catch_unwind)] + pub async fn drop_table_async( + &self, + name: String, + namespace_path: Option>, + ) -> napi::Result { + let ns = namespace_path.unwrap_or_default(); + let job = self + .get_inner()? + .drop_table_async(&name, &ns) + .await + .default_error()?; + Ok(crate::job::Job::new(job)) + } + #[napi(catch_unwind)] pub async fn drop_all_tables(&self, namespace_path: Option>) -> napi::Result<()> { let ns = namespace_path.unwrap_or_default(); diff --git a/python/python/lancedb/_lancedb.pyi b/python/python/lancedb/_lancedb.pyi index f87fd3d13..447bcc88a 100644 --- a/python/python/lancedb/_lancedb.pyi +++ b/python/python/lancedb/_lancedb.pyi @@ -198,6 +198,9 @@ class Connection(object): async def drop_table( self, name: str, namespace_path: Optional[List[str]] = None ) -> None: ... + async def drop_table_async( + self, name: str, namespace_path: Optional[List[str]] = None + ) -> Job: ... async def drop_all_tables( self, namespace_path: Optional[List[str]] = None ) -> None: ... diff --git a/python/python/lancedb/db.py b/python/python/lancedb/db.py index eeae8bf50..14b6c0b0d 100644 --- a/python/python/lancedb/db.py +++ b/python/python/lancedb/db.py @@ -524,6 +524,12 @@ class DBConnection(EnforceOverrides): namespace_path = [] raise NotImplementedError + def drop_table_async( + self, name: str, namespace_path: Optional[List[str]] = None + ) -> Job: + """Start dropping a table and return its cleanup job.""" + raise NotImplementedError + def rename_table( self, cur_name: str, @@ -1186,6 +1192,20 @@ class LanceDBConnection(DBConnection): ) ) + @override + def drop_table_async( + self, name: str, namespace_path: Optional[List[str]] = None + ) -> Job: + """Start dropping a table and return its cleanup job. + + The table may become unavailable before its data files are removed. + Call :meth:`Job.wait` to wait for cleanup to finish. + """ + if namespace_path is None: + namespace_path = [] + job = LOOP.run(self._conn.drop_table_async(name, namespace_path=namespace_path)) + return Job(job if isinstance(job, AsyncJob) else AsyncJob(job)) + @override def drop_all_tables(self, namespace_path: Optional[List[str]] = None): if namespace_path is None: @@ -1963,6 +1983,23 @@ class AsyncConnection(object): if f"Table '{name}' was not found" not in str(e): raise e + async def drop_table_async( + self, + name: str, + *, + namespace_path: Optional[List[str]] = None, + ) -> AsyncJob: + """Start dropping a table and return its cleanup job. + + The table may become unavailable before its data files are removed. + Await :meth:`AsyncJob.wait` to wait for cleanup to finish. + """ + if namespace_path is None: + namespace_path = [] + return AsyncJob( + await self._inner.drop_table_async(name, namespace_path=namespace_path) + ) + async def drop_all_tables(self, namespace_path: Optional[List[str]] = None): """Drop all tables from the database. diff --git a/python/python/lancedb/namespace.py b/python/python/lancedb/namespace.py index b151395cc..0e60bd218 100644 --- a/python/python/lancedb/namespace.py +++ b/python/python/lancedb/namespace.py @@ -49,6 +49,7 @@ from lancedb._lancedb import ( ) from lancedb.background_loop import LOOP from lancedb.db import AsyncConnection, DBConnection +from lancedb.job import AsyncJob, Job from lance_namespace import ( LanceNamespace, connect as namespace_connect, @@ -624,6 +625,18 @@ class LanceNamespaceDBConnection(DBConnection): namespace_path = [] LOOP.run(self._inner.drop_table(name, namespace_path=namespace_path)) + @override + def drop_table_async( + self, name: str, namespace_path: Optional[List[str]] = None + ) -> Job: + """Start dropping a table and return its cleanup job.""" + if namespace_path is None: + namespace_path = [] + job = LOOP.run( + self._inner.drop_table_async(name, namespace_path=namespace_path) + ) + return Job(job if isinstance(job, AsyncJob) else AsyncJob(job)) + @override def rename_table( self, @@ -1134,6 +1147,14 @@ class AsyncLanceNamespaceDBConnection: namespace_path = [] await self._inner.drop_table(name, namespace_path=namespace_path) + async def drop_table_async( + self, name: str, namespace_path: Optional[List[str]] = None + ) -> AsyncJob: + """Start dropping a table and return its cleanup job.""" + if namespace_path is None: + namespace_path = [] + return await self._inner.drop_table_async(name, namespace_path=namespace_path) + async def rename_table( self, cur_name: str, diff --git a/python/python/lancedb/remote/db.py b/python/python/lancedb/remote/db.py index 332886590..16ad65dcb 100644 --- a/python/python/lancedb/remote/db.py +++ b/python/python/lancedb/remote/db.py @@ -23,7 +23,7 @@ import pyarrow as pa from ..common import DATA from ..db import DBConnection, LOOP -from ..job import Job +from ..job import AsyncJob, Job if TYPE_CHECKING: from .._lancedb import JobDescription, JobInfo @@ -663,6 +663,16 @@ class RemoteDBConnection(DBConnection): namespace_path = [] LOOP.run(self._conn.drop_table(name, namespace_path=namespace_path)) + @override + def drop_table_async( + self, name: str, namespace_path: Optional[List[str]] = None + ) -> Job: + """Start dropping a table and return its cleanup job.""" + if namespace_path is None: + namespace_path = [] + job = LOOP.run(self._conn.drop_table_async(name, namespace_path=namespace_path)) + return Job(job if isinstance(job, AsyncJob) else AsyncJob(job)) + @override def rename_table( self, diff --git a/python/python/tests/test_db.py b/python/python/tests/test_db.py index 84e78fd8f..38bbb53fb 100644 --- a/python/python/tests/test_db.py +++ b/python/python/tests/test_db.py @@ -755,8 +755,7 @@ def test_delete_table(tmp_db: lancedb.DBConnection): assert tmp_db.table_names() == [] -@pytest.mark.asyncio -async def test_delete_table_async(tmp_db: lancedb.DBConnection): +def test_drop_table_async(tmp_db: lancedb.DBConnection): data = pd.DataFrame( { "vector": [[3.1, 4.1], [5.9, 26.5]], @@ -772,7 +771,10 @@ async def test_delete_table_async(tmp_db: lancedb.DBConnection): assert tmp_db.table_names() == ["test"] - tmp_db.drop_table("test") + job = tmp_db.drop_table_async("test") + assert job.id is None + assert job.status() == "finished" + job.wait() assert tmp_db.table_names() == [] tmp_db.create_table("test", data=data) @@ -781,6 +783,17 @@ async def test_delete_table_async(tmp_db: lancedb.DBConnection): tmp_db.drop_table("does_not_exist", ignore_missing=True) +@pytest.mark.asyncio +async def test_drop_table_async_connection(tmp_db_async: lancedb.AsyncConnection): + await tmp_db_async.create_table("test", data=pa.table({"id": [1, 2]})) + + job = await tmp_db_async.drop_table_async("test") + assert job.id is None + assert await job.status() == "finished" + await job.wait() + assert await tmp_db_async.table_names() == [] + + def test_drop_database(tmp_db: lancedb.DBConnection): data = pd.DataFrame( { diff --git a/python/src/connection.rs b/python/src/connection.rs index b97d48ad8..dbda29ba6 100644 --- a/python/src/connection.rs +++ b/python/src/connection.rs @@ -346,6 +346,23 @@ impl Connection { }) } + #[pyo3(signature = (name, namespace_path=None))] + pub fn drop_table_async( + self_: PyRef<'_, Self>, + name: String, + namespace_path: Option>, + ) -> PyResult> { + let inner = self_.get_inner()?.clone(); + let ns_path = namespace_path.unwrap_or_default(); + future_into_py(self_.py(), async move { + inner + .drop_table_async(name, &ns_path) + .await + .infer_error() + .map(crate::job::Job::new) + }) + } + #[pyo3(signature = (namespace_path=None,))] pub fn drop_all_tables( self_: PyRef<'_, Self>, diff --git a/rust/lancedb/src/connection.rs b/rust/lancedb/src/connection.rs index 1f2708d4e..12ca306b8 100644 --- a/rust/lancedb/src/connection.rs +++ b/rust/lancedb/src/connection.rs @@ -565,6 +565,21 @@ impl Connection { .await } + /// Start dropping a table and return a handle to the cleanup job. + /// + /// The table may become unavailable before its physical data is removed. + /// Call [`crate::job::Job::wait`] to wait for cleanup to finish. Local + /// backends may complete the drop before returning the handle. + pub async fn drop_table_async( + &self, + name: impl AsRef, + namespace_path: &[String], + ) -> Result { + self.internal + .drop_table_async(name.as_ref(), namespace_path) + .await + } + /// Drop the database /// /// This is the same as dropping all of the tables diff --git a/rust/lancedb/src/database.rs b/rust/lancedb/src/database.rs index f99f6e12a..f52c02439 100644 --- a/rust/lancedb/src/database.rs +++ b/rust/lancedb/src/database.rs @@ -323,6 +323,18 @@ pub trait Database: ) -> Result<()>; /// Drop a table in the database async fn drop_table(&self, name: &str, namespace_path: &[String]) -> Result<()>; + /// Start dropping a table and return a handle to the cleanup job. + /// + /// Backends without asynchronous cleanup complete the drop before + /// returning an already-finished job. + async fn drop_table_async( + &self, + name: &str, + namespace_path: &[String], + ) -> Result { + self.drop_table(name, namespace_path).await?; + Ok(crate::job::Job::new_done()) + } /// Drop all tables in the database async fn drop_all_tables(&self, namespace_path: &[String]) -> Result<()>; fn as_any(&self) -> &dyn std::any::Any; diff --git a/rust/lancedb/src/remote.rs b/rust/lancedb/src/remote.rs index 4b5f8832f..be9d0eef6 100644 --- a/rust/lancedb/src/remote.rs +++ b/rust/lancedb/src/remote.rs @@ -19,6 +19,15 @@ const ARROW_FILE_CONTENT_TYPE: &str = "application/vnd.apache.arrow.file"; #[cfg(test)] const JSON_CONTENT_TYPE: &str = "application/json"; +fn extract_job_id(body: &str) -> Option { + serde_json::from_str::(body) + .ok()? + .get("job_id")? + .as_str() + .filter(|job_id| !job_id.is_empty()) + .map(str::to_string) +} + pub use client::{ClientConfig, HeaderProvider, RetryConfig, TimeoutConfig, TlsConfig}; pub use db::{RemoteDatabaseOptions, RemoteDatabaseOptionsBuilder}; pub use oauth::{OAuthConfig, OAuthFlow, OAuthHeaderProvider}; diff --git a/rust/lancedb/src/remote/db.rs b/rust/lancedb/src/remote/db.rs index 839cb3797..45a0bd925 100644 --- a/rust/lancedb/src/remote/db.rs +++ b/rust/lancedb/src/remote/db.rs @@ -9,6 +9,7 @@ use http::StatusCode; use lance_io::object_store::StorageOptions; use lance_namespace_impls::{DynamicContextProvider, OperationInfo}; use moka::future::Cache; +use reqwest::Response; use reqwest::header::CONTENT_TYPE; use lance_namespace::models::{ @@ -23,15 +24,17 @@ use crate::database::{ JobDescription, JobInfo, OpenTableRequest, ReadConsistency, TableNamesRequest, }; use crate::error::Result; +use crate::job::Job; +use crate::remote::job::RemoteJob; use crate::remote::util::stream_as_body; use crate::table::BaseTable; -use super::ARROW_STREAM_CONTENT_TYPE; use super::client::{ ClientConfig, HeaderProvider, HttpSend, RequestResultExt, RestfulLanceDbClient, Sender, }; use super::table::RemoteTable; use super::util::parse_server_version; +use super::{ARROW_STREAM_CONTENT_TYPE, extract_job_id}; // Request structure for the remote clone table API #[derive(serde::Serialize)] @@ -326,6 +329,22 @@ impl RemoteDatabase { } } +impl RemoteDatabase { + async fn submit_drop_table( + &self, + name: &str, + namespace_path: &[String], + ) -> Result<(String, Response)> { + let identifier = build_table_identifier(name, namespace_path, &self.client.id_delimiter); + let cache_key = build_cache_key(name, namespace_path); + let req = self.client.post(&format!("/v1/table/{}/drop/", identifier)); + let (request_id, resp) = self.client.send(req).await?; + let resp = self.client.check_response(&request_id, resp).await?; + self.table_cache.remove(&cache_key).await; + Ok((request_id, resp)) + } +} + #[cfg(all(test, feature = "remote"))] mod test_utils { use super::*; @@ -894,13 +913,28 @@ impl Database for RemoteDatabase { } async fn drop_table(&self, name: &str, namespace_path: &[String]) -> Result<()> { - let identifier = build_table_identifier(name, namespace_path, &self.client.id_delimiter); - let cache_key = build_cache_key(name, namespace_path); - let req = self.client.post(&format!("/v1/table/{}/drop/", identifier)); - let (request_id, resp) = self.client.send(req).await?; - self.client.check_response(&request_id, resp).await?; - self.table_cache.remove(&cache_key).await; - Ok(()) + self.submit_drop_table(name, namespace_path) + .await + .map(|_| ()) + } + + async fn drop_table_async(&self, name: &str, namespace_path: &[String]) -> Result { + let (request_id, response) = self.submit_drop_table(name, namespace_path).await?; + let status = response.status(); + let body = response.text().await.err_to_http(request_id.clone())?; + let job_id = extract_job_id(&body); + Ok(match job_id { + Some(job_id) => Job::new(Box::new(RemoteJob::new(self.client.clone(), job_id))), + None if status == StatusCode::ACCEPTED => { + return Err(Error::Http { + source: "asynchronous drop-table response did not contain a valid job_id" + .into(), + request_id, + status_code: Some(status), + }); + } + None => Job::new_done(), + }) } async fn drop_all_tables(&self, namespace_path: &[String]) -> Result<()> { @@ -1492,6 +1526,67 @@ mod tests { // NOTE: the API will return 200 even if the table does not exist. So we shouldn't expect 404. } + #[tokio::test] + async fn test_drop_table_does_not_read_response_body() { + let conn = Connection::new_with_handler(|_| { + http::Response::builder() + .status(200) + .body(vec![0xff]) + .unwrap() + }); + + conn.drop_table("table1", &[]).await.unwrap(); + } + + #[tokio::test] + async fn test_drop_table_async_returns_job() { + let conn = Connection::new_with_handler(|request| { + assert_eq!(request.method(), &reqwest::Method::POST); + assert_eq!(request.url().path(), "/v1/table/table1/drop/"); + http::Response::builder() + .status(202) + .body(r#"{"job_id":"drop-job-123"}"#) + .unwrap() + }); + + let job = conn.drop_table_async("table1", &[]).await.unwrap(); + assert_eq!(job.id(), Some("drop-job-123")); + } + + #[tokio::test] + async fn test_drop_table_async_old_server_returns_done_job() { + let conn = Connection::new_with_handler(|_| { + http::Response::builder().status(200).body("").unwrap() + }); + + let job = conn.drop_table_async("table1", &[]).await.unwrap(); + assert_eq!(job.id(), None); + assert_eq!(job.status().await.unwrap(), "finished"); + } + + #[tokio::test] + async fn test_drop_table_async_rejects_accepted_response_without_job_id() { + let conn = Connection::new_with_handler(|_| { + http::Response::builder().status(202).body("{}").unwrap() + }); + + let error = conn.drop_table_async("table1", &[]).await.err().unwrap(); + assert!(error.to_string().contains("valid job_id")); + } + + #[tokio::test] + async fn test_drop_table_async_rejects_empty_job_id() { + let conn = Connection::new_with_handler(|_| { + http::Response::builder() + .status(202) + .body(r#"{"job_id":""}"#) + .unwrap() + }); + + let error = conn.drop_table_async("table1", &[]).await.err().unwrap(); + assert!(error.to_string().contains("valid job_id")); + } + #[tokio::test] async fn test_rename_table() { let conn = Connection::new_with_handler(|request| { diff --git a/rust/lancedb/src/remote/table.rs b/rust/lancedb/src/remote/table.rs index 0d843dd54..3816a3a86 100644 --- a/rust/lancedb/src/remote/table.rs +++ b/rust/lancedb/src/remote/table.rs @@ -8,7 +8,7 @@ use self::insert::{RemoteWriteExec, WriteOp}; use super::client::RequestResultExt; use super::client::{HttpSend, RestfulLanceDbClient, Sender}; use super::db::ServerVersion; -use super::{ARROW_FILE_CONTENT_TYPE, ARROW_STREAM_CONTENT_TYPE}; +use super::{ARROW_FILE_CONTENT_TYPE, ARROW_STREAM_CONTENT_TYPE, extract_job_id}; use crate::blob::BlobFile; use crate::data::scannable::{PeekedScannable, Scannable, estimate_write_partitions}; use crate::expr::expr_to_sql_string; @@ -392,13 +392,7 @@ impl RemoteTable { .text() .await .ok() - .and_then(|body| serde_json::from_str::(&body).ok()) - .and_then(|value| { - value - .get("job_id") - .and_then(|id| id.as_str()) - .map(str::to_string) - }); + .and_then(|body| extract_job_id(&body)); if let Some(wait_timeout) = index.wait_timeout { let index_name = index.name.unwrap_or_else(|| format!("{}_idx", column));