diff --git a/docs/src/js/classes/Table.md b/docs/src/js/classes/Table.md index 27acc9847..aba747a7c 100644 --- a/docs/src/js/classes/Table.md +++ b/docs/src/js/classes/Table.md @@ -704,6 +704,47 @@ so this is safe to call repeatedly. *** +### functionColumnStatus() + +```ts +abstract functionColumnStatus(column): Promise +``` + +Return exact freshness status for one Function-backed output column. + +#### Parameters + +* **column**: `string` + +#### Returns + +`Promise`<[`FunctionColumnStatus`](../interfaces/FunctionColumnStatus.md)> + +*** + +### functionColumnStatuses() + +```ts +abstract functionColumnStatuses(column?): Promise +``` + +Return exact freshness status for Function-backed output columns. + +All results describe one server table version. Omit `column` for every +Function output. Rows already failed remain visible and do not prevent +`waitFresh` from completing. A handle pinned to an exact version requests +that version; a latest handle retains normal read-freshness behavior. + +#### Parameters + +* **column?**: `string` + +#### Returns + +`Promise`<[`FunctionColumnStatus`](../interfaces/FunctionColumnStatus.md)[]> + +*** + ### functionErrors() ```ts @@ -1598,3 +1639,26 @@ Waits for asynchronous indexing to complete on the table. #### Returns `Promise`<`void`> + +*** + +### waitFresh() + +```ts +abstract waitFresh(column, timeoutSeconds?): Promise +``` + +Poll until no stale or never-computed rows remain. + +Only HTTP 503 with the exact finalization-pending error code is retried; +every other error propagates. Failed rows do not block completion. + +#### Parameters + +* **column**: `string` + +* **timeoutSeconds?**: `number` + +#### Returns + +`Promise`<[`FunctionColumnStatus`](../interfaces/FunctionColumnStatus.md)> diff --git a/docs/src/js/globals.md b/docs/src/js/globals.md index 50f55f937..19d7ceb97 100644 --- a/docs/src/js/globals.md +++ b/docs/src/js/globals.md @@ -95,6 +95,10 @@ - [FtsToken](interfaces/FtsToken.md) - [FullTextQuery](interfaces/FullTextQuery.md) - [FullTextSearchOptions](interfaces/FullTextSearchOptions.md) +- [FunctionColumnAutoRefreshStatus](interfaces/FunctionColumnAutoRefreshStatus.md) +- [FunctionColumnLastRefreshJob](interfaces/FunctionColumnLastRefreshJob.md) +- [FunctionColumnStatus](interfaces/FunctionColumnStatus.md) +- [FunctionColumnVersionStatus](interfaces/FunctionColumnVersionStatus.md) - [FunctionErrorFragment](interfaces/FunctionErrorFragment.md) - [FunctionErrorRecord](interfaces/FunctionErrorRecord.md) - [FunctionErrors](interfaces/FunctionErrors.md) diff --git a/docs/src/js/interfaces/FunctionColumnAutoRefreshStatus.md b/docs/src/js/interfaces/FunctionColumnAutoRefreshStatus.md new file mode 100644 index 000000000..bc8f7daab --- /dev/null +++ b/docs/src/js/interfaces/FunctionColumnAutoRefreshStatus.md @@ -0,0 +1,95 @@ +[**@lancedb/lancedb**](../README.md) • **Docs** + +*** + +[@lancedb/lancedb](../globals.md) / FunctionColumnAutoRefreshStatus + +# Interface: FunctionColumnAutoRefreshStatus + +Automatic-refresh projection for a Function output. + +## Properties + +### activeJobId? + +```ts +optional activeJobId: string; +``` + +*** + +### blockedReason? + +```ts +optional blockedReason: + | "table_has_mem_wal" + | "inputs_not_settled" + | "dependency_in_flight" + | "table_capacity" + | "cancelled" + | "preparation_backoff"; +``` + +*** + +### effectiveCooldownMillis? + +```ts +optional effectiveCooldownMillis: bigint; +``` + +*** + +### enabled + +```ts +enabled: boolean; +``` + +*** + +### failureStreak + +```ts +failureStreak: number; +``` + +*** + +### intervalSeconds? + +```ts +optional intervalSeconds: bigint; +``` + +*** + +### nextRunAtMillis + +```ts +nextRunAtMillis: null | bigint; +``` + +*** + +### nextRunAtReason? + +```ts +optional nextRunAtReason: string; +``` + +*** + +### state + +```ts +state: + | "running" + | "disabled" + | "ready" + | "up_to_date" + | "waiting_interval" + | "waiting_backoff" + | "blocked" + | "unobservable"; +``` diff --git a/docs/src/js/interfaces/FunctionColumnLastRefreshJob.md b/docs/src/js/interfaces/FunctionColumnLastRefreshJob.md new file mode 100644 index 000000000..1d7d3b300 --- /dev/null +++ b/docs/src/js/interfaces/FunctionColumnLastRefreshJob.md @@ -0,0 +1,41 @@ +[**@lancedb/lancedb**](../README.md) • **Docs** + +*** + +[@lancedb/lancedb](../globals.md) / FunctionColumnLastRefreshJob + +# Interface: FunctionColumnLastRefreshJob + +Latest matching refresh Job projected for a Function output. + +## Properties + +### automatic? + +```ts +optional automatic: boolean; +``` + +*** + +### completedAtMillis? + +```ts +optional completedAtMillis: bigint; +``` + +*** + +### jobId + +```ts +jobId: string; +``` + +*** + +### state + +```ts +state: string; +``` diff --git a/docs/src/js/interfaces/FunctionColumnStatus.md b/docs/src/js/interfaces/FunctionColumnStatus.md new file mode 100644 index 000000000..755a8af81 --- /dev/null +++ b/docs/src/js/interfaces/FunctionColumnStatus.md @@ -0,0 +1,97 @@ +[**@lancedb/lancedb**](../README.md) • **Docs** + +*** + +[@lancedb/lancedb](../globals.md) / FunctionColumnStatus + +# Interface: FunctionColumnStatus + +Exact four-way status of one Function output at one table version. + +## Properties + +### autoRefresh + +```ts +autoRefresh: FunctionColumnAutoRefreshStatus; +``` + +*** + +### bindingId + +```ts +bindingId: string; +``` + +*** + +### column + +```ts +column: string; +``` + +*** + +### failed + +```ts +failed: bigint; +``` + +*** + +### fresh + +```ts +fresh: bigint; +``` + +*** + +### function + +```ts +function: FunctionColumnVersionStatus; +``` + +*** + +### lastRefreshJob? + +```ts +optional lastRefreshJob: FunctionColumnLastRefreshJob; +``` + +*** + +### neverComputed + +```ts +neverComputed: bigint; +``` + +*** + +### rows + +```ts +rows: bigint; +``` + +*** + +### stale + +```ts +stale: bigint; +``` + +*** + +### tableVersion + +```ts +tableVersion: bigint; +``` diff --git a/docs/src/js/interfaces/FunctionColumnVersionStatus.md b/docs/src/js/interfaces/FunctionColumnVersionStatus.md new file mode 100644 index 000000000..fa02139d2 --- /dev/null +++ b/docs/src/js/interfaces/FunctionColumnVersionStatus.md @@ -0,0 +1,57 @@ +[**@lancedb/lancedb**](../README.md) • **Docs** + +*** + +[@lancedb/lancedb](../globals.md) / FunctionColumnVersionStatus + +# Interface: FunctionColumnVersionStatus + +Exact immutable Function revision attached to a table output. + +## Properties + +### location + +```ts +location: string; +``` + +*** + +### manifestDigest + +```ts +manifestDigest: string; +``` + +*** + +### name + +```ts +name: string; +``` + +*** + +### namespacePath + +```ts +namespacePath: string[]; +``` + +*** + +### objectId + +```ts +objectId: string; +``` + +*** + +### version + +```ts +version: string; +``` diff --git a/docs/src/python/python.md b/docs/src/python/python.md index 7f28d8804..eccaae6ce 100644 --- a/docs/src/python/python.md +++ b/docs/src/python/python.md @@ -523,6 +523,14 @@ with BlobReader(read_range, len(payload)) as reader: ::: lancedb.FunctionErrorFragment +::: lancedb.FunctionColumnStatus + +::: lancedb.FunctionColumnVersionStatus + +::: lancedb.FunctionColumnLastRefreshJob + +::: lancedb.FunctionColumnAutoRefreshStatus + ::: lancedb.job.Job ::: lancedb.job.AsyncJob diff --git a/docs/web/feature-engineering/refresh-column.mdx b/docs/web/feature-engineering/refresh-column.mdx index b3fcdc937..e1a9c43f3 100644 --- a/docs/web/feature-engineering/refresh-column.mdx +++ b/docs/web/feature-engineering/refresh-column.mdx @@ -150,6 +150,57 @@ does not raise, which is what makes it safe in a loop; `wait()` raises A long backfill can therefore be started, left, and checked on from a different process. +{/* +Draft for the Function freshness release. Keep this section hidden until the +Enterprise status endpoint and its integration acceptance are deployed. + +### Check Function column freshness {#freshness} + +For a Function-backed column, the SDK can report how many live rows are +`fresh`, `stale`, `never_computed`, or `failed` at the table version it +reads. `wait_fresh`, `waitFresh`, and `wait_for_function_column` poll until no +stale or never-computed rows remain. Rows that already failed stay in the +returned status and do not keep the wait pending, so inspect `failed` after a +successful wait: + + +```python Python icon="python" +status = table.column_status("embedding") +print(status.fresh, status.stale, status.never_computed, status.failed) + +final_status = table.wait_fresh("embedding") +print(final_status.failed) +``` + +```typescript TypeScript icon="typescript" +const status = await table.functionColumnStatus("embedding"); +console.log( + status.fresh, + status.stale, + status.neverComputed, + status.failed, +); + +const finalStatus = await table.waitFresh("embedding"); +console.log(finalStatus.failed); +``` + +```rust Rust icon="rust" +let status = table.function_column_status("embedding").await?; +println!( + "{} {} {} {}", + status.fresh, status.stale, status.never_computed, status.failed +); + +let final_status = table.wait_for_function_column("embedding", None).await?; +println!("{}", final_status.failed); +``` + + +The wait helpers do not submit or cancel a refresh. They retry only the typed +finalization-pending response and propagate other errors. +*/} + ### Configure workers and resources {#resources} SQL exposes the worker budget directly, which is the lever when a backfill is diff --git a/nodejs/__test__/remote.test.ts b/nodejs/__test__/remote.test.ts index 280905897..6c160ed03 100644 --- a/nodejs/__test__/remote.test.ts +++ b/nodejs/__test__/remote.test.ts @@ -647,6 +647,69 @@ describe("remote connection", () => { ]); }); + it("passes a pinned version and decodes Function scheduler status", async () => { + const queries: Record[] = []; + await withMockDatabase( + (req, res) => { + const url = new URL(req.url ?? "/", "http://localhost"); + if (url.pathname.endsWith("/describe/")) { + res.writeHead(200, { "Content-Type": "application/json" }).end( + JSON.stringify({ + name: "docs", + version: 42, + schema: { fields: [] }, + }), + ); + return; + } + if (url.pathname === "/v1/table/docs/function_columns/status") { + queries.push(Object.fromEntries(url.searchParams.entries())); + res.writeHead(200, { "Content-Type": "application/json" }).end( + `[{ + "column":"embedding", + "binding_id":"fb_01", + "function":{ + "name":"embed", + "object_id":"embed", + "location":"objects/embed", + "version":"7", + "manifest_digest":"sha256:abc" + }, + "table_version":42, + "rows":4, + "fresh":2, + "stale":2, + "never_computed":0, + "failed":0, + "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 + } + }]`, + ); + return; + } + res.writeHead(404).end(); + }, + async (db) => { + const table = await db.openTable("docs"); + 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(queries).toEqual([{ column: "embedding", version: "42" }]); + }); + it("forwards refresh options to the backfill request", async () => { const bodies: unknown[] = []; await withMockDatabase( diff --git a/nodejs/lancedb/index.ts b/nodejs/lancedb/index.ts index 6bd46c9c8..35697dc8c 100644 --- a/nodejs/lancedb/index.ts +++ b/nodejs/lancedb/index.ts @@ -154,6 +154,10 @@ export { export { Table, + FunctionColumnAutoRefreshStatus, + FunctionColumnLastRefreshJob, + FunctionColumnStatus, + FunctionColumnVersionStatus, Branches, BranchColumnSummary, BranchColumnChange, diff --git a/nodejs/lancedb/table.ts b/nodejs/lancedb/table.ts index d56f66cde..99eb0d1f9 100644 --- a/nodejs/lancedb/table.ts +++ b/nodejs/lancedb/table.ts @@ -258,6 +258,66 @@ export interface RTEWriteSpec { writerConfigDefaults?: Record; } +/** Exact immutable Function revision attached to a table output. */ +export interface FunctionColumnVersionStatus { + name: string; + namespacePath: string[]; + objectId: string; + location: string; + version: string; + manifestDigest: string; +} + +/** Latest matching refresh Job projected for a Function output. */ +export interface FunctionColumnLastRefreshJob { + jobId: string; + state: string; + automatic?: boolean; + completedAtMillis?: bigint; +} + +/** Automatic-refresh projection for a Function output. */ +export interface FunctionColumnAutoRefreshStatus { + enabled: boolean; + intervalSeconds?: bigint; + activeJobId?: string; + state: + | "disabled" + | "running" + | "ready" + | "up_to_date" + | "waiting_interval" + | "waiting_backoff" + | "blocked" + | "unobservable"; + failureStreak: number; + effectiveCooldownMillis?: bigint; + blockedReason?: + | "table_has_mem_wal" + | "inputs_not_settled" + | "dependency_in_flight" + | "table_capacity" + | "cancelled" + | "preparation_backoff"; + nextRunAtMillis: bigint | null; + nextRunAtReason?: string; +} + +/** Exact four-way status of one Function output at one table version. */ +export interface FunctionColumnStatus { + column: string; + bindingId: string; + function: FunctionColumnVersionStatus; + tableVersion: bigint; + rows: bigint; + fresh: bigint; + stale: bigint; + neverComputed: bigint; + failed: bigint; + lastRefreshJob?: FunctionColumnLastRefreshJob; + autoRefresh: FunctionColumnAutoRefreshStatus; +} + /** * A Table is a collection of Records in a LanceDB Database. * @@ -780,6 +840,32 @@ export abstract class Table { options?: FunctionErrorsOptions, ): Promise; + /** + * Return exact freshness status for Function-backed output columns. + * + * All results describe one server table version. Omit `column` for every + * Function output. Rows already failed remain visible and do not prevent + * `waitFresh` from completing. A handle pinned to an exact version requests + * that version; a latest handle retains normal read-freshness behavior. + */ + abstract functionColumnStatuses( + column?: string, + ): Promise; + + /** Return exact freshness status for one Function-backed output column. */ + abstract functionColumnStatus(column: string): Promise; + + /** + * Poll until no stale or never-computed rows remain. + * + * Only HTTP 503 with the exact finalization-pending error code is retried; + * every other error propagates. Failed rows do not block completion. + */ + abstract waitFresh( + column: string, + timeoutSeconds?: number, + ): Promise; + /** * Recompute this table's contents from its materialized-view definition. * @@ -1557,6 +1643,31 @@ export class LocalTable extends Table { return await this.inner.functionErrors(options); } + async functionColumnStatuses( + column?: string, + ): Promise { + return (await this.inner.functionColumnStatuses( + column, + )) as FunctionColumnStatus[]; + } + + async functionColumnStatus(column: string): Promise { + return (await this.inner.functionColumnStatus( + column, + )) as FunctionColumnStatus; + } + + async waitFresh( + column: string, + timeoutSeconds?: number, + ): Promise { + validateNonNegativeInteger(timeoutSeconds, "timeoutSeconds"); + return (await this.inner.waitForFunctionColumn( + column, + timeoutSeconds, + )) as FunctionColumnStatus; + } + async refreshMaterializedView( sourceVersion?: number, full?: boolean, diff --git a/nodejs/src/table.rs b/nodejs/src/table.rs index 5b52c0d16..f6d3e538e 100644 --- a/nodejs/src/table.rs +++ b/nodejs/src/table.rs @@ -534,6 +534,55 @@ impl Table { Ok(errors.into()) } + #[napi(catch_unwind)] + pub async fn function_column_statuses( + &self, + column: Option, + ) -> napi::Result> { + let statuses = self + .inner_ref()? + .function_column_statuses(column.as_deref()) + .await + .default_error()?; + Ok(statuses.into_iter().map(Into::into).collect()) + } + + #[napi(catch_unwind)] + pub async fn function_column_status( + &self, + column: String, + ) -> napi::Result { + let status = self + .inner_ref()? + .function_column_status(&column) + .await + .default_error()?; + Ok(status.into()) + } + + #[napi(catch_unwind)] + pub async fn wait_for_function_column( + &self, + column: String, + timeout_seconds: Option, + ) -> napi::Result { + let timeout = timeout_seconds + .map(|seconds| { + u64::try_from(seconds) + .map(std::time::Duration::from_secs) + .map_err(|_| { + napi::Error::from_reason("timeoutSeconds must be a non-negative integer") + }) + }) + .transpose()?; + let status = self + .inner_ref()? + .wait_for_function_column(&column, timeout) + .await + .default_error()?; + Ok(status.into()) + } + #[napi(catch_unwind)] pub async fn refresh_materialized_view( &self, @@ -1799,6 +1848,108 @@ pub struct FunctionErrors { pub truncated: bool, } +/// One exact immutable Function revision attached to a table output. +#[napi(object)] +#[derive(Clone, Debug)] +pub struct FunctionColumnVersionStatus { + pub name: String, + pub namespace_path: Vec, + pub object_id: String, + pub location: String, + pub version: String, + pub manifest_digest: String, +} + +/// Latest matching refresh Job projected for a Function output. +#[napi(object)] +#[derive(Clone, Debug)] +pub struct FunctionColumnLastRefreshJob { + pub job_id: String, + pub state: String, + pub automatic: Option, + pub completed_at_millis: Option, +} + +/// Automatic-refresh projection for a Function output. +#[napi(object)] +#[derive(Clone, Debug)] +pub struct FunctionColumnAutoRefreshStatus { + pub enabled: bool, + pub interval_seconds: Option, + pub active_job_id: Option, + pub state: String, + pub failure_streak: u32, + pub effective_cooldown_millis: Option, + pub blocked_reason: Option, + pub next_run_at_millis: Option, + pub next_run_at_reason: Option, +} + +/// Exact four-way status of one Function output at one table version. +#[napi(object)] +#[derive(Clone, Debug)] +pub struct FunctionColumnStatus { + pub column: String, + pub binding_id: String, + pub function: FunctionColumnVersionStatus, + pub table_version: BigInt, + pub rows: BigInt, + pub fresh: BigInt, + pub stale: BigInt, + pub never_computed: BigInt, + pub failed: BigInt, + pub last_refresh_job: Option, + pub auto_refresh: FunctionColumnAutoRefreshStatus, +} + +impl From for FunctionColumnStatus { + fn from(status: lancedb::table::FunctionColumnStatus) -> Self { + Self { + column: status.column, + binding_id: status.binding_id, + function: FunctionColumnVersionStatus { + name: status.function.name, + namespace_path: status.function.namespace_path, + object_id: status.function.object_id, + location: status.function.location, + version: status.function.version, + manifest_digest: status.function.manifest_digest, + }, + table_version: BigInt::from(status.table_version), + rows: BigInt::from(status.rows), + fresh: BigInt::from(status.fresh), + 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), + }), + auto_refresh: FunctionColumnAutoRefreshStatus { + enabled: status.auto_refresh.enabled, + interval_seconds: status.auto_refresh.interval_seconds.map(BigInt::from), + active_job_id: status.auto_refresh.active_job_id, + state: status.auto_refresh.state.as_str().to_string(), + failure_streak: status.auto_refresh.failure_streak, + effective_cooldown_millis: status + .auto_refresh + .effective_cooldown_millis + .map(BigInt::from), + blocked_reason: status + .auto_refresh + .blocked_reason + .map(|reason| reason.as_str().to_string()), + next_run_at_millis: status.auto_refresh.next_run_at_millis.map(BigInt::from), + next_run_at_reason: status.auto_refresh.next_run_at_reason, + }, + } + } +} + impl From for FunctionErrors { fn from(errors: lancedb::function::FunctionErrors) -> Self { Self { diff --git a/python/python/lancedb/__init__.py b/python/python/lancedb/__init__.py index 2208fd612..31285d100 100644 --- a/python/python/lancedb/__init__.py +++ b/python/python/lancedb/__init__.py @@ -17,6 +17,16 @@ __version__ = str(_Version(_native_version)) from ._lancedb import connect as lancedb_connect from ._lancedb import FtsToken +from ._lancedb import ( + FunctionColumnAutoRefreshStatus as FunctionColumnAutoRefreshStatus, +) +from ._lancedb import ( + FunctionColumnLastRefreshJob as FunctionColumnLastRefreshJob, +) +from ._lancedb import FunctionColumnStatus as FunctionColumnStatus +from ._lancedb import ( + FunctionColumnVersionStatus as FunctionColumnVersionStatus, +) from ._lancedb import FunctionErrorFragment as FunctionErrorFragment from ._lancedb import FunctionErrorRecord as FunctionErrorRecord from ._lancedb import FunctionErrors as FunctionErrors @@ -660,6 +670,10 @@ __all__ = [ "AsyncTable", "CompactionOptions", "FtsToken", + "FunctionColumnAutoRefreshStatus", + "FunctionColumnLastRefreshJob", + "FunctionColumnStatus", + "FunctionColumnVersionStatus", "col", "Expr", "func", diff --git a/python/python/lancedb/_lancedb.pyi b/python/python/lancedb/_lancedb.pyi index dfb887461..07306c1d9 100644 --- a/python/python/lancedb/_lancedb.pyi +++ b/python/python/lancedb/_lancedb.pyi @@ -671,6 +671,13 @@ class Table: column: Optional[str] = None, limit: Optional[int] = None, ) -> FunctionErrors: ... + async def column_statuses( + self, column: Optional[str] = None + ) -> list[FunctionColumnStatus]: ... + async def column_status(self, column: str) -> FunctionColumnStatus: ... + async def wait_column_fresh( + self, column: str, timeout: Optional[timedelta] = None + ) -> FunctionColumnStatus: ... async def refresh_materialized_view( self, source_version: Optional[int] = None, @@ -1088,6 +1095,44 @@ class FunctionErrors: fragments: list[FunctionErrorFragment] truncated: bool +class FunctionColumnVersionStatus: + name: str + namespace_path: list[str] + object_id: str + location: str + version: str + manifest_digest: str + +class FunctionColumnLastRefreshJob: + job_id: str + state: str + automatic: Optional[bool] + completed_at_millis: Optional[int] + +class FunctionColumnAutoRefreshStatus: + enabled: bool + interval_seconds: Optional[int] + active_job_id: Optional[str] + state: str + failure_streak: int + effective_cooldown_millis: Optional[int] + blocked_reason: Optional[str] + next_run_at_millis: Optional[int] + next_run_at_reason: Optional[str] + +class FunctionColumnStatus: + column: str + binding_id: str + function: FunctionColumnVersionStatus + table_version: int + rows: int + fresh: int + stale: int + never_computed: int + failed: int + last_refresh_job: Optional[FunctionColumnLastRefreshJob] + auto_refresh: FunctionColumnAutoRefreshStatus + class RefreshMaterializedViewResult: @staticmethod def from_json(value: str) -> RefreshMaterializedViewResult: ... diff --git a/python/python/lancedb/remote/table.py b/python/python/lancedb/remote/table.py index af648c45c..3f6bff45e 100644 --- a/python/python/lancedb/remote/table.py +++ b/python/python/lancedb/remote/table.py @@ -34,6 +34,7 @@ from lancedb._lancedb import ( DeleteResult, DropColumnsResult, FtsToken, + FunctionColumnStatus, IndexConfig, RTEWriteSpec, MergeResult, @@ -1233,6 +1234,19 @@ class RemoteTable(Table): self._table.function_errors(job_id=job_id, column=column, limit=limit) ) + def column_statuses( + self, column: Optional[str] = None + ) -> list[FunctionColumnStatus]: + return LOOP.run(self._table.column_statuses(column=column)) + + def column_status(self, column: str) -> FunctionColumnStatus: + return LOOP.run(self._table.column_status(column)) + + def wait_fresh( + self, column: str, timeout: Optional[timedelta] = None + ) -> FunctionColumnStatus: + return LOOP.run(self._table.wait_fresh(column, timeout)) + def alter_columns( self, *alterations: Iterable[Dict[str, str]] ) -> AlterColumnsResult: diff --git a/python/python/lancedb/table.py b/python/python/lancedb/table.py index 609e116ba..762f826c2 100644 --- a/python/python/lancedb/table.py +++ b/python/python/lancedb/table.py @@ -35,7 +35,7 @@ from urllib.parse import urlparse from lancedb.scannable import _hf_batches, _register_optional_converters, to_scannable from . import __version__ -from ._lancedb import _parse_field_path +from ._lancedb import FunctionColumnStatus, _parse_field_path from ._blob import ( BlobFile, _normalize_blob_descriptor, @@ -3504,6 +3504,36 @@ class Table(ABC): ... print("listing stopped at the limit") """ + @abstractmethod + def column_statuses( + self, column: Optional[str] = None + ) -> list[FunctionColumnStatus]: + """Return exact status for Function-backed output columns. + + All statuses describe one server table version. ``column=None`` returns + every Function output. This operation is available on LanceDB + Enterprise tables. A table handle pinned to an exact version requests + that version; a latest handle retains the normal read-freshness + behavior. + """ + + @abstractmethod + def column_status(self, column: str) -> FunctionColumnStatus: + """Return exact status for one Function-backed output column.""" + + @abstractmethod + def wait_fresh( + self, column: str, timeout: Optional[timedelta] = None + ) -> FunctionColumnStatus: + """Poll until the column has no stale or never-computed rows. + + Rows already failed remain visible in the returned status and do not + keep this method waiting. The method never submits or cancels work. + Transport errors and unrecoverable status-evidence errors propagate. + A typed finalization-pending response is retried; arbitrary failure + responses are not. + """ + @abstractmethod def alter_columns(self, *alterations: Iterable[Dict[str, str]]): """ @@ -5604,6 +5634,19 @@ class LanceTable(Table): self._table.function_errors(job_id=job_id, column=column, limit=limit) ) + def column_statuses( + self, column: Optional[str] = None + ) -> list[FunctionColumnStatus]: + return LOOP.run(self._table.column_statuses(column=column)) + + def column_status(self, column: str) -> FunctionColumnStatus: + return LOOP.run(self._table.column_status(column)) + + def wait_fresh( + self, column: str, timeout: Optional[timedelta] = None + ) -> FunctionColumnStatus: + return LOOP.run(self._table.wait_fresh(column, timeout)) + def alter_columns( self, *alterations: Iterable[Dict[str, str]] ) -> AlterColumnsResult: @@ -8168,6 +8211,32 @@ class AsyncTable: job_id=job_id, column=column, limit=limit ) + async def column_statuses( + self, column: Optional[str] = None + ) -> list[FunctionColumnStatus]: + """Return exact status for Function-backed output columns. + + All results describe one server table version. ``column=None`` returns + every Function output. A handle pinned to an exact version requests + that version. + """ + return await self._inner.column_statuses(column=column) + + async def column_status(self, column: str) -> FunctionColumnStatus: + """Return exact status for one Function-backed output column.""" + return await self._inner.column_status(column) + + async def wait_fresh( + self, column: str, timeout: Optional[timedelta] = None + ) -> FunctionColumnStatus: + """Poll until no stale or never-computed rows remain. + + Failed rows do not block completion. Server and transport errors are + not retried. Only the typed finalization-pending response is retried + with bounded backoff; arbitrary 500 responses surface immediately. + """ + return await self._inner.wait_column_fresh(column, timeout) + async def refresh_column_async( self, column: str, diff --git a/python/python/tests/test_remote_db.py b/python/python/tests/test_remote_db.py index 5224e5482..fd933b43c 100644 --- a/python/python/tests/test_remote_db.py +++ b/python/python/tests/test_remote_db.py @@ -2288,6 +2288,137 @@ def test_remote_function_errors_lists_the_rows_a_refresh_skipped(attempts): assert everything.truncated is True +def test_remote_function_column_status_passes_pinned_version(): + queries = [] + status = { + "column": "embedding", + "binding_id": "fb_01", + "function": { + "name": "embed", + "object_id": "embed", + "location": "objects/embed", + "version": "7", + "manifest_digest": "sha256:abc", + }, + "table_version": 42, + "rows": 4, + "fresh": 2, + "stale": 2, + "never_computed": 0, + "failed": 0, + "auto_refresh": { + "enabled": True, + "interval_seconds": 60, + "active_job_id": None, + "state": "blocked", + "failure_streak": 3, + "effective_cooldown_millis": 8000, + "blocked_reason": "preparation_backoff", + "next_run_at_millis": None, + }, + } + + def handler(request): + parsed = urlparse(request.path) + if parsed.path == "/v1/table/test/describe/": + request.send_response(200) + request.send_header("Content-Type", "application/json") + request.end_headers() + request.wfile.write( + json.dumps({"version": 42, "schema": {"fields": []}}).encode() + ) + elif parsed.path == "/v1/table/test/function_columns/status": + queries.append(parse_qs(parsed.query)) + request.send_response(200) + request.send_header("Content-Type", "application/json") + request.end_headers() + request.wfile.write(json.dumps([status]).encode()) + else: + request.send_response(404) + request.end_headers() + + with mock_lancedb_connection(handler) as db: + table = db.open_table("test") + table.checkout(42) + result = table.column_status("embedding") + + assert queries == [{"column": ["embedding"], "version": ["42"]}] + assert result.table_version == 42 + assert (result.fresh, result.stale, result.never_computed, result.failed) == ( + 2, + 2, + 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" + + +def test_remote_wait_fresh_retries_only_typed_finalization_pending(): + calls = 0 + status = { + "column": "embedding", + "binding_id": "fb_01", + "function": { + "name": "embed", + "object_id": "embed", + "location": "objects/embed", + "version": "7", + "manifest_digest": "sha256:abc", + }, + "table_version": 42, + "rows": 1, + "fresh": 1, + "stale": 0, + "never_computed": 0, + "failed": 0, + "auto_refresh": { + "enabled": False, + "state": "disabled", + "failure_streak": 0, + "next_run_at_millis": None, + }, + } + + def handler(request): + nonlocal calls + parsed = urlparse(request.path) + if parsed.path == "/v1/table/test/describe/": + request.send_response(200) + request.send_header("Content-Type", "application/json") + request.end_headers() + request.wfile.write( + json.dumps({"version": 42, "schema": {"fields": []}}).encode() + ) + elif parsed.path == "/v1/table/test/function_columns/status": + calls += 1 + if calls == 1: + request.send_response(503) + request.send_header( + "x-lancedb-error-code", + "function_column_status_finalization_pending", + ) + request.end_headers() + request.wfile.write(b"finalization pending") + else: + request.send_response(200) + request.send_header("Content-Type", "application/json") + request.end_headers() + request.wfile.write(json.dumps([status]).encode()) + else: + request.send_response(404) + request.end_headers() + + with mock_lancedb_connection(handler) as db: + table = db.open_table("test") + result = table.wait_fresh("embedding", timeout=timedelta(seconds=2)) + + assert calls == 2 + assert result.fresh == result.rows == 1 + + @pytest.mark.asyncio @pytest.mark.parametrize("sync", [False, True]) @pytest.mark.parametrize("failure", ["send", "read"]) diff --git a/python/python/tests/test_table.py b/python/python/tests/test_table.py index 7146ed86b..e3b69ced1 100644 --- a/python/python/tests/test_table.py +++ b/python/python/tests/test_table.py @@ -6036,6 +6036,16 @@ def test_function_errors_are_remote_only(tmp_path): table.function_errors() +def test_function_column_status_is_remote_only(tmp_path): + db = lancedb.connect(tmp_path) + table = db.create_table("t", [{"x": 1}]) + with pytest.raises( + NotImplementedError, + match="exact Function column status is supported only in LanceDB Enterprise", + ): + table.wait_fresh("x") + + @pytest.mark.experimental_functions @pytest.mark.parametrize("batch_size", [None, 8]) def test_refresh_column_async_returns_job(tmp_path, batch_size): diff --git a/python/src/lib.rs b/python/src/lib.rs index 27ff4cb01..f78c4e7bb 100644 --- a/python/src/lib.rs +++ b/python/src/lib.rs @@ -16,9 +16,10 @@ use query::{FTSQuery, HybridQuery, Query, VectorQuery}; use session::Session; use table::{ AddColumnsResult, AddResult, AlterColumnsResult, DeleteResult, DropColumnsResult, FtsToken, - FunctionErrorFragment, FunctionErrorRecord, FunctionErrors, MergeResult, PyBlobFile, - RTEWriteSpec, RefreshColumnResult, RefreshMaterializedViewResult, Table, - UpdateFieldMetadataResult, UpdateResult, + FunctionColumnAutoRefreshStatus, FunctionColumnLastRefreshJob, FunctionColumnStatus, + FunctionColumnVersionStatus, FunctionErrorFragment, FunctionErrorRecord, FunctionErrors, + MergeResult, PyBlobFile, RTEWriteSpec, RefreshColumnResult, RefreshMaterializedViewResult, + Table, UpdateFieldMetadataResult, UpdateResult, }; pub mod arrow; @@ -117,6 +118,10 @@ pub fn _lancedb(py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_class::()?; m.add_class::()?; m.add_class::()?; + m.add_class::()?; + m.add_class::()?; + m.add_class::()?; + m.add_class::()?; m.add_class::()?; m.add_class::()?; m.add_class::()?; diff --git a/python/src/table.rs b/python/src/table.rs index 24cc9ae5c..cf627961c 100644 --- a/python/src/table.rs +++ b/python/src/table.rs @@ -892,6 +892,136 @@ impl From for FunctionErrors { } } +/// Exact immutable Function revision attached to a table output. +#[pyclass(module = "lancedb._lancedb", get_all, from_py_object)] +#[derive(Clone, Debug)] +pub struct FunctionColumnVersionStatus { + pub name: String, + pub namespace_path: Vec, + pub object_id: String, + pub location: String, + pub version: String, + pub manifest_digest: String, +} + +impl From for FunctionColumnVersionStatus { + fn from(function: lancedb::function::FunctionVersionRef) -> Self { + Self { + name: function.name, + namespace_path: function.namespace_path, + object_id: function.object_id, + location: function.location, + version: function.version, + manifest_digest: function.manifest_digest, + } + } +} + +/// Latest matching refresh Job projected for a Function output. +#[pyclass(module = "lancedb._lancedb", get_all, from_py_object)] +#[derive(Clone, Debug)] +pub struct FunctionColumnLastRefreshJob { + pub job_id: String, + pub state: String, + pub automatic: Option, + pub completed_at_millis: Option, +} + +impl From 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, + } + } +} + +/// Automatic-refresh projection for a Function output. +#[pyclass(module = "lancedb._lancedb", get_all, from_py_object)] +#[derive(Clone, Debug)] +pub struct FunctionColumnAutoRefreshStatus { + pub enabled: bool, + pub interval_seconds: Option, + pub active_job_id: Option, + pub state: String, + pub failure_streak: u32, + pub effective_cooldown_millis: Option, + pub blocked_reason: Option, + pub next_run_at_millis: Option, + pub next_run_at_reason: Option, +} + +impl From for FunctionColumnAutoRefreshStatus { + fn from(status: lancedb::table::FunctionColumnAutoRefreshStatus) -> Self { + Self { + enabled: status.enabled, + interval_seconds: status.interval_seconds, + active_job_id: status.active_job_id, + state: status.state.as_str().to_string(), + failure_streak: status.failure_streak, + effective_cooldown_millis: status.effective_cooldown_millis, + blocked_reason: status + .blocked_reason + .map(|reason| reason.as_str().to_string()), + next_run_at_millis: status.next_run_at_millis, + next_run_at_reason: status.next_run_at_reason, + } + } +} + +/// Exact four-way status of one Function output at one table version. +#[pyclass(module = "lancedb._lancedb", get_all, from_py_object)] +#[derive(Clone, Debug)] +pub struct FunctionColumnStatus { + pub column: String, + pub binding_id: String, + pub function: FunctionColumnVersionStatus, + pub table_version: u64, + pub rows: u64, + pub fresh: u64, + pub stale: u64, + pub never_computed: u64, + pub failed: u64, + pub last_refresh_job: Option, + pub auto_refresh: FunctionColumnAutoRefreshStatus, +} + +#[pymethods] +impl FunctionColumnStatus { + pub fn __repr__(&self) -> String { + format!( + "FunctionColumnStatus(column={:?}, table_version={}, rows={}, fresh={}, stale={}, never_computed={}, failed={})", + self.column, + self.table_version, + self.rows, + self.fresh, + self.stale, + self.never_computed, + self.failed + ) + } +} + +impl From for FunctionColumnStatus { + fn from(status: lancedb::table::FunctionColumnStatus) -> Self { + Self { + column: status.column, + binding_id: status.binding_id, + function: status.function.into(), + table_version: status.table_version, + rows: status.rows, + fresh: status.fresh, + stale: status.stale, + never_computed: status.never_computed, + failed: status.failed, + last_refresh_job: status.last_refresh_job.map(Into::into), + auto_refresh: status.auto_refresh.into(), + } + } +} + #[pyclass(get_all, from_py_object)] #[derive(Clone, Debug)] pub struct RefreshMaterializedViewResult { @@ -2425,6 +2555,51 @@ impl Table { }) } + #[pyo3(signature = (column=None))] + pub fn column_statuses( + self_: PyRef<'_, Self>, + column: Option, + ) -> PyResult> { + let inner = self_.inner_ref()?.clone(); + future_into_py(self_.py(), async move { + let statuses = inner + .function_column_statuses(column.as_deref()) + .await + .infer_error()?; + Ok(statuses + .into_iter() + .map(FunctionColumnStatus::from) + .collect::>()) + }) + } + + pub fn column_status(self_: PyRef<'_, Self>, column: String) -> PyResult> { + let inner = self_.inner_ref()?.clone(); + future_into_py(self_.py(), async move { + let status = inner.function_column_status(&column).await.infer_error()?; + Ok(FunctionColumnStatus::from(status)) + }) + } + + #[pyo3(signature = (column, timeout=None))] + pub fn wait_column_fresh<'py>( + self_: PyRef<'py, Self>, + column: String, + timeout: Option>, + ) -> PyResult> { + let timeout = timeout + .map(|timeout| timeout.extract::()) + .transpose()?; + let inner = self_.inner_ref()?.clone(); + future_into_py(self_.py(), async move { + let status = inner + .wait_for_function_column(&column, timeout) + .await + .infer_error()?; + Ok(FunctionColumnStatus::from(status)) + }) + } + #[pyo3(signature = (source_version=None, *, full=false, cpu_millis=None, memory_bytes=None, gpu_count=None, gpu_class=None))] #[allow(clippy::too_many_arguments)] pub fn refresh_materialized_view( diff --git a/rust/lancedb/src/error.rs b/rust/lancedb/src/error.rs index 7499641fc..853ca831f 100644 --- a/rust/lancedb/src/error.rs +++ b/rust/lancedb/src/error.rs @@ -543,6 +543,25 @@ fn has_unsupported_local_filesystem_source(error: &(dyn std::error::Error + 'sta } impl Error { + /// HTTP status associated with a remote request error. + /// + /// Retry wrappers preserve the underlying status. Other error variants do + /// not represent an HTTP response and return `None`. + #[cfg(feature = "remote")] + pub fn status_code(&self) -> Option { + match self { + Self::Http { status_code, .. } => *status_code, + Self::Retry { status_code, .. } => *status_code, + Self::TableNotFound { source, .. } => { + source.downcast_ref::().and_then(Self::status_code) + } + Self::JobFailed { failure, .. } => { + failure.source.as_deref().and_then(Self::status_code) + } + _ => None, + } + } + /// Structured resource details reported by a remote HTTP error. /// /// Older servers do not include these fields, in which case this returns diff --git a/rust/lancedb/src/remote/table.rs b/rust/lancedb/src/remote/table.rs index 247b1925c..222ff527b 100644 --- a/rust/lancedb/src/remote/table.rs +++ b/rust/lancedb/src/remote/table.rs @@ -42,7 +42,8 @@ use crate::table::query::create_multi_vector_plan; use crate::table::rte_stats::GetRTEStatsResponse; use crate::table::write_progress::FinishOnDrop; use crate::table::{ - AlterColumnsResult, FieldMetadataUpdate, RefreshColumnResult, UpdateFieldMetadataResult, + AlterColumnsResult, FieldMetadataUpdate, FunctionColumnStatus, RefreshColumnResult, + UpdateFieldMetadataResult, }; use crate::table::{AnyQuery, Filter, Predicate, PreprocessingOutput, TableStatistics}; use crate::utils::background_cache::BackgroundCache; @@ -4077,6 +4078,54 @@ impl BaseTable for RemoteTable { }) } + async fn function_column_statuses( + &self, + column: Option<&str>, + ) -> Result> { + let mut request = self.apply_branch_query(self.client.get(&format!( + "/v1/table/{}/function_columns/status", + self.identifier + ))); + if let Some(column) = column { + request = request.query(&[("column", column)]); + } + let read_snapshot = self.snapshot_read_state().await; + if let Some(version) = read_snapshot.version { + request = request.query(&[("version", version)]); + } + let request = read_snapshot.freshness.apply(request); + // A typed finalization-pending 503 is a domain result. Let the caller's + // bounded wait policy handle it instead of a generic 5xx retry loop. + let (request_id, response) = self.client.send_with_retry(request, None, false).await?; + if response.status().is_success() { + read_snapshot + .freshness + .observe_headers(&self.freshness, response.headers()); + } + let response = self.check_table_response(&request_id, response).await?; + let body = response.text().await.err_to_http(request_id.clone())?; + let statuses: Vec = + serde_json::from_str(&body).map_err(|error| Error::Http { + source: format!("failed to parse Function column status response: {error}").into(), + request_id, + status_code: None, + })?; + crate::table::function_column_status::validate_statuses(&statuses)?; + if let Some(requested_version) = read_snapshot.version + && let Some(returned_version) = statuses + .iter() + .map(|status| status.table_version) + .find(|version| *version != requested_version) + { + return Err(Error::Runtime { + message: format!( + "Function column status response returned table version {returned_version}, but the table handle is pinned to version {requested_version}" + ), + }); + } + Ok(statuses) + } + async fn alter_columns(&self, alterations: &[ColumnAlteration]) -> Result { self.check_mutable().await?; let body = alterations @@ -11350,6 +11399,450 @@ mod tests { assert_eq!(schema_requests.load(Ordering::SeqCst), 2); } + #[tokio::test] + async fn test_function_column_status_decodes_one_exact_snapshot() { + let table = Table::new_with_handler("my_table", |request| { + assert_eq!(request.method(), "GET"); + assert_eq!( + request.url().path(), + "/v1/table/my_table/function_columns/status" + ); + assert_eq!(request.url().query(), None); + http::Response::builder() + .status(200) + .body( + r#"[{ + "column":"embedding", + "binding_id":"fb_01", + "function":{ + "name":"embed", + "object_id":"embed", + "location":"objects/embed", + "version":"7", + "manifest_digest":"sha256:abc" + }, + "table_version":42, + "rows":10, + "fresh":6, + "stale":2, + "never_computed":1, + "failed":1, + "last_refresh_job":{ + "job_id":"job-7", + "state":"finished", + "automatic":true, + "completed_at_millis":1790000000000 + }, + "auto_refresh":{ + "enabled":true, + "interval_seconds":60, + "active_job_id":null, + "state":"unobservable", + "failure_streak":0, + "next_run_at_millis":null, + "next_run_at_reason":"authoritative_scheduler_projection_unavailable" + } + }]"#, + ) + .unwrap() + }); + + let statuses = table.function_column_statuses(None).await.unwrap(); + assert_eq!(statuses.len(), 1); + assert_eq!(statuses[0].table_version, 42); + assert_eq!(statuses[0].failed, 1); + assert_eq!( + statuses[0] + .last_refresh_job + .as_ref() + .map(|job| job.job_id.as_str()), + Some("job-7") + ); + assert_eq!( + statuses[0].auto_refresh.state, + crate::table::FunctionColumnAutoRefreshState::Unobservable + ); + } + + #[tokio::test] + async fn test_function_column_status_filters_with_a_get_query() { + let table = Table::new_with_handler("my_table", |request| { + assert_eq!(request.method(), "GET"); + assert_eq!(request.url().query(), Some("column=embedding")); + http::Response::builder() + .status(200) + .body( + r#"[{ + "column":"embedding", + "binding_id":"fb_01", + "function":{ + "name":"embed", + "object_id":"embed", + "location":"objects/embed", + "version":"7", + "manifest_digest":"sha256:abc" + }, + "table_version":42, + "rows":0, + "fresh":0, + "stale":0, + "never_computed":0, + "failed":0, + "auto_refresh":{ + "enabled":false, + "state":"disabled", + "failure_streak":0, + "next_run_at_millis":null + } + }]"#, + ) + .unwrap() + }); + + let status = table.function_column_status("embedding").await.unwrap(); + assert_eq!(status.column, "embedding"); + assert!(status.last_refresh_job.is_none()); + assert_eq!( + status.auto_refresh.state, + crate::table::FunctionColumnAutoRefreshState::Disabled + ); + } + + #[tokio::test] + async fn test_wait_for_function_column_retries_only_typed_finalization_pending() { + let calls = Arc::new(AtomicUsize::new(0)); + let table = Table::new_with_handler("my_table", { + let calls = calls.clone(); + move |_request| { + if calls.fetch_add(1, Ordering::SeqCst) == 0 { + http::Response::builder() + .status(StatusCode::SERVICE_UNAVAILABLE) + .header( + "x-lancedb-error-code", + crate::table::function_column_status::FINALIZATION_PENDING_ERROR_CODE, + ) + .body(b"finalization pending".to_vec()) + .unwrap() + } else { + http::Response::builder() + .status(200) + .body( + br#"[{ + "column":"embedding", + "binding_id":"fb_01", + "function":{ + "name":"embed", + "object_id":"embed", + "location":"objects/embed", + "version":"7", + "manifest_digest":"sha256:abc" + }, + "table_version":42, + "rows":0, + "fresh":0, + "stale":0, + "never_computed":0, + "failed":0, + "auto_refresh":{ + "enabled":false, + "state":"disabled", + "failure_streak":0, + "next_run_at_millis":null + } + }]"# + .to_vec(), + ) + .unwrap() + } + } + }); + + let status = table + .wait_for_function_column("embedding", Some(Duration::from_secs(2))) + .await + .unwrap(); + assert_eq!(status.table_version, 42); + assert_eq!(calls.load(Ordering::SeqCst), 2); + } + + #[tokio::test] + async fn test_finalization_code_on_other_status_is_not_retried() { + let calls = Arc::new(AtomicUsize::new(0)); + let table = Table::new_with_handler("my_table", { + let calls = calls.clone(); + move |_request| { + calls.fetch_add(1, Ordering::SeqCst); + http::Response::builder() + .status(StatusCode::INTERNAL_SERVER_ERROR) + .header( + "x-lancedb-error-code", + crate::table::function_column_status::FINALIZATION_PENDING_ERROR_CODE, + ) + .body(b"misclassified pending".to_vec()) + .unwrap() + } + }); + + let error = table + .wait_for_function_column("embedding", Some(Duration::from_secs(2))) + .await + .unwrap_err(); + assert!(error.to_string().contains("misclassified pending")); + assert_eq!(calls.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn test_wait_for_function_column_does_not_retry_generic_500() { + let calls = Arc::new(AtomicUsize::new(0)); + let table = Table::new_with_handler("my_table", { + let calls = calls.clone(); + move |_request| { + calls.fetch_add(1, Ordering::SeqCst); + http::Response::builder() + .status(StatusCode::INTERNAL_SERVER_ERROR) + .body(b"corrupt freshness evidence".to_vec()) + .unwrap() + } + }); + + let error = table + .wait_for_function_column("embedding", Some(Duration::from_secs(2))) + .await + .unwrap_err(); + assert!(error.to_string().contains("corrupt freshness evidence")); + assert_eq!(calls.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn test_function_column_status_decodes_agent_up_to_date() { + let table = Table::new_with_handler("my_table", |_request| { + http::Response::builder() + .status(200) + .body( + r#"[{ + "column":"embedding", + "binding_id":"fb_01", + "function":{ + "name":"embed", + "object_id":"embed", + "location":"objects/embed", + "version":"7", + "manifest_digest":"sha256:abc" + }, + "table_version":42, + "rows":4, + "fresh":4, + "stale":0, + "never_computed":0, + "failed":0, + "auto_refresh":{ + "enabled":true, + "interval_seconds":60, + "state":"up_to_date", + "failure_streak":0, + "next_run_at_millis":null + } + }]"#, + ) + .unwrap() + }); + + let status = table.function_column_status("embedding").await.unwrap(); + assert_eq!( + status.auto_refresh.state, + crate::table::FunctionColumnAutoRefreshState::UpToDate + ); + } + + #[tokio::test] + async fn test_function_column_status_decodes_blocked_scheduler_fields() { + let table = Table::new_with_handler("my_table", |_request| { + http::Response::builder() + .status(200) + .body( + r#"[{ + "column":"embedding", + "binding_id":"fb_01", + "function":{ + "name":"embed", + "object_id":"embed", + "location":"objects/embed", + "version":"7", + "manifest_digest":"sha256:abc" + }, + "table_version":42, + "rows":4, + "fresh":2, + "stale":2, + "never_computed":0, + "failed":0, + "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 + } + }]"#, + ) + .unwrap() + }); + + let status = table.function_column_status("embedding").await.unwrap(); + assert_eq!( + status.auto_refresh.state, + crate::table::FunctionColumnAutoRefreshState::Blocked + ); + 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, + Some(crate::table::FunctionColumnAutoRefreshBlockedReason::PreparationBackoff) + ); + } + + #[tokio::test] + async fn test_function_column_status_passes_a_pinned_version() { + let calls = Arc::new(AtomicUsize::new(0)); + let status_calls = Arc::new(AtomicUsize::new(0)); + let table = Table::new_with_handler("my_table", { + let calls = calls.clone(); + let status_calls = status_calls.clone(); + move |request| { + calls.fetch_add(1, Ordering::SeqCst); + match request.url().path() { + "/v1/table/my_table/describe/" => { + assert_eq!(request_body_json(&request)["version"], 42); + http::Response::builder() + .status(200) + .body(r#"{"version":42,"schema":{"fields":[]}}"#.to_string()) + .unwrap() + } + "/v1/table/my_table/function_columns/status" => { + assert_eq!(request.method(), "GET"); + let query: std::collections::BTreeMap<_, _> = + request.url().query_pairs().collect(); + assert_eq!( + query.get("column").map(|value| value.as_ref()), + Some("embedding") + ); + assert_eq!(query.get("version").map(|value| value.as_ref()), Some("42")); + let returned_version = if status_calls.fetch_add(1, Ordering::SeqCst) == 0 { + 42 + } else { + 43 + }; + http::Response::builder() + .status(200) + .body(format!( + r#"[{{ + "column":"embedding", + "binding_id":"fb_01", + "function":{{ + "name":"embed", + "object_id":"embed", + "location":"objects/embed", + "version":"7", + "manifest_digest":"sha256:abc" + }}, + "table_version":{returned_version}, + "rows":0, + "fresh":0, + "stale":0, + "never_computed":0, + "failed":0, + "auto_refresh":{{ + "enabled":false, + "state":"disabled", + "failure_streak":0, + "next_run_at_millis":null + }} + }}]"#, + )) + .unwrap() + } + path => panic!("unexpected request path: {path}"), + } + } + }); + table.checkout(42).await.unwrap(); + + let status = table.function_column_status("embedding").await.unwrap(); + assert_eq!(status.table_version, 42); + assert_eq!(calls.load(Ordering::SeqCst), 2); + + let error = table.function_column_status("embedding").await.unwrap_err(); + assert!( + error.to_string().contains( + "returned table version 43, but the table handle is pinned to version 42" + ) + ); + assert_eq!(status_calls.load(Ordering::SeqCst), 2); + assert_eq!(calls.load(Ordering::SeqCst), 3); + } + + #[tokio::test] + async fn test_function_column_status_keeps_latest_freshness_headers() { + let table = RemoteTable::new_mock( + "my_table".to_string(), + |request| { + assert_eq!(request.method(), "GET"); + assert_eq!( + request.url().query_pairs().collect::>(), + vec![( + std::borrow::Cow::Borrowed("column"), + std::borrow::Cow::Borrowed("embedding") + )] + ); + assert_eq!( + request + .headers() + .get("x-lancedb-min-version") + .and_then(|value| value.to_str().ok()), + Some("7") + ); + http::Response::builder() + .status(200) + .body( + r#"[{ + "column":"embedding", + "binding_id":"fb_01", + "function":{ + "name":"embed", + "object_id":"embed", + "location":"objects/embed", + "version":"7", + "manifest_digest":"sha256:abc" + }, + "table_version":42, + "rows":0, + "fresh":0, + "stale":0, + "never_computed":0, + "failed":0, + "auto_refresh":{ + "enabled":false, + "state":"disabled", + "failure_streak":0, + "next_run_at_millis":null + } + }]"#, + ) + .unwrap() + }, + None, + ); + table.freshness.lock().unwrap().min_version = Some(7); + + let statuses = table + .function_column_statuses(Some("embedding")) + .await + .unwrap(); + assert_eq!(statuses.len(), 1); + } + /// The error listing is table-addressed with optional job and column /// filters, mirroring the server's SQL surface, and the two non-record /// signals come back as their own fields rather than as rows. diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs index bd9220754..18a5b7ddf 100644 --- a/rust/lancedb/src/table.rs +++ b/rust/lancedb/src/table.rs @@ -83,6 +83,7 @@ pub mod datafusion; pub(crate) mod dataset; pub mod delete; pub mod freshness; +pub mod function_column_status; pub mod merge; pub mod optimize; pub(crate) mod primary_key; @@ -106,6 +107,11 @@ pub use computed_columns::{ ComputedColumn, ComputedColumnKind, computed_column_from_field, computed_columns, }; pub use delete::DeleteResult; +pub use function_column_status::{ + FunctionColumnAutoRefreshBlockedReason, FunctionColumnAutoRefreshState, + FunctionColumnAutoRefreshStatus, FunctionColumnLastRefreshJob, FunctionColumnStatus, + FunctionColumnVersionStatus, +}; use futures::future::join_all; pub use lance::dataset::refs::{BranchContents, Ref, TagContents, Tags as LanceTags}; pub use lance::dataset::scanner::DatasetRecordBatchStream; @@ -961,6 +967,19 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync { .append_to("per-row Function errors are recorded only in LanceDB Enterprise"), }) } + /// Exact four-way freshness status for Function-backed output columns. + /// + /// `column` filters to one user-visible output; `None` returns every + /// Function output. The default returns `NotSupported`. + async fn function_column_statuses( + &self, + _column: Option<&str>, + ) -> Result> { + Err(Error::NotSupported { + message: DocLink::Functions + .append_to("exact Function column status is supported only in LanceDB Enterprise"), + }) + } /// Alter columns in the table. async fn alter_columns(&self, alterations: &[ColumnAlteration]) -> Result; /// Drop columns from the table. @@ -2246,6 +2265,72 @@ impl Table { .await } + /// Return exact freshness status for all Function-backed output columns. + /// + /// The server evaluates all counts at one table version. Clients never + /// derive counts by joining schema, Job, and error listings. A handle + /// pinned with `checkout` sends that exact version, while a latest handle + /// keeps the existing read-freshness headers. + pub async fn function_column_statuses( + &self, + column: Option<&str>, + ) -> Result> { + let statuses = self.inner.function_column_statuses(column).await?; + function_column_status::validate_statuses(&statuses)?; + Ok(statuses) + } + + /// Return exact freshness status for one Function-backed output column. + pub async fn function_column_status(&self, column: &str) -> Result { + let mut statuses = self.function_column_statuses(Some(column)).await?; + if statuses.len() != 1 || statuses[0].column != column { + return Err(Error::ColumnNotFound { + name: column.to_string(), + }); + } + Ok(statuses.remove(0)) + } + + /// Poll until a Function output has no stale or never-computed rows. + /// + /// Failed rows do not keep the call pending. A typed + /// `function_column_status_finalization_pending` response is retried with + /// bounded exponential backoff; every other error is returned immediately. + pub async fn wait_for_function_column( + &self, + column: &str, + timeout: Option, + ) -> Result { + let started = std::time::Instant::now(); + let mut delay = std::time::Duration::from_millis(250); + loop { + let result = self.function_column_status(column).await; + match result { + Ok(status) if status.stale == 0 && status.never_computed == 0 => return Ok(status), + Ok(_) => {} + Err(error) if function_column_status::is_finalization_pending(&error) => {} + Err(error) => return Err(error), + } + + let sleep_for = match timeout { + Some(timeout) => { + let remaining = timeout.saturating_sub(started.elapsed()); + if remaining.is_zero() { + return Err(Error::Timeout { + message: format!( + "Function column '{column}' did not become fresh before timeout" + ), + }); + } + delay.min(remaining) + } + None => delay, + }; + tokio::time::sleep(sleep_for).await; + delay = std::cmp::min(delay * 2, std::time::Duration::from_secs(1)); + } + } + /// The per-row errors Function refreshes recorded on this table: the /// rows a refresh skipped under its skip policy, with the failing input /// and the error, plus a summary for any fragment whose detail was diff --git a/rust/lancedb/src/table/freshness.rs b/rust/lancedb/src/table/freshness.rs index 50434143e..477020fc3 100644 --- a/rust/lancedb/src/table/freshness.rs +++ b/rust/lancedb/src/table/freshness.rs @@ -21,6 +21,7 @@ //! removed by [`prune_sidecars`]. use std::collections::{BTreeMap, HashMap, HashSet}; +use std::fmt; use std::io::{Cursor, Read}; use std::ops::Range; @@ -98,6 +99,56 @@ impl DirtyRows { /// `{fragment id -> row-level dirty state}`. A missing fragment is clean. pub type RowStateMap = BTreeMap; +/// Physical rows whose latest Function outcome is an unresolved failure. +pub type FailureRows = BTreeMap; + +/// Evidence for rows whose latest Function outcome is known. +/// +/// `known_rows` includes both successful and failed observations. +/// `failed_rows` is the failure subset. A missing row is unknown, not clean. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct FailureEvidence { + pub known_rows: FailureRows, + pub failed_rows: FailureRows, +} + +impl FailureEvidence { + fn is_empty(&self) -> bool { + self.known_rows.is_empty() && self.failed_rows.is_empty() + } + + fn validate(&self) -> Result<()> { + if self.failed_rows.iter().any(|(fragment_id, failed)| { + !self + .known_rows + .get(fragment_id) + .is_some_and(|known| failed.is_subset(known)) + }) { + return Err(invalid( + "failed rows are not covered by the failure evidence update".to_string(), + )); + } + Ok(()) + } + + fn for_fragments(&self, fragments: &HashSet) -> Self { + Self { + known_rows: self + .known_rows + .iter() + .filter(|(fragment_id, _)| fragments.contains(fragment_id)) + .map(|(fragment_id, rows)| (*fragment_id, rows.clone())) + .collect(), + failed_rows: self + .failed_rows + .iter() + .filter(|(fragment_id, _)| fragments.contains(fragment_id)) + .map(|(fragment_id, rows)| (*fragment_id, rows.clone())) + .collect(), + } + } +} + /// The single owning value of the freshness sidecar. /// /// `SignatureMap` remains the public facade for callers that only need input @@ -107,6 +158,13 @@ pub type RowStateMap = BTreeMap; struct FreshnessState { signatures: SignatureMap, row_state: RowStateMap, + /// Physical offsets with an exact current-generation outcome. + failure_coverage: FailureRows, + /// The failed subset of `failure_coverage`. + failed_rows: FailureRows, + /// Whether the decoded source used the v3 sidecar format that carries + /// failure evidence. Older formats leave failure rows unclassified. + failure_evidence_recorded: bool, basis: String, // A fresh identity for each begin, even when the basis and dirty rows repeat. intent_id: Uuid, @@ -115,6 +173,8 @@ struct FreshnessState { impl FreshnessState { fn normalize(mut self) -> Result { self.row_state.retain(|_, rows| !rows.is_explicitly_empty()); + self.failure_coverage.retain(|_, rows| !rows.is_empty()); + self.failed_rows.retain(|_, rows| !rows.is_empty()); if self .row_state .keys() @@ -124,6 +184,26 @@ impl FreshnessState { "row state names a fragment without a signature".to_string(), )); } + if self + .failure_coverage + .keys() + .chain(self.failed_rows.keys()) + .any(|fragment_id| !self.signatures.contains_key(fragment_id)) + { + return Err(invalid( + "failure evidence names a fragment without a signature".to_string(), + )); + } + if self.failed_rows.iter().any(|(fragment_id, failed)| { + !self + .failure_coverage + .get(fragment_id) + .is_some_and(|covered| failed.is_subset(covered)) + }) { + return Err(invalid( + "failed rows are not covered by failure evidence".to_string(), + )); + } if self .signatures .values() @@ -138,6 +218,38 @@ impl FreshnessState { } Ok(self) } + + fn invalidate_failure_evidence(&mut self, fragments: impl IntoIterator) { + for fragment_id in fragments { + self.failure_coverage.remove(&fragment_id); + self.failed_rows.remove(&fragment_id); + } + } + + fn invalidate_failure_rows(&mut self, rows: &BTreeMap) { + for (fragment_id, rows) in rows { + if let Some(covered) = self.failure_coverage.get_mut(fragment_id) { + *covered -= rows; + } + if self + .failure_coverage + .get(fragment_id) + .is_some_and(RoaringBitmap::is_empty) + { + self.failure_coverage.remove(fragment_id); + } + if let Some(failed) = self.failed_rows.get_mut(fragment_id) { + *failed -= rows; + } + if self + .failed_rows + .get(fragment_id) + .is_some_and(RoaringBitmap::is_empty) + { + self.failed_rows.remove(fragment_id); + } + } + } } /// Opaque accepted-basis evidence returned by [`begin_row_repair`] and required by @@ -398,7 +510,8 @@ const SIDECAR_DIR: &str = "_computed"; const SIDECAR_REF: &str = "sidecar:"; const SIDECAR_MAGIC: &[u8; 4] = b"CSIG"; const SIDECAR_FORMAT_V1: u8 = 1; -const SIDECAR_FORMAT: u8 = 2; +const SIDECAR_FORMAT_V2: u8 = 2; +const SIDECAR_FORMAT: u8 = 3; /// How long an unreferenced sidecar is presumed to be a stamp in flight: /// lance's own threshold for unverified files, independent of how much /// version history a cleanup keeps. @@ -446,9 +559,85 @@ fn encode_legacy_sidecar(map: &SignatureMap) -> Result> { Ok(bytes) } +#[cfg(test)] +fn encode_v2_sidecar(state: &FreshnessState) -> Result> { + let state = state.clone().normalize()?; + let basis_len = u16::try_from(state.basis.len()) + .map_err(|_| invalid("freshness basis exceeds u16 bytes".to_string()))?; + let fragment_count = u32::try_from(state.signatures.len()) + .map_err(|_| invalid("too many fragments for a freshness sidecar".to_string()))?; + let mut bytes = Vec::new(); + bytes.extend_from_slice(SIDECAR_MAGIC); + bytes.push(SIDECAR_FORMAT_V2); + bytes.extend_from_slice(state.intent_id.as_bytes()); + bytes.extend_from_slice(&basis_len.to_le_bytes()); + bytes.extend_from_slice(state.basis.as_bytes()); + bytes.extend_from_slice(&fragment_count.to_le_bytes()); + for (fragment_id, signature) in &state.signatures { + let hash = u64::from_str_radix(signature, 16).map_err(|_| { + invalid(format!( + "signature '{signature}' is not a 64-bit hex digest" + )) + })?; + let rows = state + .row_state + .get(fragment_id) + .cloned() + .unwrap_or_else(|| DirtyRows::Rows(RoaringBitmap::new())); + let (kind, bitmap) = match rows { + DirtyRows::Rows(bitmap) => (0u8, bitmap), + DirtyRows::AllExcept(clean) => (1u8, clean), + }; + bytes.extend_from_slice(&fragment_id.to_le_bytes()); + bytes.extend_from_slice(&hash.to_le_bytes()); + bytes.push(kind); + encode_bitmap(&mut bytes, Some(&bitmap))?; + } + Ok(bytes) +} + +fn encode_bitmap(bytes: &mut Vec, bitmap: Option<&RoaringBitmap>) -> Result<()> { + let Some(bitmap) = bitmap.filter(|bitmap| !bitmap.is_empty()) else { + bytes.extend_from_slice(&0u32.to_le_bytes()); + return Ok(()); + }; + let mut encoded = Vec::with_capacity(bitmap.serialized_size()); + bitmap + .serialize_into(&mut encoded) + .map_err(|error| invalid(format!("encoding freshness bitmap: {error}")))?; + let bitmap_len = u32::try_from(encoded.len()) + .map_err(|_| invalid("freshness bitmap exceeds u32 bytes".to_string()))?; + bytes.extend_from_slice(&bitmap_len.to_le_bytes()); + bytes.extend_from_slice(&encoded); + Ok(()) +} + +fn decode_bitmap( + cursor: &mut Cursor<&[u8]>, + malformed: &impl Fn() -> Error, +) -> Result> { + let mut bitmap_len = [0u8; 4]; + cursor + .read_exact(&mut bitmap_len) + .map_err(|_| malformed())?; + let bitmap_len = u32::from_le_bytes(bitmap_len) as usize; + if bitmap_len == 0 { + return Ok(None); + } + let mut encoded = vec![0u8; bitmap_len]; + cursor.read_exact(&mut encoded).map_err(|_| malformed())?; + let mut bitmap_reader = Cursor::new(encoded.as_slice()); + let bitmap = RoaringBitmap::deserialize_from(&mut bitmap_reader).map_err(|_| malformed())?; + if bitmap_reader.position() != bitmap_len as u64 { + return Err(malformed()); + } + Ok(Some(bitmap)) +} + /// `CSIG`, format byte, 16-byte intent UUID, basis, entry count, then for each fragment its id, -/// 8-byte signature, dirty-row mode and serialized bitmap. All integers are -/// little-endian. Format 1 remains readable as clean row state. +/// 8-byte signature, dirty-row mode and serialized bitmap. Format 3 appends +/// failure-coverage and failed-row bitmaps. All integers are little-endian. +/// Format 1 remains readable as clean row state. fn encode_sidecar(state: &FreshnessState) -> Result> { let state = state.clone().normalize()?; let basis_len = u16::try_from(state.basis.len()) @@ -477,17 +666,12 @@ fn encode_sidecar(state: &FreshnessState) -> Result> { DirtyRows::Rows(bitmap) => (0u8, bitmap), DirtyRows::AllExcept(clean) => (1u8, clean), }; - let mut encoded = Vec::with_capacity(bitmap.serialized_size()); - bitmap - .serialize_into(&mut encoded) - .map_err(|error| invalid(format!("encoding freshness bitmap: {error}")))?; - let bitmap_len = u32::try_from(encoded.len()) - .map_err(|_| invalid("freshness bitmap exceeds u32 bytes".to_string()))?; bytes.extend_from_slice(&fragment_id.to_le_bytes()); bytes.extend_from_slice(&hash.to_le_bytes()); bytes.push(kind); - bytes.extend_from_slice(&bitmap_len.to_le_bytes()); - bytes.extend_from_slice(&encoded); + encode_bitmap(&mut bytes, Some(&bitmap))?; + encode_bitmap(&mut bytes, state.failure_coverage.get(fragment_id))?; + encode_bitmap(&mut bytes, state.failed_rows.get(fragment_id))?; } Ok(bytes) } @@ -526,7 +710,7 @@ fn decode_sidecar(bytes: &[u8]) -> Result { cursor.read_exact(&mut magic).map_err(|_| malformed())?; let mut format = [0u8; 1]; cursor.read_exact(&mut format).map_err(|_| malformed())?; - if &magic != SIDECAR_MAGIC || format[0] != SIDECAR_FORMAT { + if &magic != SIDECAR_MAGIC || !matches!(format[0], SIDECAR_FORMAT_V2 | SIDECAR_FORMAT) { return Err(malformed()); } let mut intent_id = [0u8; 16]; @@ -542,6 +726,7 @@ fn decode_sidecar(bytes: &[u8]) -> Result { let mut state = FreshnessState { basis, intent_id: Uuid::from_bytes(intent_id), + failure_evidence_recorded: format[0] == SIDECAR_FORMAT, ..Default::default() }; for _ in 0..count { @@ -554,19 +739,7 @@ fn decode_sidecar(bytes: &[u8]) -> Result { cursor.read_exact(&mut signature).map_err(|_| malformed())?; let mut kind = [0u8; 1]; cursor.read_exact(&mut kind).map_err(|_| malformed())?; - let mut bitmap_len = [0u8; 4]; - cursor - .read_exact(&mut bitmap_len) - .map_err(|_| malformed())?; - let bitmap_len = u32::from_le_bytes(bitmap_len) as usize; - let mut encoded = vec![0u8; bitmap_len]; - cursor.read_exact(&mut encoded).map_err(|_| malformed())?; - let mut bitmap_reader = Cursor::new(encoded.as_slice()); - let bitmap = - RoaringBitmap::deserialize_from(&mut bitmap_reader).map_err(|_| malformed())?; - if bitmap_reader.position() != bitmap_len as u64 { - return Err(malformed()); - } + let bitmap = decode_bitmap(&mut cursor, &malformed)?.unwrap_or_default(); let dirty = match kind[0] { 0 => DirtyRows::Rows(bitmap), 1 => DirtyRows::AllExcept(bitmap), @@ -582,6 +755,14 @@ fn decode_sidecar(bytes: &[u8]) -> Result { if !dirty.is_explicitly_empty() { state.row_state.insert(fragment_id, dirty); } + if format[0] == SIDECAR_FORMAT { + if let Some(covered) = decode_bitmap(&mut cursor, &malformed)? { + state.failure_coverage.insert(fragment_id, covered); + } + if let Some(failed) = decode_bitmap(&mut cursor, &malformed)? { + state.failed_rows.insert(fragment_id, failed); + } + } } if cursor.position() != bytes.len() as u64 { return Err(malformed()); @@ -763,12 +944,15 @@ enum StoredFreshnessState { Unreadable, } -async fn stored_freshness_state(dataset: &Dataset, column: &str) -> StoredFreshnessState { +async fn load_freshness_state( + dataset: &Dataset, + column: &str, +) -> Result)>> { let Some(encoded) = field_meta(dataset, column, SOURCE_SIGNATURE_META_KEY) else { - return StoredFreshnessState::Absent; + return Ok(None); }; // Inline JSON is the declaration's empty seed and the pre-sidecar form. - let (read, digest) = match encoded.strip_prefix(SIDECAR_REF) { + let read = match encoded.strip_prefix(SIDECAR_REF) { Some(digest) => ( read_sidecar(dataset, digest).await, Some(digest.to_string()), @@ -783,8 +967,14 @@ async fn stored_freshness_state(dataset: &Dataset, column: &str) -> StoredFreshn None, ), }; - match read { - Ok(state) => StoredFreshnessState::Present { state, digest }, + let (state, digest) = read; + Ok(Some((state?, digest))) +} + +async fn stored_freshness_state(dataset: &Dataset, column: &str) -> StoredFreshnessState { + match load_freshness_state(dataset, column).await { + Ok(Some((state, digest))) => StoredFreshnessState::Present { state, digest }, + Ok(None) => StoredFreshnessState::Absent, Err(error) => { log::warn!( "computed column '{column}' freshness state is unreadable ({error}); every fragment will be recomputed" @@ -853,6 +1043,38 @@ impl StalenessPlan { } } +/// Strict, internally consistent evidence for a Function-column status read. +/// +/// Unlike [`staleness_against`], this value never converts an unreadable +/// sidecar into a conservative recompute plan. Failure coverage may be +/// partial: the caller combines it with live-row freshness and value evidence. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum StrictFreshness { + /// No freshness certificate is recorded on this dataset version. + /// + /// This does not prove that no refresh was ever attempted; older failed + /// attempts may be untracked. Assignment and value evidence still decide + /// stale versus never computed. + Absent, + /// The sidecar predates failure evidence. Its staleness plan is still + /// usable; row classification may use freshness, values, and assignment + /// evidence outside this sidecar. + Uncertified { staleness: StalenessPlan }, + /// Exact staleness and current-generation failed rows. + Certified { + staleness: StalenessPlan, + failures: FailureEvidence, + }, +} + +fn incomplete_certificate(column: &str, reason: impl fmt::Display) -> Error { + Error::Runtime { + message: format!( + "computed column '{column}' has no complete freshness certificate: {reason}" + ), + } +} + /// A version of the log since the stamp that bears on carrying freshness /// forward: an append, whose fragments hold no computed value yet, or a /// compaction's rewrite groups, consumed and produced fragment ids. @@ -1158,6 +1380,16 @@ pub async fn staleness_against( } StoredFreshnessState::Present { state: stored, .. } => stored, }; + staleness_from_state(dataset, column, definition_version, inputs, stored).await +} + +async fn staleness_from_state( + dataset: &Dataset, + column: &str, + definition_version: &str, + inputs: &InputFields, + stored: FreshnessState, +) -> Result { let stored_version = field_meta(dataset, column, DEFINITION_VERSION_META_KEY); if stored_version.is_some_and(|version| version != definition_version) { return Ok(StalenessPlan { @@ -1221,6 +1453,71 @@ pub async fn staleness_against( }) } +/// Read exact status evidence without the planner's conservative fallback. +/// +/// A missing metadata key returns [`StrictFreshness::Absent`]. A referenced +/// sidecar that cannot be read or decoded propagates the existing error. The +/// returned plan is certified when the sidecar records a definition and can be +/// decoded. Failure coverage is returned only for live fragments whose stored +/// input signature still matches; stale evidence is omitted. The status +/// provider remains responsible for classifying uncovered live rows as fresh, +/// stale, or never computed, and for reporting an error only when a row is +/// genuinely ambiguous. +pub async fn strict_freshness_against( + dataset: &Dataset, + column: &str, + definition_version: &str, + inputs: &InputFields, +) -> Result { + let Some((stored, _)) = load_freshness_state(dataset, column).await? else { + return Ok(StrictFreshness::Absent); + }; + let Some(stored_definition) = field_meta(dataset, column, DEFINITION_VERSION_META_KEY) else { + return Err(incomplete_certificate( + column, + "the recorded definition is missing", + )); + }; + let staleness = + staleness_from_state(dataset, column, definition_version, inputs, stored.clone()).await?; + if !stored.failure_evidence_recorded { + return Ok(StrictFreshness::Uncertified { staleness }); + } + if stored_definition != definition_version { + return Ok(StrictFreshness::Certified { + staleness, + failures: FailureEvidence::default(), + }); + } + let mut failures = FailureEvidence::default(); + for fragment in dataset.get_fragments() { + let fragment_id = fragment.id() as u32; + let current = signature_of(dataset, fragment_id, inputs)?.ok_or_else(|| { + incomplete_certificate(column, format!("fragment {fragment_id} is not live")) + })?; + if stored.signatures.get(&fragment_id) != Some(¤t) { + // The old generation's failure evidence no longer describes this + // fragment. Its live rows are stale or never computed, not + // ambiguously failed. + continue; + } + if let Some(covered) = stored.failure_coverage.get(&fragment_id) + && !covered.is_empty() + { + failures.known_rows.insert(fragment_id, covered.clone()); + } + if let Some(failed) = stored.failed_rows.get(&fragment_id) + && !failed.is_empty() + { + failures.failed_rows.insert(fragment_id, failed.clone()); + } + } + Ok(StrictFreshness::Certified { + staleness, + failures, + }) +} + /// What [`record_freshness`] wrote: the table version the stamp landed at, /// if it wrote one, and the entries recorded and dropped for having moved. #[derive(Debug, Default, PartialEq, Eq)] @@ -1251,6 +1548,55 @@ pub async fn record_freshness( definition_version: &str, inputs: &InputFields, computed: SignatureMap, +) -> Result { + record_freshness_impl( + latest, + pinned, + column, + definition_version, + inputs, + computed, + None, + ) + .await +} + +/// Record signatures and an explicit current-generation failure outcome in one +/// immutable sidecar transition. +/// +/// `failures.known_rows` must include only physical offsets whose final +/// outcome is known. `failures.failed_rows` is the failed subset. Rows omitted +/// from coverage remain unknown to failure classification; freshness, values, +/// and assignment evidence may still classify them. +pub async fn record_freshness_with_failures( + latest: &mut Dataset, + pinned: Option<(&Dataset, &StalenessPlan)>, + column: &str, + definition_version: &str, + inputs: &InputFields, + computed: SignatureMap, + failures: &FailureEvidence, +) -> Result { + record_freshness_impl( + latest, + pinned, + column, + definition_version, + inputs, + computed, + Some(failures), + ) + .await +} + +async fn record_freshness_impl( + latest: &mut Dataset, + pinned: Option<(&Dataset, &StalenessPlan)>, + column: &str, + definition_version: &str, + inputs: &InputFields, + computed: SignatureMap, + failures: Option<&FailureEvidence>, ) -> Result { let absent = matches!( stored_signatures(latest, column).await, @@ -1270,7 +1616,7 @@ pub async fn record_freshness( } } entries.extend(computed); - if entries.is_empty() && !absent { + if entries.is_empty() && !absent && failures.is_none_or(FailureEvidence::is_empty) { return Ok(FreshnessRecord::default()); } let fragments: Vec = entries.keys().copied().collect(); @@ -1280,7 +1626,19 @@ pub async fn record_freshness( .filter(|(fragment_id, signature)| current.get(fragment_id) == Some(signature)) .collect(); let recorded = verified.len(); - let version = write_signatures(latest, column, definition_version, verified).await?; + let verified_fragments: HashSet = verified.keys().copied().collect(); + let scoped_failures = failures + .map(|failures| failures.for_fragments(&verified_fragments)) + .filter(|failures| !failures.is_empty()); + let version = write_signatures_impl( + latest, + column, + definition_version, + Some(inputs), + verified, + scoped_failures.as_ref(), + ) + .await?; Ok(FreshnessRecord { version: Some(version), recorded, @@ -1582,6 +1940,25 @@ pub async fn write_signatures( definition_version: &str, entries: SignatureMap, ) -> Result { + write_signatures_impl(dataset, column, definition_version, None, entries, None).await +} + +async fn write_signatures_impl( + dataset: &mut Dataset, + column: &str, + definition_version: &str, + inputs: Option<&InputFields>, + entries: SignatureMap, + failures: Option<&FailureEvidence>, +) -> Result { + if let Some(failures) = failures { + failures.validate()?; + if inputs.is_none() { + return Err(invalid( + "failure evidence requires input fields for signature validation".to_string(), + )); + } + } let definition_changed = field_meta(dataset, column, DEFINITION_VERSION_META_KEY) .is_some_and(|version| version != definition_version); let live: HashSet = dataset @@ -1619,17 +1996,40 @@ pub async fn write_signatures( } // A signature entry written through this API means the fragment was // recomputed in full. Whole callers keep their old semantics while - // clearing any partial retry state left by a previous publication. + // clearing any partial retry state left by a previous publication. The + // old failure outcome is also unknown unless the failure-aware wrapper + // supplies it below. for fragment_id in &recomputed { state.row_state.remove(fragment_id); } - state.signatures.extend(entries); + state.invalidate_failure_evidence(recomputed.iter().copied()); + state.signatures.extend( + entries + .iter() + .map(|(fragment_id, signature)| (*fragment_id, signature.clone())), + ); state .signatures .retain(|fragment_id, _| live.contains(fragment_id)); state .row_state .retain(|fragment_id, _| live.contains(fragment_id)); + state + .failure_coverage + .retain(|fragment_id, _| live.contains(fragment_id)); + state + .failed_rows + .retain(|fragment_id, _| live.contains(fragment_id)); + if let Some(failures) = failures { + apply_failure_evidence( + dataset, + inputs.expect("failure evidence requires input fields"), + &entries, + &mut state, + failures, + ) + .await?; + } if state.row_state.is_empty() || state_unreadable || signatures_changed { state.basis.clear(); } @@ -1645,6 +2045,70 @@ struct FreshnessWrite { intent_digest: Option, } +async fn apply_failure_evidence( + dataset: &Dataset, + inputs: &InputFields, + accepted: &SignatureMap, + state: &mut FreshnessState, + update: &FailureEvidence, +) -> Result<()> { + update.validate()?; + let physical_rows: BTreeMap = { + let mut rows = BTreeMap::new(); + for fragment in dataset.get_fragments() { + rows.insert(fragment.id() as u32, fragment.physical_rows().await? as u64); + } + rows + }; + let mut applied = HashSet::new(); + for (fragment_id, covered) in &update.known_rows { + let Some(rows) = physical_rows.get(fragment_id) else { + continue; + }; + let Some(current) = signature_of(dataset, *fragment_id, inputs)? else { + continue; + }; + // Evidence belongs to the attempt that produced this accepted + // signature. A concurrent attempt may have stamped a newer signature + // into `state`; that does not make this attempt's rows valid for it. + if accepted.get(fragment_id) != Some(¤t) { + continue; + } + if state.signatures.get(fragment_id) != Some(¤t) { + continue; + } + if covered + .max() + .is_some_and(|offset| u64::from(offset) >= *rows) + { + return Err(invalid(format!( + "failure evidence for fragment {fragment_id} exceeds its {rows} physical rows" + ))); + } + if let Some(failed) = state.failed_rows.get_mut(fragment_id) { + *failed -= covered; + } + let coverage = state.failure_coverage.entry(*fragment_id).or_default(); + *coverage |= covered; + applied.insert(*fragment_id); + } + for (fragment_id, failed) in &update.failed_rows { + if !applied.contains(fragment_id) { + continue; + } + if !update.known_rows.contains_key(fragment_id) { + return Err(invalid(format!( + "failed rows for fragment {fragment_id} have no coverage update" + ))); + } + let current = state.failed_rows.entry(*fragment_id).or_default(); + *current |= failed; + } + state.failed_rows.retain(|_, rows| !rows.is_empty()); + state.failure_coverage.retain(|_, rows| !rows.is_empty()); + Ok(()) +} + async fn write_freshness_state( dataset: &mut Dataset, column: &str, @@ -1656,6 +2120,8 @@ async fn write_freshness_state( let state = state.normalize()?; let inline_empty = state.signatures.is_empty() && state.row_state.is_empty() + && state.failure_coverage.is_empty() + && state.failed_rows.is_empty() && state.basis.is_empty() && !retain_empty_sidecar; // The sidecar is durable before the metadata commit references it. @@ -1817,6 +2283,8 @@ pub async fn begin_row_repair( freshness.signatures = signatures_for(dataset, &live.iter().copied().collect::>(), inputs)?; freshness.row_state.clear(); + freshness.failure_coverage.clear(); + freshness.failed_rows.clear(); freshness.basis.clear(); } if state_unreadable || definition_changed { @@ -1826,6 +2294,8 @@ pub async fn begin_row_repair( .iter() .map(|id| (*id, DirtyRows::AllExcept(RoaringBitmap::new()))) .collect(); + freshness.failure_coverage.clear(); + freshness.failed_rows.clear(); } for (fragment_id, repair) in repairs { @@ -1842,6 +2312,9 @@ pub async fn begin_row_repair( let preserve = !state_unreadable && !definition_changed && freshness.signatures.get(fragment_id) == Some(¤t); + if !preserve { + freshness.invalidate_failure_evidence([*fragment_id]); + } let dirty = if preserve { freshness .row_state @@ -1865,6 +2338,12 @@ pub async fn begin_row_repair( freshness .row_state .retain(|fragment_id, _| live.contains(fragment_id)); + freshness + .failure_coverage + .retain(|fragment_id, _| live.contains(fragment_id)); + freshness + .failed_rows + .retain(|fragment_id, _| live.contains(fragment_id)); freshness.basis = basis.clone(); freshness.intent_id = Uuid::new_v4(); @@ -1958,6 +2437,66 @@ pub async fn finish_row_repair( filled_rows: &BTreeMap, skipped_rows: &BTreeMap, verified_version: Option, +) -> Result { + finish_row_repair_impl( + dataset, + column, + definition_version, + inputs, + token, + filled_rows, + skipped_rows, + None, + verified_version, + ) + .await +} + +/// Finalize a row repair together with the exact failure evidence observed for +/// those rows. +/// +/// `failures.known_rows` must be a subset of the rows assigned either a +/// value or an explicit skipped outcome. This permits a deferral to become +/// known non-failed without turning it into a failure. `failed_rows` must be a +/// subset of `known_rows`. The same token, input-signature and version +/// fences as [`finish_row_repair`] apply. +#[allow(clippy::too_many_arguments)] +pub async fn finish_row_repair_with_failures( + dataset: &mut Dataset, + column: &str, + definition_version: &str, + inputs: &InputFields, + token: &RowStateToken, + filled_rows: &BTreeMap, + skipped_rows: &BTreeMap, + failures: &FailureEvidence, + verified_version: Option, +) -> Result { + finish_row_repair_impl( + dataset, + column, + definition_version, + inputs, + token, + filled_rows, + skipped_rows, + Some(failures), + verified_version, + ) + .await +} + +#[allow(clippy::too_many_arguments)] +async fn finish_row_repair_impl( + dataset: &mut Dataset, + column: &str, + definition_version: &str, + inputs: &InputFields, + token: &RowStateToken, + filled_rows: &BTreeMap, + skipped_rows: &BTreeMap, + failures: Option<&FailureEvidence>, + verified_version: Option, ) -> Result { if !is_sha256_hex(&token.intent_digest) { return Err(invalid("row repair token is malformed".to_string())); @@ -1971,6 +2510,36 @@ pub async fn finish_row_repair( "row repair result names a fragment outside its token".to_string(), )); } + if let Some(failures) = failures { + failures.validate()?; + for (fragment_id, covered) in &failures.known_rows { + let selected = token.selected_rows.get(fragment_id).ok_or_else(|| { + invalid(format!( + "failure evidence names fragment {fragment_id} outside its token" + )) + })?; + let mut known = filled_rows.get(fragment_id).cloned().unwrap_or_default(); + if let Some(skipped) = skipped_rows.get(fragment_id) { + known |= skipped; + } + if !covered.is_subset(selected) || !covered.is_subset(&known) { + return Err(invalid(format!( + "failure evidence for fragment {fragment_id} covers rows without a final outcome" + ))); + } + } + for (fragment_id, failed) in &failures.failed_rows { + if !failures.known_rows.contains_key(fragment_id) + || skipped_rows + .get(fragment_id) + .is_none_or(|skipped| !failed.is_subset(skipped)) + { + return Err(invalid(format!( + "failed rows for fragment {fragment_id} must be an explicit skipped subset" + ))); + } + } + } dataset.checkout_latest().await?; if verified_version.is_some_and(|version| dataset.version().version != version) { return Err(invalid( @@ -2015,14 +2584,32 @@ pub async fn finish_row_repair( ))); } } - let before = state.row_state.clone(); + let before = state.clone(); + state.invalidate_failure_rows(&token.selected_rows); apply_row_repair( &mut state.row_state, &token.selected_rows, filled_rows, skipped_rows, )?; - if state.row_state == before { + if let Some(failures) = failures { + let accepted = token + .selected_rows + .keys() + .map(|fragment_id| { + state + .signatures + .get(fragment_id) + .cloned() + .map(|signature| (*fragment_id, signature)) + }) + .collect::>() + .ok_or_else(|| { + invalid("row repair failure evidence lost an accepted signature".to_string()) + })?; + apply_failure_evidence(dataset, inputs, &accepted, &mut state, failures).await?; + } + if state == before { return Ok(dataset.version().version); } if current_digest != token.intent_digest.as_str() { @@ -2868,6 +3455,41 @@ mod tests { .unwrap(); } + async fn stamp_v2( + dataset: &mut Dataset, + column: &str, + definition: &str, + entries: SignatureMap, + ) { + let reference = write_sidecar_bytes( + dataset, + encode_v2_sidecar(&FreshnessState { + signatures: entries, + ..Default::default() + }) + .unwrap(), + ) + .await + .unwrap(); + let describes = dataset.version().version.to_string(); + dataset + .update_field_metadata() + .update( + column, + [ + (SOURCE_SIGNATURE_META_KEY.to_string(), reference), + ( + DEFINITION_VERSION_META_KEY.to_string(), + definition.to_string(), + ), + (RECORDED_AT_VERSION_META_KEY.to_string(), describes), + ], + ) + .unwrap() + .await + .unwrap(); + } + fn fragment_signature(dataset: &Dataset, fragment_id: u32) -> String { signature_of(dataset, fragment_id, &inputs(dataset)) .unwrap() @@ -2894,6 +3516,15 @@ mod tests { (3, DirtyRows::Rows(RoaringBitmap::from_iter([1, 4, 9]))), (7, DirtyRows::AllExcept(RoaringBitmap::from_iter([0, 2]))), ]), + failure_coverage: FailureRows::from([ + (3, RoaringBitmap::from_iter([1, 2, 4, 9])), + (7, RoaringBitmap::from_iter([0, 1, 2, 3])), + ]), + failed_rows: FailureRows::from([ + (3, RoaringBitmap::from_iter([1])), + (7, RoaringBitmap::from_iter([1, 3])), + ]), + failure_evidence_recorded: true, }; let encoded = encode_sidecar(&state).unwrap(); assert_eq!(encoded[4], SIDECAR_FORMAT); @@ -2913,9 +3544,283 @@ mod tests { let state = decode_sidecar(&encode_legacy_sidecar(&signatures).unwrap()).unwrap(); assert_eq!(state.signatures, signatures); assert!(state.row_state.is_empty()); + assert!(state.failure_coverage.is_empty()); + assert!(state.failed_rows.is_empty()); + assert!(!state.failure_evidence_recorded); assert!(state.basis.is_empty()); } + #[test] + fn v2_sidecar_decodes_with_unknown_failure_evidence() { + let state = FreshnessState { + signatures: SignatureMap::from([(1, "0000000000000001".to_string())]), + row_state: RowStateMap::from([(1, DirtyRows::Rows(RoaringBitmap::from_iter([2, 3])))]), + basis: "legacy-intent".to_string(), + intent_id: Uuid::from_u128(1), + ..Default::default() + }; + let decoded = decode_sidecar(&encode_v2_sidecar(&state).unwrap()).unwrap(); + assert_eq!(decoded, state); + assert!(decoded.failure_coverage.is_empty()); + assert!(decoded.failed_rows.is_empty()); + assert!(!decoded.failure_evidence_recorded); + } + + #[tokio::test] + async fn a_first_partial_write_keeps_untouched_legacy_rows_unknown() { + let dir = tempfile::tempdir().unwrap(); + let mut dataset = table(dir.path().to_str().unwrap()).await; + let definition = definition_version(EXPRESSION); + let ids = inputs(&dataset); + let entries = signatures_for(&dataset, &[0, 1], &ids).unwrap(); + stamp_v2(&mut dataset, COLUMN, &definition, entries).await; + + let StrictFreshness::Uncertified { staleness } = + strict_freshness_against(&dataset, COLUMN, &definition, &ids) + .await + .unwrap() + else { + panic!("the v2 sidecar should yield a readable uncertified plan"); + }; + assert_eq!(staleness, StalenessPlan::default()); + + let selected = RoaringBitmap::from_iter([1, 2]); + let repairs = BTreeMap::from([(0, repair(&dataset, 0, selected.clone()))]); + let token = begin_row_repair( + &mut dataset, + COLUMN, + &definition, + &ids, + "legacy-partial", + &repairs, + ) + .await + .unwrap(); + finish_row_repair_with_failures( + &mut dataset, + COLUMN, + &definition, + &ids, + &token, + &BTreeMap::from([(0, RoaringBitmap::from_iter([2]))]), + &BTreeMap::from([(0, RoaringBitmap::from_iter([1]))]), + &FailureEvidence { + known_rows: BTreeMap::from([(0, selected)]), + failed_rows: BTreeMap::from([(0, RoaringBitmap::from_iter([1]))]), + }, + None, + ) + .await + .unwrap(); + + let StrictFreshness::Certified { + staleness, + failures, + } = strict_freshness_against(&dataset, COLUMN, &definition, &ids) + .await + .unwrap() + else { + panic!("partial v3 evidence is still readable"); + }; + assert_eq!( + staleness.row_dirty.get(&0), + Some(&DirtyRows::Rows(RoaringBitmap::from_iter([1]))) + ); + assert_eq!( + failures.known_rows.get(&0), + Some(&RoaringBitmap::from_iter([1, 2])), + "only the rows this partial publication observed are covered" + ); + assert_eq!( + failures.failed_rows.get(&0), + Some(&RoaringBitmap::from_iter([1])) + ); + assert!( + !failures.known_rows.contains_key(&1), + "the untouched legacy fragment remains unknown" + ); + } + + #[tokio::test] + async fn a_successful_retry_clears_only_its_failed_rows() { + let dir = tempfile::tempdir().unwrap(); + let mut dataset = table(dir.path().to_str().unwrap()).await; + let definition = definition_version(EXPRESSION); + let ids = inputs(&dataset); + let signatures = signatures_for(&dataset, &[0, 1], &ids).unwrap(); + stamp_v2(&mut dataset, COLUMN, &definition, signatures).await; + + let first_selected = RoaringBitmap::from_iter([1, 2]); + let first_repairs = BTreeMap::from([(0, repair(&dataset, 0, first_selected.clone()))]); + let first = begin_row_repair( + &mut dataset, + COLUMN, + &definition, + &ids, + "failed-wave", + &first_repairs, + ) + .await + .unwrap(); + finish_row_repair_with_failures( + &mut dataset, + COLUMN, + &definition, + &ids, + &first, + &BTreeMap::from([(0, RoaringBitmap::from_iter([2]))]), + &BTreeMap::from([(0, RoaringBitmap::from_iter([1]))]), + &FailureEvidence { + known_rows: BTreeMap::from([(0, first_selected)]), + failed_rows: BTreeMap::from([(0, RoaringBitmap::from_iter([1]))]), + }, + None, + ) + .await + .unwrap(); + + let retry_selected = RoaringBitmap::from_iter([1]); + let retry_repairs = BTreeMap::from([(0, repair(&dataset, 0, retry_selected.clone()))]); + let retry = begin_row_repair( + &mut dataset, + COLUMN, + &definition, + &ids, + "successful-retry", + &retry_repairs, + ) + .await + .unwrap(); + finish_row_repair_with_failures( + &mut dataset, + COLUMN, + &definition, + &ids, + &retry, + &BTreeMap::from([(0, retry_selected.clone())]), + &BTreeMap::new(), + &FailureEvidence { + known_rows: BTreeMap::from([(0, retry_selected)]), + failed_rows: BTreeMap::new(), + }, + None, + ) + .await + .unwrap(); + + let StrictFreshness::Certified { failures, .. } = + strict_freshness_against(&dataset, COLUMN, &definition, &ids) + .await + .unwrap() + else { + panic!("status should remain readable"); + }; + assert!(!failures.failed_rows.contains_key(&0)); + assert_eq!( + failures.known_rows.get(&0), + Some(&RoaringBitmap::from_iter([1, 2])) + ); + } + + #[tokio::test] + async fn an_old_attempt_cannot_rewrite_a_newer_signature_evidence() { + let dir = tempfile::tempdir().unwrap(); + let mut dataset = table(dir.path().to_str().unwrap()).await; + let definition = definition_version(EXPRESSION); + let ids = inputs(&dataset); + let old = dataset.clone(); + let old_inputs = ids.clone(); + let old_signature = signature_of(&old, 0, &old_inputs).unwrap().unwrap(); + + rewrite_value(&mut dataset, 0).await; + let new_signature = signature_of(&dataset, 0, &inputs(&dataset)) + .unwrap() + .unwrap(); + assert_ne!(old_signature, new_signature); + + let current_entries = signatures_for(&dataset, &[0, 1], &inputs(&dataset)).unwrap(); + write_signatures(&mut dataset, COLUMN, &definition, current_entries.clone()) + .await + .unwrap(); + let StoredFreshnessState::Present { + state: mut current_state, + .. + } = stored_freshness_state(&dataset, COLUMN).await + else { + panic!("new attempt state"); + }; + current_state + .failure_coverage + .insert(0, RoaringBitmap::from_iter([3, 4])); + current_state + .failed_rows + .insert(0, RoaringBitmap::from_iter([4])); + let write = write_freshness_state( + &mut dataset, + COLUMN, + &definition, + current_state.clone(), + false, + None, + ) + .await + .unwrap(); + assert!(write.version > 0); + let new_evidence = FailureEvidence { + known_rows: current_state.failure_coverage.clone(), + failed_rows: current_state.failed_rows.clone(), + }; + + let record = record_freshness_with_failures( + &mut dataset, + None, + COLUMN, + &definition, + &old_inputs, + SignatureMap::from([(0, old_signature)]), + &FailureEvidence { + known_rows: BTreeMap::from([(0, RoaringBitmap::from_iter([1, 2]))]), + failed_rows: BTreeMap::from([(0, RoaringBitmap::from_iter([1]))]), + }, + ) + .await + .unwrap(); + assert_eq!((record.recorded, record.moved), (0, 1)); + + let StrictFreshness::Certified { failures, .. } = + strict_freshness_against(&dataset, COLUMN, &definition, &inputs(&dataset)) + .await + .unwrap() + else { + panic!("current signature evidence should remain readable"); + }; + assert_eq!(failures.known_rows, new_evidence.known_rows); + assert_eq!(failures.failed_rows, new_evidence.failed_rows); + assert_eq!( + failures.failed_rows.get(&0), + Some(&RoaringBitmap::from_iter([4])), + "old S1 failure must not replace the S2 failure" + ); + assert!(!failures.known_rows.contains_key(&1)); + } + + #[test] + fn failure_evidence_must_be_a_subset_of_coverage() { + let state = FreshnessState { + signatures: SignatureMap::from([(3, "0000000000000003".to_string())]), + failure_coverage: FailureRows::from([(3, RoaringBitmap::from_iter([1]))]), + failed_rows: FailureRows::from([(3, RoaringBitmap::from_iter([2]))]), + ..Default::default() + }; + let error = encode_sidecar(&state).unwrap_err(); + assert!( + error + .to_string() + .contains("failed rows are not covered by failure evidence"), + "{error}" + ); + } + #[tokio::test] async fn a_row_repair_clears_filled_rows_and_retains_skipped_rows() { let dir = tempfile::tempdir().unwrap(); diff --git a/rust/lancedb/src/table/function_column_status.rs b/rust/lancedb/src/table/function_column_status.rs new file mode 100644 index 000000000..729ea33ba --- /dev/null +++ b/rust/lancedb/src/table/function_column_status.rs @@ -0,0 +1,303 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The LanceDB Authors + +//! Read-only status for Function-backed table columns. + +use serde::{Deserialize, Serialize}; + +use crate::function::FunctionVersionRef; +use crate::{Error, Result}; + +/// Stable HTTP error code for a status read waiting on its matching finalize. +pub const FINALIZATION_PENDING_ERROR_CODE: &str = "function_column_status_finalization_pending"; + +/// One exact immutable Function revision attached to a table output. +pub type FunctionColumnVersionStatus = FunctionVersionRef; + +/// Latest matching refresh Job projected for a Function output. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct FunctionColumnLastRefreshJob { + pub job_id: String, + pub state: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub automatic: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub completed_at_millis: Option, +} + +/// Automatic-refresh scheduling state that the server can prove. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum FunctionColumnAutoRefreshState { + Disabled, + Running, + UpToDate, + Ready, + WaitingInterval, + WaitingBackoff, + Blocked, + Unobservable, +} + +impl FunctionColumnAutoRefreshState { + pub const fn as_str(self) -> &'static str { + match self { + Self::Disabled => "disabled", + Self::Running => "running", + Self::UpToDate => "up_to_date", + Self::Ready => "ready", + Self::WaitingInterval => "waiting_interval", + Self::WaitingBackoff => "waiting_backoff", + Self::Blocked => "blocked", + Self::Unobservable => "unobservable", + } + } +} + +/// Why an enabled automatic refresh is not currently schedulable. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum FunctionColumnAutoRefreshBlockedReason { + TableHasMemWal, + InputsNotSettled, + DependencyInFlight, + TableCapacity, + Cancelled, + PreparationBackoff, +} + +impl FunctionColumnAutoRefreshBlockedReason { + pub const fn as_str(self) -> &'static str { + match self { + Self::TableHasMemWal => "table_has_mem_wal", + Self::InputsNotSettled => "inputs_not_settled", + Self::DependencyInFlight => "dependency_in_flight", + Self::TableCapacity => "table_capacity", + Self::Cancelled => "cancelled", + Self::PreparationBackoff => "preparation_backoff", + } + } +} + +/// Automatic-refresh projection for a Function output. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct FunctionColumnAutoRefreshStatus { + pub enabled: bool, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub interval_seconds: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub active_job_id: Option, + pub state: FunctionColumnAutoRefreshState, + pub failure_streak: u32, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub effective_cooldown_millis: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub blocked_reason: Option, + /// Null is intentional until the Lance Agent exposes its scheduler decision. + pub next_run_at_millis: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub next_run_at_reason: Option, +} + +/// Exact status of one Function output at one table version. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct FunctionColumnStatus { + pub column: String, + pub binding_id: String, + pub function: FunctionColumnVersionStatus, + pub table_version: u64, + pub rows: u64, + pub fresh: u64, + pub stale: u64, + pub never_computed: u64, + pub failed: u64, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub last_refresh_job: Option, + pub auto_refresh: FunctionColumnAutoRefreshStatus, +} + +impl FunctionColumnStatus { + /// Whether this status is safe to expose as an exact four-way count. + pub fn validate(&self) -> Result<()> { + let classified = self + .fresh + .checked_add(self.stale) + .and_then(|count| count.checked_add(self.never_computed)) + .and_then(|count| count.checked_add(self.failed)); + if classified != Some(self.rows) { + return Err(Error::Runtime { + message: format!( + "Function column '{}' status counts do not partition {} rows", + self.column, self.rows + ), + }); + } + if self.auto_refresh.state == FunctionColumnAutoRefreshState::Blocked + && self.auto_refresh.blocked_reason.is_none() + { + return Err(Error::Runtime { + message: format!( + "Function column '{}' reported a blocked scheduler state without a reason", + self.column + ), + }); + } + if self.auto_refresh.state == FunctionColumnAutoRefreshState::Unobservable + && self.auto_refresh.next_run_at_millis.is_none() + && self + .auto_refresh + .next_run_at_reason + .as_deref() + .is_none_or(str::is_empty) + { + return Err(Error::Runtime { + message: format!( + "Function column '{}' omitted a reason for an unavailable next run", + self.column + ), + }); + } + Ok(()) + } +} + +/// Validate one GET response, whose objects must all describe one snapshot. +pub fn validate_statuses(statuses: &[FunctionColumnStatus]) -> Result<()> { + let mut table_versions = statuses.iter().map(|status| status.table_version); + let Some(first) = table_versions.next() else { + return Ok(()); + }; + if table_versions.any(|version| version != first) { + return Err(Error::Runtime { + message: "Function column statuses came from different table versions".to_string(), + }); + } + statuses.iter().try_for_each(FunctionColumnStatus::validate) +} + +/// Whether an error is the typed transient response used while the data +/// version's matching freshness finalization is still in progress. +pub fn is_finalization_pending(error: &Error) -> bool { + #[cfg(feature = "remote")] + { + error.status_code() == Some(reqwest::StatusCode::SERVICE_UNAVAILABLE) + && error + .remote_error_metadata() + .and_then(|metadata| metadata.error_code.as_deref()) + == Some(FINALIZATION_PENDING_ERROR_CODE) + } + #[cfg(not(feature = "remote"))] + { + let _ = error; + false + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn status(column: &str, table_version: u64) -> FunctionColumnStatus { + FunctionColumnStatus { + column: column.to_string(), + binding_id: "fb_01".to_string(), + function: FunctionVersionRef { + name: "embed".to_string(), + namespace_path: Vec::new(), + object_id: "embed".to_string(), + location: "objects/embed".to_string(), + version: "7".to_string(), + manifest_digest: "sha256:abc".to_string(), + }, + table_version, + rows: 10, + fresh: 6, + stale: 2, + never_computed: 1, + failed: 1, + last_refresh_job: None, + auto_refresh: FunctionColumnAutoRefreshStatus { + enabled: true, + interval_seconds: Some(60), + active_job_id: None, + state: FunctionColumnAutoRefreshState::Unobservable, + failure_streak: 0, + effective_cooldown_millis: None, + blocked_reason: None, + next_run_at_millis: None, + next_run_at_reason: Some( + "authoritative_scheduler_projection_unavailable".to_string(), + ), + }, + } + } + + #[test] + fn validates_a_single_version_and_four_way_partition() { + validate_statuses(&[status("a", 42), status("b", 42)]).unwrap(); + + let error = validate_statuses(&[status("a", 42), status("b", 43)]).unwrap_err(); + assert!(error.to_string().contains("different table versions")); + + let mut invalid = status("a", 42); + invalid.fresh += 1; + let error = validate_statuses(&[invalid]).unwrap_err(); + assert!(error.to_string().contains("do not partition 10 rows")); + } + + #[test] + fn wire_keeps_job_and_auto_refresh_metadata() { + let mut value = status("a", 42); + value.last_refresh_job = Some(FunctionColumnLastRefreshJob { + job_id: "job-7".to_string(), + state: "finished".to_string(), + automatic: Some(true), + completed_at_millis: Some(1_790_000_000_000), + }); + let encoded = serde_json::to_value(value).unwrap(); + assert_eq!(encoded["last_refresh_job"]["job_id"], "job-7"); + assert_eq!(encoded["auto_refresh"]["enabled"], true); + assert_eq!( + encoded["auto_refresh"]["next_run_at_reason"], + "authoritative_scheduler_projection_unavailable" + ); + } + + #[test] + fn up_to_date_wire_is_stable() { + assert_eq!( + serde_json::to_string(&FunctionColumnAutoRefreshState::UpToDate).unwrap(), + "\"up_to_date\"" + ); + } + + #[test] + fn blocked_auto_refresh_wire_keeps_its_reason() { + let mut value = status("a", 42); + value.auto_refresh.state = FunctionColumnAutoRefreshState::Blocked; + value.auto_refresh.failure_streak = 3; + value.auto_refresh.effective_cooldown_millis = Some(8000); + value.auto_refresh.blocked_reason = + Some(FunctionColumnAutoRefreshBlockedReason::PreparationBackoff); + value.auto_refresh.next_run_at_reason = None; + + let encoded = serde_json::to_value(&value).unwrap(); + assert_eq!(encoded["auto_refresh"]["state"], "blocked"); + assert_eq!( + encoded["auto_refresh"]["blocked_reason"], + "preparation_backoff" + ); + assert_eq!(encoded["auto_refresh"]["failure_streak"], 3); + assert_eq!(encoded["auto_refresh"]["effective_cooldown_millis"], 8000); + value.validate().unwrap(); + + value.auto_refresh.blocked_reason = None; + assert!( + value + .validate() + .unwrap_err() + .to_string() + .contains("blocked scheduler state without a reason") + ); + } +}