mirror of
https://github.com/lancedb/lancedb.git
synced 2026-11-13 14:16:54 +00:00
feat: expose queued Function refresh status (#4959)
## Problem The existing Function `column_status` surface cannot decode the Agent scheduler's `queued` state or the `failure_threshold_reached` and `binding_revision_missing` blocked reasons. It also preserves only the latest refresh job, so a later cancellation can hide an earlier actual failure. ## Change - Adds the `queued`, `failure_threshold_reached`, and `binding_revision_missing` wire states in Rust, Python, and Node. The wire change is additive, while Rust callers must update exhaustive enum matches and public struct literals. - Adds optional `last_failure_job` alongside `last_refresh_job`. The latest cancellation and the most recent actual failure retain separate meanings. - Reuses the canonical Rust `JobFailure` and existing Python/Node `JobFailureInfo` models for nested job failure details instead of introducing a parallel status error model. - Maps the existing public `sql_job_id` field through Node's `JobFailureInfo`. - Keeps `column_status`, `column_statuses`, and `wait_fresh` as the existing API surfaces. There is no new endpoint, row-error list, durable ledger, or unknown-enum fallback. - Documents `queued` versus `running` and the separate latest-attempt and latest-failure jobs in the existing hidden freshness release draft. The new status values are additive but cross-service: `queued`, the two new blocked reasons, and `last_failure_job` require matching Agent/QN and SDK deployments before clients can rely on them. This SDK follow-up does not change the normal `7cc3904a` dependency pin or claim whole ENT-3487 completion. ## Validation The tested private source is `b19c8adbdc936c5976def6c17be47f6d5e3faf3e`, based on product commit `c9bac25ecb8b73b1bd0256bbea65c2621bcf6a94`. The final product docs commit changes generated TypeDoc only. - Rust: 2 selected tests passed, 0 failed, exit 0. The same two Rust blobs are unchanged from their tested `788ef85b` source through `c9bac25e`. - Python: the PyO3 extension built successfully, and the selected native PyO3 SDK test against a mock HTTP REST transport passed 1/0, exit 0. The test requests pinned version 42. - Node: the addon and TypeScript builds passed, Biome passed, and the selected Jest transport test passed 1/0 with 57 filtered tests. Addon SHA-256: `c888e260b61d41e79689ed2f20dd0dc98586a2377ce34f21610b06111734d426`. - Python extension SHA-256: `36c961d8a0cbd2e102a7765f2e47efb4c5af6310baa6262c8e3120407463da2b`. - TypeDoc was generated from the native `native.d.ts` produced by that Node build. A second generation produced no remaining diff. The public diff contains only product, test, and generated API documentation changes. Native runner scripts and preserved compile-failure logs remain private validation evidence. Agent/QN production mapping, mixed-version rollout, and the remaining ENT-3487 server-side work are tracked separately.
This commit is contained in:
Xuanwo
@@ -27,7 +27,9 @@ optional blockedReason:
|
||||
| "dependency_in_flight"
|
||||
| "table_capacity"
|
||||
| "cancelled"
|
||||
| "preparation_backoff";
|
||||
| "preparation_backoff"
|
||||
| "failure_threshold_reached"
|
||||
| "binding_revision_missing";
|
||||
```
|
||||
|
||||
***
|
||||
@@ -85,6 +87,7 @@ optional nextRunAtReason: string;
|
||||
```ts
|
||||
state:
|
||||
| "running"
|
||||
| "queued"
|
||||
| "disabled"
|
||||
| "ready"
|
||||
| "up_to_date"
|
||||
|
||||
@@ -26,6 +26,16 @@ optional completedAtMillis: bigint;
|
||||
|
||||
***
|
||||
|
||||
### failure?
|
||||
|
||||
```ts
|
||||
optional failure: JobFailureInfo;
|
||||
```
|
||||
|
||||
Structured failure details for a failed Job.
|
||||
|
||||
***
|
||||
|
||||
### jobId
|
||||
|
||||
```ts
|
||||
|
||||
@@ -58,6 +58,17 @@ function: FunctionColumnVersionStatus;
|
||||
|
||||
***
|
||||
|
||||
### lastFailureJob?
|
||||
|
||||
```ts
|
||||
optional lastFailureJob: FunctionColumnLastRefreshJob;
|
||||
```
|
||||
|
||||
Newest failed refresh Job. This remains present when `lastRefreshJob`
|
||||
is a later cancellation.
|
||||
|
||||
***
|
||||
|
||||
### lastRefreshJob?
|
||||
|
||||
```ts
|
||||
|
||||
@@ -34,6 +34,16 @@ optional retryable: boolean;
|
||||
|
||||
***
|
||||
|
||||
### sqlJobId?
|
||||
|
||||
```ts
|
||||
optional sqlJobId: string;
|
||||
```
|
||||
|
||||
Isolated SQL job that raised this failure, when supplied by the server.
|
||||
|
||||
***
|
||||
|
||||
### traceback?
|
||||
|
||||
```ts
|
||||
|
||||
@@ -197,6 +197,17 @@ println!("{}", final_status.failed);
|
||||
```
|
||||
</CodeGroup>
|
||||
|
||||
`auto_refresh.state` reports `queued` while the scheduler has accepted the
|
||||
refresh but a worker has not started it; `running` means execution has started.
|
||||
`last_refresh_job` describes the most recent attempt, including a cancellation
|
||||
or a failure. If a later attempt was cancelled, `last_failure_job` retains the
|
||||
most recent job that actually failed, so its nested `failure` continues to
|
||||
explain the failed rows.
|
||||
|
||||
These fields require a matching Agent, Query Node, and SDK deployment. An older
|
||||
client cannot decode the `queued` state, and an older server can omit the new
|
||||
optional failure fields.
|
||||
|
||||
The wait helpers do not submit or cancel a refresh. They retry only the typed
|
||||
finalization-pending response and propagate other errors.
|
||||
*/}
|
||||
|
||||
@@ -681,14 +681,33 @@ describe("remote connection", () => {
|
||||
"stale":2,
|
||||
"never_computed":0,
|
||||
"failed":0,
|
||||
"last_refresh_job":{
|
||||
"job_id":"job-cancelled",
|
||||
"state":"cancelled",
|
||||
"automatic":true,
|
||||
"completed_at_millis":1790000002000
|
||||
},
|
||||
"last_failure_job":{
|
||||
"job_id":"job-failed",
|
||||
"state":"failed",
|
||||
"automatic":true,
|
||||
"completed_at_millis":1790000001000,
|
||||
"failure":{
|
||||
"phase":"finalize",
|
||||
"message":"worker timed out",
|
||||
"traceback":"Traceback: timeout",
|
||||
"retryable":true,
|
||||
"sql_job_id":"sql-job-7"
|
||||
}
|
||||
},
|
||||
"auto_refresh":{
|
||||
"enabled":true,
|
||||
"interval_seconds":60,
|
||||
"state":"blocked",
|
||||
"failure_streak":3,
|
||||
"effective_cooldown_millis":8000,
|
||||
"blocked_reason":"preparation_backoff",
|
||||
"next_run_at_millis":null
|
||||
"active_job_id":"job-queued",
|
||||
"state":"queued",
|
||||
"failure_streak":1,
|
||||
"next_run_at_millis":null,
|
||||
"next_run_at_reason":"job_queued_not_started"
|
||||
}
|
||||
}]`,
|
||||
);
|
||||
@@ -701,10 +720,22 @@ describe("remote connection", () => {
|
||||
await table.checkout(42);
|
||||
const status = await table.functionColumnStatus("embedding");
|
||||
expect(status.tableVersion).toBe(42n);
|
||||
expect(status.autoRefresh.state).toBe("blocked");
|
||||
expect(status.autoRefresh.failureStreak).toBe(3);
|
||||
expect(status.autoRefresh.effectiveCooldownMillis).toBe(8000n);
|
||||
expect(status.autoRefresh.blockedReason).toBe("preparation_backoff");
|
||||
expect(status.autoRefresh.state).toBe("queued");
|
||||
expect(status.autoRefresh.activeJobId).toBe("job-queued");
|
||||
expect(status.autoRefresh.failureStreak).toBe(1);
|
||||
expect(status.autoRefresh.nextRunAtReason).toBe(
|
||||
"job_queued_not_started",
|
||||
);
|
||||
expect(status.lastRefreshJob?.state).toBe("cancelled");
|
||||
expect(status.lastRefreshJob?.failure).toBeUndefined();
|
||||
expect(status.lastFailureJob?.jobId).toBe("job-failed");
|
||||
expect(status.lastFailureJob?.failure).toEqual({
|
||||
phase: "finalize",
|
||||
message: "worker timed out",
|
||||
traceback: "Traceback: timeout",
|
||||
retryable: true,
|
||||
sqlJobId: "sql-job-7",
|
||||
});
|
||||
},
|
||||
);
|
||||
expect(queries).toEqual([{ column: "embedding", version: "42" }]);
|
||||
|
||||
+12
-1
@@ -35,6 +35,7 @@ import {
|
||||
FunctionErrorsOptions,
|
||||
IndexConfig,
|
||||
IndexStatistics,
|
||||
JobFailureInfo,
|
||||
Branches as NativeBranches,
|
||||
OptimizeStats,
|
||||
RTEStats,
|
||||
@@ -274,6 +275,8 @@ export interface FunctionColumnLastRefreshJob {
|
||||
state: string;
|
||||
automatic?: boolean;
|
||||
completedAtMillis?: bigint;
|
||||
/** Structured failure details for a failed Job. */
|
||||
failure?: JobFailureInfo;
|
||||
}
|
||||
|
||||
/** Automatic-refresh projection for a Function output. */
|
||||
@@ -284,6 +287,7 @@ export interface FunctionColumnAutoRefreshStatus {
|
||||
state:
|
||||
| "disabled"
|
||||
| "running"
|
||||
| "queued"
|
||||
| "ready"
|
||||
| "up_to_date"
|
||||
| "waiting_interval"
|
||||
@@ -298,7 +302,9 @@ export interface FunctionColumnAutoRefreshStatus {
|
||||
| "dependency_in_flight"
|
||||
| "table_capacity"
|
||||
| "cancelled"
|
||||
| "preparation_backoff";
|
||||
| "preparation_backoff"
|
||||
| "failure_threshold_reached"
|
||||
| "binding_revision_missing";
|
||||
nextRunAtMillis: bigint | null;
|
||||
nextRunAtReason?: string;
|
||||
}
|
||||
@@ -315,6 +321,11 @@ export interface FunctionColumnStatus {
|
||||
neverComputed: bigint;
|
||||
failed: bigint;
|
||||
lastRefreshJob?: FunctionColumnLastRefreshJob;
|
||||
/**
|
||||
* Newest failed refresh Job. This remains present when `lastRefreshJob`
|
||||
* is a later cancellation.
|
||||
*/
|
||||
lastFailureJob?: FunctionColumnLastRefreshJob;
|
||||
autoRefresh: FunctionColumnAutoRefreshStatus;
|
||||
}
|
||||
|
||||
|
||||
+16
-6
@@ -138,12 +138,7 @@ impl Job {
|
||||
/// Why the job failed, when it failed and the server reports a reason.
|
||||
#[napi(getter)]
|
||||
pub fn failure(&self) -> Option<JobFailureInfo> {
|
||||
self.inner.failure().map(|failure| JobFailureInfo {
|
||||
phase: failure.phase,
|
||||
message: failure.message,
|
||||
traceback: failure.traceback,
|
||||
retryable: failure.retryable,
|
||||
})
|
||||
self.inner.failure().map(Into::into)
|
||||
}
|
||||
|
||||
/// This job's recorded lifecycle events, as an Arrow IPC stream buffer.
|
||||
@@ -216,10 +211,25 @@ impl From<lancedb::database::JobInfo> for JobInfo {
|
||||
|
||||
/// The server's account of why a job failed.
|
||||
#[napi(object)]
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct JobFailureInfo {
|
||||
pub phase: Option<String>,
|
||||
pub message: Option<String>,
|
||||
/// User-code traceback, possibly truncated by the server.
|
||||
pub traceback: Option<String>,
|
||||
pub retryable: Option<bool>,
|
||||
/// Isolated SQL job that raised this failure, when supplied by the server.
|
||||
pub sql_job_id: Option<String>,
|
||||
}
|
||||
|
||||
impl From<lancedb::error::JobFailure> for JobFailureInfo {
|
||||
fn from(failure: lancedb::error::JobFailure) -> Self {
|
||||
Self {
|
||||
phase: failure.phase,
|
||||
message: failure.message,
|
||||
traceback: failure.traceback,
|
||||
retryable: failure.retryable,
|
||||
sql_job_id: failure.sql_job_id,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+17
-8
@@ -20,6 +20,7 @@ use napi_derive::napi;
|
||||
use crate::blob::{BlobFile, copy_blob_buffers, parse_row_ids, spawn_abortable};
|
||||
use crate::error::NapiErrorExt;
|
||||
use crate::index::Index;
|
||||
use crate::job::JobFailureInfo;
|
||||
use crate::merge::NativeMergeInsertBuilder;
|
||||
use crate::query::{Query, TakeQuery, VectorQuery};
|
||||
use crate::util::schema_to_buffer;
|
||||
@@ -1868,6 +1869,19 @@ pub struct FunctionColumnLastRefreshJob {
|
||||
pub state: String,
|
||||
pub automatic: Option<bool>,
|
||||
pub completed_at_millis: Option<BigInt>,
|
||||
pub failure: Option<JobFailureInfo>,
|
||||
}
|
||||
|
||||
impl From<lancedb::table::FunctionColumnLastRefreshJob> for FunctionColumnLastRefreshJob {
|
||||
fn from(job: lancedb::table::FunctionColumnLastRefreshJob) -> Self {
|
||||
Self {
|
||||
job_id: job.job_id,
|
||||
state: job.state,
|
||||
automatic: job.automatic,
|
||||
completed_at_millis: job.completed_at_millis.map(BigInt::from),
|
||||
failure: job.failure.map(Into::into),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Automatic-refresh projection for a Function output.
|
||||
@@ -1899,6 +1913,7 @@ pub struct FunctionColumnStatus {
|
||||
pub never_computed: BigInt,
|
||||
pub failed: BigInt,
|
||||
pub last_refresh_job: Option<FunctionColumnLastRefreshJob>,
|
||||
pub last_failure_job: Option<FunctionColumnLastRefreshJob>,
|
||||
pub auto_refresh: FunctionColumnAutoRefreshStatus,
|
||||
}
|
||||
|
||||
@@ -1921,14 +1936,8 @@ impl From<lancedb::table::FunctionColumnStatus> for FunctionColumnStatus {
|
||||
stale: BigInt::from(status.stale),
|
||||
never_computed: BigInt::from(status.never_computed),
|
||||
failed: BigInt::from(status.failed),
|
||||
last_refresh_job: status
|
||||
.last_refresh_job
|
||||
.map(|job| FunctionColumnLastRefreshJob {
|
||||
job_id: job.job_id,
|
||||
state: job.state,
|
||||
automatic: job.automatic,
|
||||
completed_at_millis: job.completed_at_millis.map(BigInt::from),
|
||||
}),
|
||||
last_refresh_job: status.last_refresh_job.map(Into::into),
|
||||
last_failure_job: status.last_failure_job.map(Into::into),
|
||||
auto_refresh: FunctionColumnAutoRefreshStatus {
|
||||
enabled: status.auto_refresh.enabled,
|
||||
interval_seconds: status.auto_refresh.interval_seconds.map(BigInt::from),
|
||||
|
||||
@@ -1108,6 +1108,7 @@ class FunctionColumnLastRefreshJob:
|
||||
state: str
|
||||
automatic: Optional[bool]
|
||||
completed_at_millis: Optional[int]
|
||||
failure: Optional[JobFailureInfo]
|
||||
|
||||
class FunctionColumnAutoRefreshStatus:
|
||||
enabled: bool
|
||||
@@ -1131,6 +1132,7 @@ class FunctionColumnStatus:
|
||||
never_computed: int
|
||||
failed: int
|
||||
last_refresh_job: Optional[FunctionColumnLastRefreshJob]
|
||||
last_failure_job: Optional[FunctionColumnLastRefreshJob]
|
||||
auto_refresh: FunctionColumnAutoRefreshStatus
|
||||
|
||||
class RefreshMaterializedViewResult:
|
||||
|
||||
@@ -2306,15 +2306,33 @@ def test_remote_function_column_status_passes_pinned_version():
|
||||
"stale": 2,
|
||||
"never_computed": 0,
|
||||
"failed": 0,
|
||||
"last_refresh_job": {
|
||||
"job_id": "job-cancelled",
|
||||
"state": "cancelled",
|
||||
"automatic": True,
|
||||
"completed_at_millis": 1790000002000,
|
||||
},
|
||||
"last_failure_job": {
|
||||
"job_id": "job-failed",
|
||||
"state": "failed",
|
||||
"automatic": True,
|
||||
"completed_at_millis": 1790000001000,
|
||||
"failure": {
|
||||
"phase": "finalize",
|
||||
"message": "worker timed out",
|
||||
"traceback": "Traceback: timeout",
|
||||
"retryable": True,
|
||||
"sql_job_id": "sql-job-7",
|
||||
},
|
||||
},
|
||||
"auto_refresh": {
|
||||
"enabled": True,
|
||||
"interval_seconds": 60,
|
||||
"active_job_id": None,
|
||||
"state": "blocked",
|
||||
"failure_streak": 3,
|
||||
"effective_cooldown_millis": 8000,
|
||||
"blocked_reason": "preparation_backoff",
|
||||
"active_job_id": "job-queued",
|
||||
"state": "queued",
|
||||
"failure_streak": 1,
|
||||
"next_run_at_millis": None,
|
||||
"next_run_at_reason": "job_queued_not_started",
|
||||
},
|
||||
}
|
||||
|
||||
@@ -2350,10 +2368,19 @@ def test_remote_function_column_status_passes_pinned_version():
|
||||
0,
|
||||
0,
|
||||
)
|
||||
assert result.auto_refresh.state == "blocked"
|
||||
assert result.auto_refresh.failure_streak == 3
|
||||
assert result.auto_refresh.effective_cooldown_millis == 8000
|
||||
assert result.auto_refresh.blocked_reason == "preparation_backoff"
|
||||
assert result.auto_refresh.state == "queued"
|
||||
assert result.auto_refresh.active_job_id == "job-queued"
|
||||
assert result.auto_refresh.failure_streak == 1
|
||||
assert result.auto_refresh.next_run_at_reason == "job_queued_not_started"
|
||||
assert result.last_refresh_job.state == "cancelled"
|
||||
assert result.last_refresh_job.failure is None
|
||||
assert result.last_failure_job.job_id == "job-failed"
|
||||
failure = result.last_failure_job.failure
|
||||
assert failure.phase == "finalize"
|
||||
assert failure.message == "worker timed out"
|
||||
assert failure.traceback == "Traceback: timeout"
|
||||
assert failure.retryable is True
|
||||
assert failure.sql_job_id == "sql-job-7"
|
||||
|
||||
|
||||
def test_remote_wait_fresh_retries_only_typed_finalization_pending():
|
||||
|
||||
+14
-8
@@ -228,7 +228,7 @@ impl From<lancedb::database::JobInfo> for JobInfo {
|
||||
|
||||
/// The server's account of why a job failed.
|
||||
#[pyclass(module = "lancedb._lancedb", get_all, skip_from_py_object)]
|
||||
#[derive(Clone)]
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct JobFailureInfo {
|
||||
phase: Option<String>,
|
||||
message: Option<String>,
|
||||
@@ -238,6 +238,18 @@ pub struct JobFailureInfo {
|
||||
sql_job_id: Option<String>,
|
||||
}
|
||||
|
||||
impl From<lancedb::error::JobFailure> for JobFailureInfo {
|
||||
fn from(failure: lancedb::error::JobFailure) -> Self {
|
||||
Self {
|
||||
phase: failure.phase,
|
||||
message: failure.message,
|
||||
traceback: failure.traceback,
|
||||
retryable: failure.retryable,
|
||||
sql_job_id: failure.sql_job_id,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[pymethods]
|
||||
impl JobFailureInfo {
|
||||
fn __repr__(&self) -> String {
|
||||
@@ -341,13 +353,7 @@ impl From<lancedb::database::JobDescription> for JobDescription {
|
||||
.filter(|result| !result.is_null())
|
||||
.map(|result| result.to_string()),
|
||||
_progress_json: Some(progress_json),
|
||||
failure: description.failure.map(|failure| JobFailureInfo {
|
||||
phase: failure.phase,
|
||||
message: failure.message,
|
||||
traceback: failure.traceback,
|
||||
retryable: failure.retryable,
|
||||
sql_job_id: failure.sql_job_id,
|
||||
}),
|
||||
failure: description.failure.map(Into::into),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -15,6 +15,7 @@ use crate::{
|
||||
error::PythonErrorExt,
|
||||
expr::PyExpr,
|
||||
index::{CodeAnalyzerOptions, IndexConfig, extract_index_params},
|
||||
job::JobFailureInfo,
|
||||
query::{Query, TakeQuery},
|
||||
table::scannable::PyScannable,
|
||||
};
|
||||
@@ -925,6 +926,7 @@ pub struct FunctionColumnLastRefreshJob {
|
||||
pub state: String,
|
||||
pub automatic: Option<bool>,
|
||||
pub completed_at_millis: Option<i64>,
|
||||
pub failure: Option<JobFailureInfo>,
|
||||
}
|
||||
|
||||
impl From<lancedb::table::FunctionColumnLastRefreshJob> for FunctionColumnLastRefreshJob {
|
||||
@@ -934,6 +936,7 @@ impl From<lancedb::table::FunctionColumnLastRefreshJob> for FunctionColumnLastRe
|
||||
state: job.state,
|
||||
automatic: job.automatic,
|
||||
completed_at_millis: job.completed_at_millis,
|
||||
failure: job.failure.map(Into::into),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -985,6 +988,7 @@ pub struct FunctionColumnStatus {
|
||||
pub never_computed: u64,
|
||||
pub failed: u64,
|
||||
pub last_refresh_job: Option<FunctionColumnLastRefreshJob>,
|
||||
pub last_failure_job: Option<FunctionColumnLastRefreshJob>,
|
||||
pub auto_refresh: FunctionColumnAutoRefreshStatus,
|
||||
}
|
||||
|
||||
@@ -1017,6 +1021,7 @@ impl From<lancedb::table::FunctionColumnStatus> for FunctionColumnStatus {
|
||||
never_computed: status.never_computed,
|
||||
failed: status.failed,
|
||||
last_refresh_job: status.last_refresh_job.map(Into::into),
|
||||
last_failure_job: status.last_failure_job.map(Into::into),
|
||||
auto_refresh: status.auto_refresh.into(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -253,24 +253,48 @@ impl snafu::AsErrorSource for InterruptionSource {
|
||||
/// A job run remotely carries whatever the server reported, which older servers
|
||||
/// do not report at all. Every field is absent rather than invented when the
|
||||
/// backend does not supply it.
|
||||
#[derive(Debug, Clone, Default)]
|
||||
#[derive(Debug, Clone, Default, serde::Deserialize, serde::Serialize)]
|
||||
pub struct JobFailure {
|
||||
/// The stage the job was in, when known.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub phase: Option<String>,
|
||||
/// A human-readable reason, when known.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub message: Option<String>,
|
||||
/// User-code traceback, possibly truncated by the server.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub traceback: Option<String>,
|
||||
/// Whether a retry could clear the failure, when known.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub retryable: Option<bool>,
|
||||
/// Isolated SQL job that raised this failure, when the server supplies
|
||||
/// one across the Flight SQL boundary.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub sql_job_id: Option<String>,
|
||||
/// The underlying error from a job run in this process or a structured
|
||||
/// remote Flight SQL failure.
|
||||
#[serde(skip)]
|
||||
pub source: Option<Arc<Error>>,
|
||||
}
|
||||
|
||||
impl PartialEq for JobFailure {
|
||||
fn eq(&self, other: &Self) -> bool {
|
||||
let source_matches = match (&self.source, &other.source) {
|
||||
(None, None) => true,
|
||||
(Some(left), Some(right)) => Arc::ptr_eq(left, right),
|
||||
_ => false,
|
||||
};
|
||||
self.phase == other.phase
|
||||
&& self.message == other.message
|
||||
&& self.traceback == other.traceback
|
||||
&& self.retryable == other.retryable
|
||||
&& self.sql_job_id == other.sql_job_id
|
||||
&& source_matches
|
||||
}
|
||||
}
|
||||
|
||||
impl Eq for JobFailure {}
|
||||
|
||||
impl JobFailure {
|
||||
/// A failure whose only known detail is the error that caused it.
|
||||
pub(crate) fn from_source(source: Arc<Error>) -> Self {
|
||||
|
||||
@@ -11655,12 +11655,13 @@ mod tests {
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_function_column_status_decodes_blocked_scheduler_fields() {
|
||||
async fn test_function_column_status_decodes_scheduler_lifecycle_and_failures() {
|
||||
let table = Table::new_with_handler("my_table", |_request| {
|
||||
http::Response::builder()
|
||||
.status(200)
|
||||
.body(
|
||||
r#"[{
|
||||
r#"[
|
||||
{
|
||||
"column":"embedding",
|
||||
"binding_id":"fb_01",
|
||||
"function":{
|
||||
@@ -11676,6 +11677,101 @@ mod tests {
|
||||
"stale":2,
|
||||
"never_computed":0,
|
||||
"failed":0,
|
||||
"last_refresh_job":{
|
||||
"job_id":"job-cancelled",
|
||||
"state":"cancelled",
|
||||
"automatic":true,
|
||||
"completed_at_millis":1790000002000
|
||||
},
|
||||
"last_failure_job":{
|
||||
"job_id":"job-failed",
|
||||
"state":"failed",
|
||||
"automatic":true,
|
||||
"completed_at_millis":1790000001000,
|
||||
"failure":{
|
||||
"phase":"finalize",
|
||||
"message":"worker timed out",
|
||||
"traceback":"Traceback: timeout",
|
||||
"retryable":true,
|
||||
"sql_job_id":"sql-job-7"
|
||||
}
|
||||
},
|
||||
"auto_refresh":{
|
||||
"enabled":true,
|
||||
"interval_seconds":60,
|
||||
"active_job_id":"job-queued",
|
||||
"state":"queued",
|
||||
"failure_streak":1,
|
||||
"next_run_at_millis":null,
|
||||
"next_run_at_reason":"job_queued_not_started"
|
||||
}
|
||||
},
|
||||
{
|
||||
"column":"rerank",
|
||||
"binding_id":"fb_02",
|
||||
"function":{
|
||||
"name":"rerank",
|
||||
"object_id":"rerank",
|
||||
"location":"objects/rerank",
|
||||
"version":"3",
|
||||
"manifest_digest":"sha256:def"
|
||||
},
|
||||
"table_version":42,
|
||||
"rows":0,
|
||||
"fresh":0,
|
||||
"stale":0,
|
||||
"never_computed":0,
|
||||
"failed":0,
|
||||
"auto_refresh":{
|
||||
"enabled":true,
|
||||
"interval_seconds":60,
|
||||
"state":"blocked",
|
||||
"failure_streak":5,
|
||||
"blocked_reason":"failure_threshold_reached",
|
||||
"next_run_at_millis":null
|
||||
}
|
||||
},
|
||||
{
|
||||
"column":"classify",
|
||||
"binding_id":"fb_03",
|
||||
"function":{
|
||||
"name":"classify",
|
||||
"object_id":"classify",
|
||||
"location":"objects/classify",
|
||||
"version":"2",
|
||||
"manifest_digest":"sha256:ghi"
|
||||
},
|
||||
"table_version":42,
|
||||
"rows":0,
|
||||
"fresh":0,
|
||||
"stale":0,
|
||||
"never_computed":0,
|
||||
"failed":0,
|
||||
"auto_refresh":{
|
||||
"enabled":true,
|
||||
"interval_seconds":60,
|
||||
"state":"blocked",
|
||||
"failure_streak":0,
|
||||
"blocked_reason":"binding_revision_missing",
|
||||
"next_run_at_millis":null
|
||||
}
|
||||
},
|
||||
{
|
||||
"column":"summarize",
|
||||
"binding_id":"fb_04",
|
||||
"function":{
|
||||
"name":"summarize",
|
||||
"object_id":"summarize",
|
||||
"location":"objects/summarize",
|
||||
"version":"4",
|
||||
"manifest_digest":"sha256:jkl"
|
||||
},
|
||||
"table_version":42,
|
||||
"rows":0,
|
||||
"fresh":0,
|
||||
"stale":0,
|
||||
"never_computed":0,
|
||||
"failed":0,
|
||||
"auto_refresh":{
|
||||
"enabled":true,
|
||||
"interval_seconds":60,
|
||||
@@ -11690,15 +11786,52 @@ mod tests {
|
||||
.unwrap()
|
||||
});
|
||||
|
||||
let status = table.function_column_status("embedding").await.unwrap();
|
||||
let statuses = table.function_column_statuses(None).await.unwrap();
|
||||
assert_eq!(statuses.len(), 4);
|
||||
let status = &statuses[0];
|
||||
assert_eq!(
|
||||
status.auto_refresh.state,
|
||||
crate::table::FunctionColumnAutoRefreshState::Blocked
|
||||
crate::table::FunctionColumnAutoRefreshState::Queued
|
||||
);
|
||||
assert_eq!(status.auto_refresh.failure_streak, 3);
|
||||
assert_eq!(status.auto_refresh.effective_cooldown_millis, Some(8000));
|
||||
assert_eq!(
|
||||
status.auto_refresh.blocked_reason,
|
||||
status.auto_refresh.active_job_id.as_deref(),
|
||||
Some("job-queued")
|
||||
);
|
||||
assert_eq!(
|
||||
status.auto_refresh.next_run_at_reason.as_deref(),
|
||||
Some("job_queued_not_started")
|
||||
);
|
||||
assert_eq!(
|
||||
status
|
||||
.last_refresh_job
|
||||
.as_ref()
|
||||
.map(|job| job.state.as_str()),
|
||||
Some("cancelled")
|
||||
);
|
||||
let failure_job = status.last_failure_job.as_ref().unwrap();
|
||||
assert_eq!(failure_job.job_id, "job-failed");
|
||||
let failure = failure_job.failure.as_ref().unwrap();
|
||||
assert_eq!(failure.phase.as_deref(), Some("finalize"));
|
||||
assert_eq!(failure.message.as_deref(), Some("worker timed out"));
|
||||
assert_eq!(failure.retryable, Some(true));
|
||||
assert_eq!(failure.sql_job_id.as_deref(), Some("sql-job-7"));
|
||||
|
||||
assert_eq!(statuses[1].auto_refresh.failure_streak, 5);
|
||||
assert_eq!(
|
||||
statuses[1].auto_refresh.blocked_reason,
|
||||
Some(crate::table::FunctionColumnAutoRefreshBlockedReason::FailureThresholdReached)
|
||||
);
|
||||
assert_eq!(
|
||||
statuses[2].auto_refresh.blocked_reason,
|
||||
Some(crate::table::FunctionColumnAutoRefreshBlockedReason::BindingRevisionMissing)
|
||||
);
|
||||
assert_eq!(statuses[3].auto_refresh.failure_streak, 3);
|
||||
assert_eq!(
|
||||
statuses[3].auto_refresh.effective_cooldown_millis,
|
||||
Some(8000)
|
||||
);
|
||||
assert_eq!(
|
||||
statuses[3].auto_refresh.blocked_reason,
|
||||
Some(crate::table::FunctionColumnAutoRefreshBlockedReason::PreparationBackoff)
|
||||
);
|
||||
}
|
||||
|
||||
@@ -5,6 +5,7 @@
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
use crate::error::JobFailure;
|
||||
use crate::function::FunctionVersionRef;
|
||||
use crate::{Error, Result};
|
||||
|
||||
@@ -23,6 +24,9 @@ pub struct FunctionColumnLastRefreshJob {
|
||||
pub automatic: Option<bool>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub completed_at_millis: Option<i64>,
|
||||
/// Why this Job failed, when the server retained a structured failure.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub failure: Option<JobFailure>,
|
||||
}
|
||||
|
||||
/// Automatic-refresh scheduling state that the server can prove.
|
||||
@@ -31,6 +35,7 @@ pub struct FunctionColumnLastRefreshJob {
|
||||
pub enum FunctionColumnAutoRefreshState {
|
||||
Disabled,
|
||||
Running,
|
||||
Queued,
|
||||
UpToDate,
|
||||
Ready,
|
||||
WaitingInterval,
|
||||
@@ -44,6 +49,7 @@ impl FunctionColumnAutoRefreshState {
|
||||
match self {
|
||||
Self::Disabled => "disabled",
|
||||
Self::Running => "running",
|
||||
Self::Queued => "queued",
|
||||
Self::UpToDate => "up_to_date",
|
||||
Self::Ready => "ready",
|
||||
Self::WaitingInterval => "waiting_interval",
|
||||
@@ -64,6 +70,8 @@ pub enum FunctionColumnAutoRefreshBlockedReason {
|
||||
TableCapacity,
|
||||
Cancelled,
|
||||
PreparationBackoff,
|
||||
FailureThresholdReached,
|
||||
BindingRevisionMissing,
|
||||
}
|
||||
|
||||
impl FunctionColumnAutoRefreshBlockedReason {
|
||||
@@ -75,6 +83,8 @@ impl FunctionColumnAutoRefreshBlockedReason {
|
||||
Self::TableCapacity => "table_capacity",
|
||||
Self::Cancelled => "cancelled",
|
||||
Self::PreparationBackoff => "preparation_backoff",
|
||||
Self::FailureThresholdReached => "failure_threshold_reached",
|
||||
Self::BindingRevisionMissing => "binding_revision_missing",
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -113,6 +123,10 @@ pub struct FunctionColumnStatus {
|
||||
pub failed: u64,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub last_refresh_job: Option<FunctionColumnLastRefreshJob>,
|
||||
/// Newest failed refresh Job. Unlike `last_refresh_job`, this remains
|
||||
/// visible when the latest attempt was cancelled.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub last_failure_job: Option<FunctionColumnLastRefreshJob>,
|
||||
pub auto_refresh: FunctionColumnAutoRefreshStatus,
|
||||
}
|
||||
|
||||
@@ -216,6 +230,7 @@ mod tests {
|
||||
never_computed: 1,
|
||||
failed: 1,
|
||||
last_refresh_job: None,
|
||||
last_failure_job: None,
|
||||
auto_refresh: FunctionColumnAutoRefreshStatus {
|
||||
enabled: true,
|
||||
interval_seconds: Some(60),
|
||||
@@ -253,6 +268,7 @@ mod tests {
|
||||
state: "finished".to_string(),
|
||||
automatic: Some(true),
|
||||
completed_at_millis: Some(1_790_000_000_000),
|
||||
failure: None,
|
||||
});
|
||||
let encoded = serde_json::to_value(value).unwrap();
|
||||
assert_eq!(encoded["last_refresh_job"]["job_id"], "job-7");
|
||||
@@ -264,11 +280,31 @@ mod tests {
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn up_to_date_wire_is_stable() {
|
||||
assert_eq!(
|
||||
serde_json::to_string(&FunctionColumnAutoRefreshState::UpToDate).unwrap(),
|
||||
"\"up_to_date\""
|
||||
);
|
||||
fn scheduler_wire_names_are_stable() {
|
||||
for (state, expected) in [
|
||||
(FunctionColumnAutoRefreshState::Queued, "queued"),
|
||||
(FunctionColumnAutoRefreshState::UpToDate, "up_to_date"),
|
||||
] {
|
||||
assert_eq!(
|
||||
serde_json::to_string(&state).unwrap(),
|
||||
format!("\"{expected}\"")
|
||||
);
|
||||
}
|
||||
for (reason, expected) in [
|
||||
(
|
||||
FunctionColumnAutoRefreshBlockedReason::FailureThresholdReached,
|
||||
"failure_threshold_reached",
|
||||
),
|
||||
(
|
||||
FunctionColumnAutoRefreshBlockedReason::BindingRevisionMissing,
|
||||
"binding_revision_missing",
|
||||
),
|
||||
] {
|
||||
assert_eq!(
|
||||
serde_json::to_string(&reason).unwrap(),
|
||||
format!("\"{expected}\"")
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
Reference in New Issue
Block a user