feat: pause and resume jobs from the Python SDK

Cancellation is the only job control the SDK exposes, and it is terminal,
so a long-running server-side job cannot be parked and picked up again.
This adds pause_job and resume_job to the Rust core connection and the
Python bindings (sync and async), posting to the server's /v1/jobs/pause
and /v1/jobs/resume endpoints.

A pause parks the job until it is resumed: its workers drain and stop.
The outcome strings mirror the server's answers -- a job finalizing its
results reports "committing" and cannot be parked, and a resume before
the drain is confirmed reports "still_pausing"; both are retried rather
than failed. Resuming re-queues the job and its workers pick their work
back up from checkpoints. Local connections report the operations as
unsupported, like the rest of the jobs API.

python/python/lancedb/_lancedb.pyi

@@ -154,6 +154,8 @@ class Connection(object):
     async def list_jobs(self) -> List[JobInfo]: ...
     async def get_job(self, job_id: str) -> Optional[JobDescription]: ...
     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 job_history(
         self, job_id: Optional[str] = None
     ) -> List[pa.RecordBatch]: ...

python/python/lancedb/db.py

@@ -753,6 +753,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 job_history(self, job_id: Optional[str] = None) -> List[pa.RecordBatch]:
         """The lifecycle event history of a server-side job, as Arrow batches.
 
@@ -1450,6 +1470,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 job_history(self, job_id: Optional[str] = None) -> List[pa.RecordBatch]:
         """The lifecycle event history of a server-side job, as Arrow batches.
@@ -2281,6 +2317,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 job_history(self, job_id: Optional[str] = None) -> List[pa.RecordBatch]:
         """The lifecycle event history of a server-side job, as Arrow batches.
 

python/python/lancedb/remote/db.py

@@ -776,6 +776,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 job_history(self, job_id: Optional[str] = None) -> List[pa.RecordBatch]:
         """The lifecycle event history of a server-side job, as Arrow batches.

python/python/tests/test_remote_db.py

@@ -2534,6 +2534,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":
             assert payload["job_id"] == "job-1"
             request.send_response(200)
@@ -2562,6 +2582,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"
+
         batches = db.job_history("job-1")
         assert len(batches) == 1
         assert batches[0].num_rows == 2

python/src/connection.rs

@@ -666,6 +666,30 @@ impl Connection {
         })
     }
 
+    pub fn pause_job(self_: PyRef<'_, Self>, job_id: String) -> PyResult<Bound<'_, PyAny>> {
+        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<Bound<'_, PyAny>> {
+        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",
+            })
+        })
+    }
+
     #[pyo3(signature = (job_id=None))]
     pub fn job_history(
         self_: PyRef<'_, Self>,