diff --git a/python/python/lancedb/_lancedb.pyi b/python/python/lancedb/_lancedb.pyi index ba4e3a0df..df5dde88b 100644 --- a/python/python/lancedb/_lancedb.pyi +++ b/python/python/lancedb/_lancedb.pyi @@ -185,6 +185,8 @@ class Connection(object): ) -> List[str]: ... async def list_jobs(self) -> List[JobInfo]: ... async def cancel_job(self, job_id: str) -> bool: ... + async def pause_job(self, job_id: str) -> str: ... + async def resume_job(self, job_id: str) -> str: ... async def execute_query_async( self, query: str, diff --git a/python/python/lancedb/db.py b/python/python/lancedb/db.py index 10d52c63e..c4ca0f528 100644 --- a/python/python/lancedb/db.py +++ b/python/python/lancedb/db.py @@ -1031,6 +1031,26 @@ class DBConnection(EnforceOverrides): "cancel_job is not supported for this connection type" ) + def pause_job(self, job_id: str) -> str: + """Pause a server-side job by id. + + The job's workers drain and it stays parked until resumed. Returns + "pausing", "already_paused", or "committing" -- a job finalizing its + results cannot be parked; retry shortly. + """ + raise NotImplementedError("pause_job is not supported for this connection type") + + def resume_job(self, job_id: str) -> str: + """Resume a paused server-side job by id. + + Its workers pick their work back up from checkpoints. Returns + "resumed", "still_pausing" -- the pause's worker drain is not + confirmed yet; retry shortly -- or "not_paused". + """ + raise NotImplementedError( + "resume_job is not supported for this connection type" + ) + def execute_query( self, query: str, @@ -1855,6 +1875,22 @@ class LanceDBConnection(DBConnection): """ return LOOP.run(self._conn.cancel_job(job_id)) + @override + def pause_job(self, job_id: str) -> str: + """Pause a server-side job by id. + + Returns "pausing", "already_paused", or "committing". + """ + return LOOP.run(self._conn.pause_job(job_id)) + + @override + def resume_job(self, job_id: str) -> str: + """Resume a paused server-side job by id. + + Returns "resumed", "still_pausing", or "not_paused". + """ + return LOOP.run(self._conn.resume_job(job_id)) + @override def namespace_client(self) -> LanceNamespace: """Get the equivalent namespace client for this connection. @@ -2843,6 +2879,23 @@ class AsyncConnection(object): """ return await self._inner.cancel_job(job_id) + async def pause_job(self, job_id: str) -> str: + """Pause a server-side job by id. + + The job's workers drain and it stays parked until resumed. Returns + "pausing", "already_paused", or "committing" -- a job finalizing its + results cannot be parked; retry shortly. + """ + return await self._inner.pause_job(job_id) + + async def resume_job(self, job_id: str) -> str: + """Resume a paused server-side job by id. + + Its workers pick their work back up from checkpoints. Returns + "resumed", "still_pausing" -- retry shortly -- or "not_paused". + """ + return await self._inner.resume_job(job_id) + async def execute_query( self, query: str, diff --git a/python/python/lancedb/remote/db.py b/python/python/lancedb/remote/db.py index 962ef66e1..ee25e1e33 100644 --- a/python/python/lancedb/remote/db.py +++ b/python/python/lancedb/remote/db.py @@ -950,6 +950,22 @@ class RemoteDBConnection(DBConnection): """ return LOOP.run(self._conn.cancel_job(job_id)) + @override + def pause_job(self, job_id: str) -> str: + """Pause a server-side job by id. + + Returns "pausing", "already_paused", or "committing". + """ + return LOOP.run(self._conn.pause_job(job_id)) + + @override + def resume_job(self, job_id: str) -> str: + """Resume a paused server-side job by id. + + Returns "resumed", "still_pausing", or "not_paused". + """ + return LOOP.run(self._conn.resume_job(job_id)) + @override def execute_query_async( self, diff --git a/python/python/tests/test_remote_db.py b/python/python/tests/test_remote_db.py index c03862a49..f829b9985 100644 --- a/python/python/tests/test_remote_db.py +++ b/python/python/tests/test_remote_db.py @@ -2624,6 +2624,26 @@ def test_remote_connection_jobs_surface(): request.send_header("Content-Type", "application/json") request.end_headers() request.wfile.write(b'{"job_id": "job-1"}') + elif request.path == "/v1/jobs/pause": + if payload["job_id"] != "job-1": + request.send_response(404) + request.end_headers() + return + request.send_response(200) + request.send_header("Content-Type", "application/json") + request.end_headers() + request.wfile.write(b'{"job_id": "job-1", "paused": true}') + elif request.path == "/v1/jobs/resume": + if payload["job_id"] != "job-1": + request.send_response(404) + request.end_headers() + return + request.send_response(200) + request.send_header("Content-Type", "application/json") + request.end_headers() + request.wfile.write( + b'{"job_id": "job-1", "resumed": false, "still_pausing": true}' + ) elif request.path == "/v1/jobs/query_events": query_events_payloads.append(payload) request.send_response(200) @@ -2644,6 +2664,9 @@ def test_remote_connection_jobs_surface(): assert db.cancel_job("job-1") is True assert db.cancel_job("missing") is False + assert db.pause_job("job-1") == "pausing" + assert db.resume_job("job-1") == "still_pausing" + # Opening a job hands back a populated handle; a missing one fails. with pytest.raises(JobNotFoundError, match="missing"): db.open_job("missing") diff --git a/python/src/connection.rs b/python/src/connection.rs index 69f54c683..df827cecc 100644 --- a/python/src/connection.rs +++ b/python/src/connection.rs @@ -965,6 +965,30 @@ impl Connection { inner.cancel_job(&job_id).await.infer_error() }) } + + pub fn pause_job(self_: PyRef<'_, Self>, job_id: String) -> PyResult> { + let inner = self_.get_inner()?.clone(); + future_into_py(self_.py(), async move { + let status = inner.pause_job(&job_id).await.infer_error()?; + Ok(match status { + lancedb::database::PauseJobStatus::Pausing => "pausing", + lancedb::database::PauseJobStatus::AlreadyPaused => "already_paused", + lancedb::database::PauseJobStatus::Committing => "committing", + }) + }) + } + + pub fn resume_job(self_: PyRef<'_, Self>, job_id: String) -> PyResult> { + let inner = self_.get_inner()?.clone(); + future_into_py(self_.py(), async move { + let status = inner.resume_job(&job_id).await.infer_error()?; + Ok(match status { + lancedb::database::ResumeJobStatus::Resumed => "resumed", + lancedb::database::ResumeJobStatus::StillPausing => "still_pausing", + lancedb::database::ResumeJobStatus::NotPaused => "not_paused", + }) + }) + } } #[pyfunction] diff --git a/rust/lancedb/src/connection.rs b/rust/lancedb/src/connection.rs index 962e9e586..4c85065c9 100644 --- a/rust/lancedb/src/connection.rs +++ b/rust/lancedb/src/connection.rs @@ -23,8 +23,8 @@ use crate::connection::create_table::CreateTableBuilder; use crate::data::scannable::Scannable; use crate::database::listing::ListingDatabase; use crate::database::{ - CloneTableRequest, Database, DatabaseOptions, JobInfo, OpenTableRequest, ReadConsistency, - TableNamesRequest, + CloneTableRequest, Database, DatabaseOptions, JobInfo, OpenTableRequest, PauseJobStatus, + ReadConsistency, ResumeJobStatus, TableNamesRequest, }; use crate::embeddings::{EmbeddingRegistry, MemoryRegistry}; use crate::error::{Error, Result}; @@ -905,6 +905,18 @@ impl Connection { self.internal.cancel_job(job_id.as_ref()).await } + /// Pause a server-side job by id. Its workers drain and it stays parked + /// until resumed; see [`PauseJobStatus`] for the outcomes. + pub async fn pause_job(&self, job_id: impl AsRef) -> Result { + self.internal.pause_job(job_id.as_ref()).await + } + + /// Resume a paused server-side job by id. Its workers pick their work + /// back up from checkpoints; see [`ResumeJobStatus`] for the outcomes. + pub async fn resume_job(&self, job_id: impl AsRef) -> Result { + self.internal.resume_job(job_id.as_ref()).await + } + /// Drop a table in the database. /// /// # Arguments diff --git a/rust/lancedb/src/database.rs b/rust/lancedb/src/database.rs index 086d1df35..7ffc42c4c 100644 --- a/rust/lancedb/src/database.rs +++ b/rust/lancedb/src/database.rs @@ -241,6 +241,29 @@ pub struct JobDescription { pub failure: Option, } +/// The server's answer to a pause request. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum PauseJobStatus { + /// The pause was accepted; workers drain and the job stays parked. + Pausing, + /// The job was already paused, so a repeated pause changed nothing. + AlreadyPaused, + /// The job is finalizing its results and cannot be parked right now. + /// The commit is the short tail of a long job; retry shortly. + Committing, +} + +/// The server's answer to a resume request. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ResumeJobStatus { + /// The job re-entered the queue and will run again. + Resumed, + /// The pause's worker drain is not confirmed yet; retry shortly. + StillPausing, + /// The job was not paused, so there was nothing to resume. + NotPaused, +} + fn job_op_not_supported(what: &str) -> Result { Err(crate::error::Error::NotSupported { message: format!("{} is not supported by this database", what), @@ -495,6 +518,16 @@ pub trait Database: async fn cancel_job(&self, _job_id: &str) -> Result { job_op_not_supported("cancel_job") } + /// Pause a job by id. The job's workers drain and it stays parked until + /// resumed; see [`PauseJobStatus`] for the outcomes. + async fn pause_job(&self, _job_id: &str) -> Result { + job_op_not_supported("pause_job") + } + /// Resume a paused job by id. It re-enters the queue and its workers pick + /// their work back up from checkpoints; see [`ResumeJobStatus`]. + async fn resume_job(&self, _job_id: &str) -> Result { + job_op_not_supported("resume_job") + } /// Start executing a SQL statement on a remote database. async fn execute_query_async( &self, diff --git a/rust/lancedb/src/remote/db.rs b/rust/lancedb/src/remote/db.rs index 893c584e1..4297879a4 100644 --- a/rust/lancedb/src/remote/db.rs +++ b/rust/lancedb/src/remote/db.rs @@ -30,7 +30,7 @@ use crate::function::{ }; use crate::job::Job; use crate::materialized_view::CreateMaterializedViewRequest; -use crate::remote::job::{RemoteJob, job_state_to_client}; +use crate::remote::job::{PauseJobResponse, RemoteJob, ResumeJobResponse, job_state_to_client}; use crate::remote::util::stream_as_body; use crate::secrets::SecretBinding; use crate::secrets::SecretInfo; @@ -1322,6 +1322,40 @@ impl Database for RemoteDatabase { } } + async fn pause_job(&self, job_id: &str) -> Result { + let req = self + .client + .post("/v1/jobs/pause") + .json(&serde_json::json!({ "job_id": job_id })); + let (request_id, rsp) = self.client.send(req).await?; + let rsp = self.client.check_response(&request_id, rsp).await?; + let body: PauseJobResponse = rsp.json().await.err_to_http(request_id)?; + Ok(if body.paused { + crate::database::PauseJobStatus::Pausing + } else if body.committing { + crate::database::PauseJobStatus::Committing + } else { + crate::database::PauseJobStatus::AlreadyPaused + }) + } + + async fn resume_job(&self, job_id: &str) -> Result { + let req = self + .client + .post("/v1/jobs/resume") + .json(&serde_json::json!({ "job_id": job_id })); + let (request_id, rsp) = self.client.send(req).await?; + let rsp = self.client.check_response(&request_id, rsp).await?; + let body: ResumeJobResponse = rsp.json().await.err_to_http(request_id)?; + Ok(if body.resumed { + crate::database::ResumeJobStatus::Resumed + } else if body.still_pausing { + crate::database::ResumeJobStatus::StillPausing + } else { + crate::database::ResumeJobStatus::NotPaused + }) + } + async fn execute_query_async( &self, query: &str, @@ -3480,6 +3514,45 @@ mod tests { ); } + #[tokio::test] + async fn test_pause_and_resume_job() { + use crate::database::{PauseJobStatus, ResumeJobStatus}; + let conn = Connection::new_with_handler(|request| { + assert_eq!(request.url().path(), "/v1/jobs/pause"); + http::Response::builder() + .status(200) + .body(r#"{"job_id": "job-1", "paused": true}"#) + .unwrap() + }); + assert_eq!( + conn.pause_job("job-1").await.unwrap(), + PauseJobStatus::Pausing + ); + + let conn = Connection::new_with_handler(|_| { + http::Response::builder() + .status(200) + .body(r#"{"job_id": "job-1", "paused": false, "committing": true}"#) + .unwrap() + }); + assert_eq!( + conn.pause_job("job-1").await.unwrap(), + PauseJobStatus::Committing + ); + + let conn = Connection::new_with_handler(|request| { + assert_eq!(request.url().path(), "/v1/jobs/resume"); + http::Response::builder() + .status(200) + .body(r#"{"job_id": "job-1", "resumed": false, "still_pausing": true}"#) + .unwrap() + }); + assert_eq!( + conn.resume_job("job-1").await.unwrap(), + ResumeJobStatus::StillPausing + ); + } + #[tokio::test] async fn test_job_events_scope_to_that_job() { let schema = Arc::new(Schema::new(vec![Field::new( diff --git a/rust/lancedb/src/remote/job.rs b/rust/lancedb/src/remote/job.rs index 80256421a..41d8971ac 100644 --- a/rust/lancedb/src/remote/job.rs +++ b/rust/lancedb/src/remote/job.rs @@ -75,6 +75,28 @@ pub(super) struct ReportedFailure { retryable: Option, } +/// Forward-compatible `/v1/jobs/pause` wire envelope. +#[derive(Deserialize)] +pub(super) struct PauseJobResponse { + /// False when the job was already paused, so a repeated pause changed + /// nothing. + #[serde(default)] + pub(super) paused: bool, + /// The job is finalizing its results and cannot be parked right now. + #[serde(default)] + pub(super) committing: bool, +} + +/// Forward-compatible `/v1/jobs/resume` wire envelope. +#[derive(Deserialize)] +pub(super) struct ResumeJobResponse { + #[serde(default)] + pub(super) resumed: bool, + /// The pause's worker drain is not confirmed yet. + #[serde(default)] + pub(super) still_pausing: bool, +} + /// Forward-compatible `/v1/jobs/describe` wire envelope. #[derive(Deserialize)] pub(super) struct DescribeJobResponse {