feat: expose exact Function column freshness (#4958)

## Problem

Function-backed columns need an exact `fresh`, `stale`,
`never_computed`, and
`failed` partition for the table version a caller is reading. The
existing
error listing is capped, and composing it with job and schema listings
cannot
prove which rows currently failed.

## Behavior

- Adds a strict freshness reader beside the existing permissive planner.
It
  distinguishes an absent certificate from legacy evidence that omits
  failures, and from a current certificate carrying failure coverage.
- Extends the immutable freshness sidecar with per-fragment known and
failed
  row bitmaps. Partial or `WHERE` publications cover only observed rows;
  untouched rows stay unknown. Deleted rows do not block classification.
- Adds failure-aware whole-fragment and row-repair writers. A rejected
older
attempt cannot overwrite failure evidence already recorded for a newer
input
  signature.
- Adds `column_status`, `column_statuses`, and `wait_fresh` to Python
and the
  corresponding Node methods. The client calls the server's exact status
endpoint, validates the four-way partition and one-version guarantee,
and
preserves the complete Function version, last-job, and automatic-refresh
  metadata.
- Adds an MDX-commented draft usage section to the Enterprise
computed-column
refresh guide, covering Python, Rust, and TypeScript and remaining
hidden
  until the server integration ships.
- Sends `column`, `branch`, and exact `version` selectors to the status
route.
A pinned handle rejects a nonempty response from any other table
version;
  latest handles keep the existing read-freshness headers.
- Retries `wait_fresh` only for HTTP 503 with
`function_column_status_finalization_pending`. Existing failed rows
remain
  visible and do not keep the wait pending; other errors propagate.
- Models the shared scheduler projection, including `up_to_date`, real
`blocked` states, their reason, failure streak, and effective cooldown.

## Validation

This is the SDK side of ENT-3563. Sophon's failure-coverage publication
and
status endpoint are a separate integration; deployed end-to-end and
whole-ticket
acceptance remain pending.

- Rust: 9 exact freshness/status tests passed, plus 1 exact
latest-handle
  freshness-header test. The pinned transport test was extended with a
wrong-version response and rerun at `ebd6649a` (1 passed). Coverage
includes untouched partial rows,
  selected-row failure clearing, cross-generation evidence, typed and
non-retried errors, `up_to_date`, blocked scheduler fields, and
exact-version
  requests.
- Python: 2 native PyO3 remote-transport tests passed together,
including
typed retry; one local facade test reaches the existing unsupported
boundary.
- Node: the NAPI addon built successfully, TypeScript and Biome passed,
and the
selected native remote-transport test passed 1/0, including
exact-version
  selection and blocked scheduler metadata.
- `cargo fmt`, `git diff --check`, Ruff format/lint, Python compile,
Biome, and
  TypeScript checks passed.

Validation is mapped by the relevant unchanged source rather than
treated as one
final-head run: the nine Rust tests ran at `e4d51fc8`, the latest-header
and Python
remote tests at `31e68b7d`, the local Python test and Node build at
`6f213584`, and
the final TypeScript/Jest delta at `4f2d5c27`. The final product tree
excludes those five private EC2 runner files and adds
six generated TypeDoc files plus one hidden draft usage section in the
computed-column refresh guide. The Node delta corrected a private
TypeScript
command and reused the unchanged native addon.
This commit is contained in:
Xuanwo
2026-11-11 19:12:06 +08:00
committed by GitHub
parent 34e8b8087b
commit 7cc3904a7e
25 changed files with 3055 additions and 41 deletions
+64
View File
@@ -704,6 +704,47 @@ so this is safe to call repeatedly.
***
### functionColumnStatus()
```ts
abstract functionColumnStatus(column): Promise<FunctionColumnStatus>
```
Return exact freshness status for one Function-backed output column.
#### Parameters
* **column**: `string`
#### Returns
`Promise`&lt;[`FunctionColumnStatus`](../interfaces/FunctionColumnStatus.md)&gt;
***
### functionColumnStatuses()
```ts
abstract functionColumnStatuses(column?): Promise<FunctionColumnStatus[]>
```
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`&lt;[`FunctionColumnStatus`](../interfaces/FunctionColumnStatus.md)[]&gt;
***
### functionErrors()
```ts
@@ -1598,3 +1639,26 @@ Waits for asynchronous indexing to complete on the table.
#### Returns
`Promise`&lt;`void`&gt;
***
### waitFresh()
```ts
abstract waitFresh(column, timeoutSeconds?): Promise<FunctionColumnStatus>
```
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`&lt;[`FunctionColumnStatus`](../interfaces/FunctionColumnStatus.md)&gt;
+4
View File
@@ -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)
@@ -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";
```
@@ -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;
```
@@ -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;
```
@@ -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;
```
+8
View File
@@ -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
@@ -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:
<CodeGroup>
```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);
```
</CodeGroup>
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
+63
View File
@@ -647,6 +647,69 @@ describe("remote connection", () => {
]);
});
it("passes a pinned version and decodes Function scheduler status", async () => {
const queries: Record<string, string>[] = [];
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(
+4
View File
@@ -154,6 +154,10 @@ export {
export {
Table,
FunctionColumnAutoRefreshStatus,
FunctionColumnLastRefreshJob,
FunctionColumnStatus,
FunctionColumnVersionStatus,
Branches,
BranchColumnSummary,
BranchColumnChange,
+111
View File
@@ -258,6 +258,66 @@ export interface RTEWriteSpec {
writerConfigDefaults?: Record<string, string>;
}
/** 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<FunctionErrors>;
/**
* 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<FunctionColumnStatus[]>;
/** Return exact freshness status for one Function-backed output column. */
abstract functionColumnStatus(column: string): Promise<FunctionColumnStatus>;
/**
* 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<FunctionColumnStatus>;
/**
* 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<FunctionColumnStatus[]> {
return (await this.inner.functionColumnStatuses(
column,
)) as FunctionColumnStatus[];
}
async functionColumnStatus(column: string): Promise<FunctionColumnStatus> {
return (await this.inner.functionColumnStatus(
column,
)) as FunctionColumnStatus;
}
async waitFresh(
column: string,
timeoutSeconds?: number,
): Promise<FunctionColumnStatus> {
validateNonNegativeInteger(timeoutSeconds, "timeoutSeconds");
return (await this.inner.waitForFunctionColumn(
column,
timeoutSeconds,
)) as FunctionColumnStatus;
}
async refreshMaterializedView(
sourceVersion?: number,
full?: boolean,
+151
View File
@@ -534,6 +534,55 @@ impl Table {
Ok(errors.into())
}
#[napi(catch_unwind)]
pub async fn function_column_statuses(
&self,
column: Option<String>,
) -> napi::Result<Vec<FunctionColumnStatus>> {
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<FunctionColumnStatus> {
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<i64>,
) -> napi::Result<FunctionColumnStatus> {
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<String>,
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<bool>,
pub completed_at_millis: Option<BigInt>,
}
/// Automatic-refresh projection for a Function output.
#[napi(object)]
#[derive(Clone, Debug)]
pub struct FunctionColumnAutoRefreshStatus {
pub enabled: bool,
pub interval_seconds: Option<BigInt>,
pub active_job_id: Option<String>,
pub state: String,
pub failure_streak: u32,
pub effective_cooldown_millis: Option<BigInt>,
pub blocked_reason: Option<String>,
pub next_run_at_millis: Option<BigInt>,
pub next_run_at_reason: Option<String>,
}
/// 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<FunctionColumnLastRefreshJob>,
pub auto_refresh: FunctionColumnAutoRefreshStatus,
}
impl From<lancedb::table::FunctionColumnStatus> 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<lancedb::function::FunctionErrors> for FunctionErrors {
fn from(errors: lancedb::function::FunctionErrors) -> Self {
Self {
+14
View File
@@ -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",
+45
View File
@@ -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: ...
+14
View File
@@ -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:
+70 -1
View File
@@ -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,
+131
View File
@@ -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"])
+10
View File
@@ -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):
+8 -3
View File
@@ -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::<FunctionErrors>()?;
m.add_class::<FunctionErrorRecord>()?;
m.add_class::<FunctionErrorFragment>()?;
m.add_class::<FunctionColumnStatus>()?;
m.add_class::<FunctionColumnVersionStatus>()?;
m.add_class::<FunctionColumnLastRefreshJob>()?;
m.add_class::<FunctionColumnAutoRefreshStatus>()?;
m.add_class::<RefreshMaterializedViewResult>()?;
m.add_class::<AlterColumnsResult>()?;
m.add_class::<UpdateFieldMetadataResult>()?;
+175
View File
@@ -892,6 +892,136 @@ impl From<lancedb::function::FunctionErrors> 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<String>,
pub object_id: String,
pub location: String,
pub version: String,
pub manifest_digest: String,
}
impl From<lancedb::function::FunctionVersionRef> 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<bool>,
pub completed_at_millis: Option<i64>,
}
impl From<lancedb::table::FunctionColumnLastRefreshJob> for FunctionColumnLastRefreshJob {
fn from(job: lancedb::table::FunctionColumnLastRefreshJob) -> Self {
Self {
job_id: job.job_id,
state: job.state,
automatic: job.automatic,
completed_at_millis: job.completed_at_millis,
}
}
}
/// 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<u64>,
pub active_job_id: Option<String>,
pub state: String,
pub failure_streak: u32,
pub effective_cooldown_millis: Option<u64>,
pub blocked_reason: Option<String>,
pub next_run_at_millis: Option<i64>,
pub next_run_at_reason: Option<String>,
}
impl From<lancedb::table::FunctionColumnAutoRefreshStatus> 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<FunctionColumnLastRefreshJob>,
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<lancedb::table::FunctionColumnStatus> 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<String>,
) -> PyResult<Bound<'_, PyAny>> {
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::<Vec<_>>())
})
}
pub fn column_status(self_: PyRef<'_, Self>, column: String) -> PyResult<Bound<'_, PyAny>> {
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<Bound<'py, PyAny>>,
) -> PyResult<Bound<'py, PyAny>> {
let timeout = timeout
.map(|timeout| timeout.extract::<std::time::Duration>())
.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(
+19
View File
@@ -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<reqwest::StatusCode> {
match self {
Self::Http { status_code, .. } => *status_code,
Self::Retry { status_code, .. } => *status_code,
Self::TableNotFound { source, .. } => {
source.downcast_ref::<Self>().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
+494 -1
View File
@@ -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<S: HttpSend> BaseTable for RemoteTable<S> {
})
}
async fn function_column_statuses(
&self,
column: Option<&str>,
) -> Result<Vec<FunctionColumnStatus>> {
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<FunctionColumnStatus> =
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<AlterColumnsResult> {
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<_>>(),
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.
+85
View File
@@ -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<Vec<FunctionColumnStatus>> {
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<AlterColumnsResult>;
/// 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<Vec<FunctionColumnStatus>> {
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<FunctionColumnStatus> {
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<std::time::Duration>,
) -> Result<FunctionColumnStatus> {
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
File diff suppressed because it is too large Load Diff
@@ -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<bool>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub completed_at_millis: Option<i64>,
}
/// 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<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub active_job_id: Option<String>,
pub state: FunctionColumnAutoRefreshState,
pub failure_streak: u32,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub effective_cooldown_millis: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub blocked_reason: Option<FunctionColumnAutoRefreshBlockedReason>,
/// Null is intentional until the Lance Agent exposes its scheduler decision.
pub next_run_at_millis: Option<i64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub next_run_at_reason: Option<String>,
}
/// 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<FunctionColumnLastRefreshJob>,
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")
);
}
}