diff --git a/rust/lancedb/src/connection.rs b/rust/lancedb/src/connection.rs index fba72cd7c..8aedd69fa 100644 --- a/rust/lancedb/src/connection.rs +++ b/rust/lancedb/src/connection.rs @@ -25,8 +25,8 @@ use crate::database::listing::ListingDatabase; use crate::database::{ CloneTableRequest, CreateFunctionRequest, CreateMaterializedViewRequest, Database, DatabaseOptions, FunctionInfo, JobErrorInfo, JobHistoryInfo, JobInfo, MaterializedViewInfo, - MvRefreshPlan, OpenTableRequest, ReadConsistency, RefreshMaterializedViewRequest, - TableLineageRequest, TableNamesRequest, + MvRefreshPlan, OpenTableRequest, PlatformJobDescription, ReadConsistency, + RefreshMaterializedViewRequest, TableLineageRequest, TableNamesRequest, }; use crate::embeddings::{EmbeddingRegistry, MemoryRegistry}; use crate::error::{Error, Result}; @@ -573,6 +573,33 @@ impl Connection { self.internal.cancel_job(job_id).await } + /// Describe a platform job (`POST /v1/jobs/describe`): registry-backed + /// lifecycle state plus the owner-written status payload. `None` when the + /// registry has no such job. + pub async fn describe_platform_job( + &self, + platform_job_id: &str, + ) -> Result> { + self.internal.describe_platform_job(platform_job_id).await + } + + /// Resolve a submission (manifest) job id to its platform job id. `None` + /// until the job has registered (dispatch is async). + pub async fn resolve_platform_job_id( + &self, + manifest_job_id: &str, + table_hint: Option<&str>, + ) -> Result> { + self.internal + .resolve_platform_job_id(manifest_job_id, table_hint) + .await + } + + /// Cancel a platform job. Idempotent on already-terminal jobs. + pub async fn cancel_platform_job(&self, platform_job_id: &str) -> Result<()> { + self.internal.cancel_platform_job(platform_job_id).await + } + /// Look up a single server-side job by id -- the `wait()`/status poll path. /// `table_hint` (the job's table) enables an O(1) server-side lookup; `None` /// scans the database's active jobs. A `None` result means unknown / not diff --git a/rust/lancedb/src/database.rs b/rust/lancedb/src/database.rs index 1bc12c4fc..533a2590a 100644 --- a/rust/lancedb/src/database.rs +++ b/rust/lancedb/src/database.rs @@ -321,6 +321,23 @@ pub struct MaterializedViewInfo { pub auto_refresh: bool, } +/// A described platform job (`POST /v1/jobs/describe`): the job registry's +/// lifecycle state plus the owner-written status payload. +#[derive(Debug, Clone)] +pub struct PlatformJobDescription { + /// The platform (registry) job id -- what describe/cancel accept. + pub job_id: String, + pub job_type: String, + pub job_subtype: String, + /// "IN_PROGRESS" | "CANCELLED" | "FAILED" | "DONE". + pub job_state: String, + pub creation_ms: i64, + /// The owner-written status payload -- `units_done` / `units_total` / + /// `rows_committed` / `error` when present. Records whose owner has not + /// written a payload yet carry the raw status-store URI string instead. + pub status: serde_json::Value, +} + /// A row from `list_jobs`: one inflight server-side job (index build, /// compaction, column refresh, view refresh, ...). #[derive(Debug, Clone)] @@ -510,6 +527,33 @@ pub trait Database: async fn list_jobs(&self) -> Result> { not_supported("list_jobs") } + + /// Describe a platform job (`POST /v1/jobs/describe`): registry-backed + /// lifecycle state plus the owner-written status payload. `None` when the + /// registry has no such job. + async fn describe_platform_job( + &self, + _platform_job_id: &str, + ) -> Result> { + not_supported("describe_platform_job") + } + + /// Resolve a submission (manifest) job id to its platform job id via the + /// registry's manifest-id filter (`POST /v1/jobs/list`). `None` until the + /// job has registered (dispatch is async). + async fn resolve_platform_job_id( + &self, + _manifest_job_id: &str, + _table_hint: Option<&str>, + ) -> Result> { + not_supported("resolve_platform_job_id") + } + + /// Cancel a platform job (`POST /v1/jobs/cancel`). Idempotent: cancelling + /// an already-terminal job is a no-op success. + async fn cancel_platform_job(&self, _platform_job_id: &str) -> Result<()> { + not_supported("cancel_platform_job") + } /// Cancel an inflight server-side job by id. Returns true if a /// matching inflight job was found and flagged for cancellation, /// false if none was inflight (best-effort, like SQL `CANCEL JOB`). diff --git a/rust/lancedb/src/remote/db.rs b/rust/lancedb/src/remote/db.rs index 727496149..f548fb4ab 100644 --- a/rust/lancedb/src/remote/db.rs +++ b/rust/lancedb/src/remote/db.rs @@ -158,6 +158,30 @@ struct RemoteJobEntry { error: Option, } +#[derive(Deserialize)] +struct RemoteDescribePlatformJobResponse { + job_id: String, + job_type: String, + #[serde(default)] + job_subtype: String, + job_state: String, + #[serde(default)] + creation_ms: i64, + #[serde(default)] + status: serde_json::Value, +} + +#[derive(Deserialize)] +struct RemoteListPlatformJobsResponse { + #[serde(default)] + jobs: Vec, +} + +#[derive(Deserialize)] +struct RemotePlatformJobRow { + job_id: String, +} + #[derive(serde::Deserialize)] struct RemoteListJobsResponse { jobs: Vec, @@ -1074,6 +1098,56 @@ impl Database for RemoteDatabase { Ok(body.job.map(JobInfo::from)) } + async fn describe_platform_job( + &self, + platform_job_id: &str, + ) -> Result> { + let req = self + .client + .post("/v1/jobs/describe") + .json(&serde_json::json!({ "job_id": platform_job_id })); + let (request_id, rsp) = self.client.send(req).await?; + if rsp.status().as_u16() == 404 { + return Ok(None); + } + let rsp = self.client.check_response(&request_id, rsp).await?; + let body: RemoteDescribePlatformJobResponse = rsp.json().await.err_to_http(request_id)?; + Ok(Some(PlatformJobDescription { + job_id: body.job_id, + job_type: body.job_type, + job_subtype: body.job_subtype, + job_state: body.job_state, + creation_ms: body.creation_ms, + status: body.status, + })) + } + + async fn resolve_platform_job_id( + &self, + manifest_job_id: &str, + table_hint: Option<&str>, + ) -> Result> { + let req = self.client.post("/v1/jobs/list").json(&serde_json::json!({ + "manifest_job_id": manifest_job_id, + "table_name": table_hint, + "job_type": "indexer", + })); + let (request_id, rsp) = self.client.send(req).await?; + let rsp = self.client.check_response(&request_id, rsp).await?; + let body: RemoteListPlatformJobsResponse = rsp.json().await.err_to_http(request_id)?; + Ok(body.jobs.into_iter().next().map(|row| row.job_id)) + } + + async fn cancel_platform_job(&self, platform_job_id: &str) -> Result<()> { + let req = self + .client + .post("/v1/jobs/cancel") + .json(&serde_json::json!({ "job_id": platform_job_id })); + let (request_id, rsp) = self.client.send(req).await?; + self.client.check_response(&request_id, rsp).await?; + Ok(()) + } + async fn cancel_job(&self, job_id: &str) -> Result { let req = self.client.post(&format!("/v1/job/{}/cancel", job_id)); let (request_id, rsp) = self.client.send(req).await?;