feat: expose the remaining refresh options on the SDKs (#4890)

The server's backfill request takes a row filter, failure limits,
per-call timeouts and worker sizing, and SQL exposes all of them through
`REFRESH COLUMN ... WHERE ... WITH (...)`, but the typed SDKs could send
only batch_size, GPU placement and the retry and error policies. Anyone
needing a filtered refresh or a bounded skip policy had to drop to SQL.

This adds `where`, `max_failed_fraction`, `max_failed_rows`,
`row_timeout_seconds`, `init_timeout_seconds`, `num_workers` and
`max_executor_processes_per_worker` to Python, Rust and Node, passed
through unchanged. `RefreshColumnOptions` now serializes straight into
the request body, so the three SDKs share one producer. Node gets a
`RefreshColumnOptions` object, which also carries the options it lacked
before. A LanceDB Embedded refresh rejects every option other than
batch_size and GPU placement, naming the option.
This commit is contained in:
Wyatt Alt
2026-11-06 09:19:27 -08:00
committed by GitHub
parent d379bbb722
commit 2af85bea89
18 changed files with 974 additions and 53 deletions
+22 -3
View File
@@ -1082,7 +1082,7 @@ for await (const batch of table.query()) {
### refreshColumn()
```ts
abstract refreshColumn(column): Promise<RefreshColumnResult>
abstract refreshColumn(column, options?): Promise<RefreshColumnResult>
```
Fill the rows of a computed column that hold no value yet.
@@ -1098,6 +1098,10 @@ Enterprise a refresh runs as a server job, through [Table#refreshColumnAsync](Ta
* **column**: `string`
The name of the computed column to fill.
* **options?**: [`RefreshColumnOptions`](../interfaces/RefreshColumnOptions.md)
Refresh options, all LanceDB
Enterprise only; see [Table#refreshColumnAsync](Table.md#refreshcolumnasync).
#### Returns
`Promise`&lt;[`RefreshColumnResult`](../interfaces/RefreshColumnResult.md)&gt;
@@ -1110,7 +1114,7 @@ number of rows filled and the new version number of the table.
### refreshColumnAsync()
```ts
abstract refreshColumnAsync(column): Promise<Job>
abstract refreshColumnAsync(column, options?): Promise<Job>
```
Like [Table#refreshColumn](Table.md#refreshcolumn), but returns a handle to the refresh
@@ -1130,6 +1134,16 @@ LanceDB Enterprise it is the server's backfill job.
* **column**: `string`
The name of the computed column to fill.
* **options?**: [`RefreshColumnOptions`](../interfaces/RefreshColumnOptions.md)
Refresh options, mirroring the SQL
`REFRESH COLUMN ... WHERE ... WITH (...)` form: `where` narrows the rows,
`errorPolicy`, `maxFailedFraction`, `maxFailedRows`, `rowTimeoutSeconds`
and `initTimeoutSeconds` govern failing rows, and `batchSize`,
`gpuClass`, `gpuCount`, `numWorkers` and `maxExecutorProcessesPerWorker`
size the workers. All are LanceDB Enterprise only; a LanceDB Embedded
refresh rejects every option other than `batchSize`, `gpuClass` and
`gpuCount`.
#### Returns
`Promise`&lt;[`Job`](Job.md)&gt;
@@ -1137,7 +1151,12 @@ LanceDB Enterprise it is the server's backfill job.
#### Example
```ts
const job = await table.refreshColumnAsync("doubled");
const job = await table.refreshColumnAsync("embedding", {
where: "published_at > '2026-01-01'",
errorPolicy: "skip",
maxFailedFraction: 0.01,
rowTimeoutSeconds: 30,
});
await job.wait();
console.log(await job.status()); // "finished"
```
+1
View File
@@ -122,6 +122,7 @@
- [QueryExecutionOptions](interfaces/QueryExecutionOptions.md)
- [RTEStats](interfaces/RTEStats.md)
- [RTEWriteSpec](interfaces/RTEWriteSpec.md)
- [RefreshColumnOptions](interfaces/RefreshColumnOptions.md)
- [RefreshColumnResult](interfaces/RefreshColumnResult.md)
- [RefreshMaterializedViewResult](interfaces/RefreshMaterializedViewResult.md)
- [RemovalStats](interfaces/RemovalStats.md)
@@ -0,0 +1,124 @@
[**@lancedb/lancedb**](../README.md) • **Docs**
***
[@lancedb/lancedb](../globals.md) / RefreshColumnOptions
# Interface: RefreshColumnOptions
Options for `Table.refreshColumn` and `Table.refreshColumnAsync`. Every
option is LanceDB Enterprise only; a LanceDB Embedded refresh ignores
`batchSize`, `gpuClass` and `gpuCount` and rejects the rest.
## Properties
### batchSize?
```ts
optional batchSize: number;
```
Maximum rows per Function call; must be greater than zero.
***
### errorPolicy?
```ts
optional errorPolicy: string;
```
`"fail"` (default) stops on a row error; `"skip"` writes null and
records the error, subject to the failure limits.
***
### gpuClass?
```ts
optional gpuClass: string;
```
GPU class for Function workers.
***
### gpuCount?
```ts
optional gpuCount: number;
```
Positive aggregate GPU budget for Function workers.
***
### initTimeoutSeconds?
```ts
optional initTimeoutSeconds: number;
```
Deadline for Function initialization, in seconds.
***
### maxExecutorProcessesPerWorker?
```ts
optional maxExecutorProcessesPerWorker: number;
```
Cap on executor processes per Function worker.
***
### maxFailedFraction?
```ts
optional maxFailedFraction: number;
```
Under a skip policy, fail the job once more than this share of rows
(0 to 1) has failed.
***
### maxFailedRows?
```ts
optional maxFailedRows: number;
```
Under a skip policy, fail the job once more than this many rows have
failed.
***
### numWorkers?
```ts
optional numWorkers: number;
```
Worker count for an expression column; Function columns reject it.
***
### rowTimeoutSeconds?
```ts
optional rowTimeoutSeconds: number;
```
Deadline for one Function call, in seconds.
***
### where?
```ts
optional where: string;
```
Only refresh the rows matching this SQL filter.
@@ -241,10 +241,59 @@ GPU class discovery.
<Note>
The Rust GPU discovery and refresh controls are not in the latest Rust
release. TypeScript does not expose GPU class discovery or placement; a
tracking issue is needed before adding equivalent controls.
release. TypeScript does not expose GPU class discovery; pass `gpuClass`
and `gpuCount` in the `refreshColumnAsync` options.
</Note>
The job APIs also take the row filter and failure handling of the SQL
`REFRESH COLUMN ... WHERE ... WITH (...)` form, so a refresh can be narrowed
to new rows and told to skip rows whose Function call fails:
<CodeGroup>
```python
job = table.refresh_column_async(
"embedding",
where="published_at > '2026-01-01'",
error_policy="skip",
max_failed_fraction=0.01,
row_timeout_seconds=30,
)
```
```typescript
const job = await table.refreshColumnAsync("embedding", {
where: "published_at > '2026-01-01'",
errorPolicy: "skip",
maxFailedFraction: 0.01,
rowTimeoutSeconds: 30,
});
```
```rust
use lancedb::{function::FunctionErrorPolicy, table::RefreshColumnOptions};
let _job = table
.refresh_column_async_with_options(
"embedding",
RefreshColumnOptions::default()
.where_clause("published_at > '2026-01-01'")
.error_policy(FunctionErrorPolicy::Skip)
.max_failed_fraction(0.01)
.row_timeout_seconds(30.0),
)
.await?;
```
</CodeGroup>
`max_failed_rows`, `init_timeout_seconds`, `num_workers` and
`max_executor_processes_per_worker` are accepted the same way; the
[SQL refresh options](/sql/dml/refresh-column#options) describe each one.
The SDK rejects a blank `where`, a `max_failed_fraction` outside 0 to 1, a
timeout that is not a positive number and a zero worker or process count
before sending the request; the server validates everything else. A LanceDB
Embedded refresh rejects every option other than `batch_size`, `gpu_class`
and `gpu_count`.
## When a refresh fails {#failures}
What a failure keeps depends on where the refresh ran. The local, blocking
+74
View File
@@ -647,6 +647,80 @@ describe("remote connection", () => {
]);
});
it("forwards refresh options to the backfill request", async () => {
const bodies: unknown[] = [];
await withMockDatabase(
(req, res) => {
const path = req.url ?? "";
if (path.endsWith("/describe/")) {
res.writeHead(200, { "Content-Type": "application/json" }).end(
JSON.stringify({
name: "docs",
version: 1,
schema: { fields: [] },
}),
);
return;
}
if (path === "/v1/table/docs/backfill_column") {
let body = "";
req.on("data", (chunk) => {
body += chunk;
});
req.on("end", () => {
bodies.push(JSON.parse(body));
res
.writeHead(202, { "Content-Type": "application/json" })
.end('{"job_id": "j-9"}');
});
return;
}
res.writeHead(404).end();
},
async (db) => {
const table = await db.openTable("docs");
const job = await table.refreshColumnAsync("embedding", {
where: "id % 2 = 0",
batchSize: 16,
gpuClass: "a10g",
gpuCount: 2,
errorPolicy: "skip",
maxFailedFraction: 0.01,
maxFailedRows: 50,
rowTimeoutSeconds: 30,
initTimeoutSeconds: 120,
maxExecutorProcessesPerWorker: 2,
});
expect(job.id).toBe("j-9");
await table.refreshColumnAsync("embedding");
// Invalid options reject before any request is sent.
for (const [options, message] of [
[{ maxFailedRows: -1 }, "maxFailedRows"],
[{ errorPolicy: "retry" }, "errorPolicy"],
[{ where: " " }, "where"],
[{ maxFailedFraction: 2 }, "max_failed_fraction"],
[{ rowTimeoutSeconds: 0 }, "row_timeout_seconds"],
[{ numWorkers: 0 }, "num_workers"],
[{ gpuCount: 0 }, "gpu_count"],
] as const) {
await expect(
table.refreshColumnAsync("embedding", options),
).rejects.toThrow(message);
}
},
);
expect(bodies).toEqual([
JSON.parse(
`{"column": "embedding", "where": "id % 2 = 0", "batch_size": 16,
"gpu_class": "a10g", "gpu_count": 2, "error_policy": "skip",
"max_failed_fraction": 0.01, "max_failed_rows": 50,
"row_timeout_seconds": 30, "init_timeout_seconds": 120,
"max_executor_processes_per_worker": 2}`,
),
{ column: "embedding" },
]);
});
it("surfaces JSON server errors from remote table operations", async () => {
await withMockDatabase(
(req, res) => {
+11
View File
@@ -5590,6 +5590,17 @@ describe("computed columns", () => {
expect(rows.map((r) => r.doubled).sort()).toEqual([2, 4]);
});
it("rejects Enterprise refresh options on local tables", async () => {
const db = await connect(tmpDir.name);
const table = await db.createTable("refresh_options_local", [{ x: 1 }]);
await expect(table.refreshColumn("x", { where: "x > 0" })).rejects.toThrow(
"where requires a LanceDB Enterprise refresh",
);
await expect(
table.refreshColumnAsync("x", { errorPolicy: "skip" }),
).rejects.toThrow("error_policy requires a LanceDB Enterprise refresh");
});
it("records Function errors only on remote tables", async () => {
const db = await connect(tmpDir.name);
const table = await db.createTable("errors_local", [{ x: 1 }]);
+1
View File
@@ -56,6 +56,7 @@ export {
MergeResult,
AddResult,
AddColumnsResult,
RefreshColumnOptions,
RefreshColumnResult,
FunctionErrors,
FunctionErrorsOptions,
+35 -7
View File
@@ -38,6 +38,7 @@ import {
Branches as NativeBranches,
OptimizeStats,
RTEStats,
RefreshColumnOptions,
RefreshColumnResult,
RefreshMaterializedViewResult,
TableStatistics,
@@ -671,10 +672,15 @@ export abstract class Table {
* everything else is left as it is. LanceDB Embedded only: in LanceDB
* Enterprise a refresh runs as a server job, through {@link Table#refreshColumnAsync}.
* @param {string} column The name of the computed column to fill.
* @param {RefreshColumnOptions} options Refresh options, all LanceDB
* Enterprise only; see {@link Table#refreshColumnAsync}.
* @returns {Promise<RefreshColumnResult>} A promise that resolves to the
* number of rows filled and the new version number of the table.
*/
abstract refreshColumn(column: string): Promise<RefreshColumnResult>;
abstract refreshColumn(
column: string,
options?: RefreshColumnOptions,
): Promise<RefreshColumnResult>;
/**
* Like {@link Table#refreshColumn}, but returns a handle to the refresh
@@ -689,14 +695,30 @@ export abstract class Table {
* than failing the job. In LanceDB Embedded the job runs in-process; in
* LanceDB Enterprise it is the server's backfill job.
* @param {string} column The name of the computed column to fill.
* @param {RefreshColumnOptions} options Refresh options, mirroring the SQL
* `REFRESH COLUMN ... WHERE ... WITH (...)` form: `where` narrows the rows,
* `errorPolicy`, `maxFailedFraction`, `maxFailedRows`, `rowTimeoutSeconds`
* and `initTimeoutSeconds` govern failing rows, and `batchSize`,
* `gpuClass`, `gpuCount`, `numWorkers` and `maxExecutorProcessesPerWorker`
* size the workers. All are LanceDB Enterprise only; a LanceDB Embedded
* refresh rejects every option other than `batchSize`, `gpuClass` and
* `gpuCount`.
* @example
* ```ts
* const job = await table.refreshColumnAsync("doubled");
* const job = await table.refreshColumnAsync("embedding", {
* where: "published_at > '2026-01-01'",
* errorPolicy: "skip",
* maxFailedFraction: 0.01,
* rowTimeoutSeconds: 30,
* });
* await job.wait();
* console.log(await job.status()); // "finished"
* ```
*/
abstract refreshColumnAsync(column: string): Promise<Job>;
abstract refreshColumnAsync(
column: string,
options?: RefreshColumnOptions,
): Promise<Job>;
/**
* The per-row errors Function refreshes recorded on this table.
@@ -1472,12 +1494,18 @@ export class LocalTable extends Table {
throw new Error("Invalid input type for addColumns");
}
async refreshColumn(column: string): Promise<RefreshColumnResult> {
return await this.inner.refreshColumn(column);
async refreshColumn(
column: string,
options?: RefreshColumnOptions,
): Promise<RefreshColumnResult> {
return await this.inner.refreshColumn(column, options);
}
async refreshColumnAsync(column: string): Promise<Job> {
return new Job(await this.inner.refreshColumnAsync(column));
async refreshColumnAsync(
column: string,
options?: RefreshColumnOptions,
): Promise<Job> {
return new Job(await this.inner.refreshColumnAsync(column, options));
}
async functionErrors(
+84 -4
View File
@@ -421,20 +421,30 @@ impl Table {
}
#[napi(catch_unwind)]
pub async fn refresh_column(&self, column: String) -> napi::Result<RefreshColumnResult> {
pub async fn refresh_column(
&self,
column: String,
options: Option<RefreshColumnOptions>,
) -> napi::Result<RefreshColumnResult> {
let options = options.unwrap_or_default().try_into()?;
let res = self
.inner_ref()?
.refresh_column(column)
.refresh_column_with_options(column, options)
.await
.default_error()?;
Ok(res.into())
}
#[napi(catch_unwind)]
pub async fn refresh_column_async(&self, column: String) -> napi::Result<crate::job::Job> {
pub async fn refresh_column_async(
&self,
column: String,
options: Option<RefreshColumnOptions>,
) -> napi::Result<crate::job::Job> {
let options = options.unwrap_or_default().try_into()?;
let job = self
.inner_ref()?
.refresh_column_async(column)
.refresh_column_async_with_options(column, options)
.await
.default_error()?;
Ok(crate::job::Job::new(job))
@@ -1515,6 +1525,76 @@ pub struct FunctionErrorsOptions {
pub limit: Option<i64>,
}
/// Options for `Table.refreshColumn` and `Table.refreshColumnAsync`. Every
/// option is LanceDB Enterprise only; a LanceDB Embedded refresh ignores
/// `batchSize`, `gpuClass` and `gpuCount` and rejects the rest.
#[napi(object)]
#[derive(Clone, Debug, Default)]
pub struct RefreshColumnOptions {
/// Only refresh the rows matching this SQL filter.
#[napi(js_name = "where")]
pub where_clause: Option<String>,
/// Maximum rows per Function call; must be greater than zero.
pub batch_size: Option<u32>,
/// GPU class for Function workers.
pub gpu_class: Option<String>,
/// Positive aggregate GPU budget for Function workers.
pub gpu_count: Option<u32>,
/// `"fail"` (default) stops on a row error; `"skip"` writes null and
/// records the error, subject to the failure limits.
pub error_policy: Option<String>,
/// Under a skip policy, fail the job once more than this share of rows
/// (0 to 1) has failed.
pub max_failed_fraction: Option<f64>,
/// Under a skip policy, fail the job once more than this many rows have
/// failed.
pub max_failed_rows: Option<i64>,
/// Deadline for one Function call, in seconds.
pub row_timeout_seconds: Option<f64>,
/// Deadline for Function initialization, in seconds.
pub init_timeout_seconds: Option<f64>,
/// Worker count for an expression column; Function columns reject it.
pub num_workers: Option<u32>,
/// Cap on executor processes per Function worker.
pub max_executor_processes_per_worker: Option<u32>,
}
impl TryFrom<RefreshColumnOptions> for lancedb::table::RefreshColumnOptions {
type Error = napi::Error;
fn try_from(value: RefreshColumnOptions) -> napi::Result<Self> {
let mut options = Self::default();
options.where_clause = value.where_clause;
options.batch_size = value.batch_size.map(|size| size as usize);
options.gpu_class = value.gpu_class;
options.gpu_count = value.gpu_count;
options.error_policy = match value.error_policy.as_deref() {
None => None,
Some("fail") => Some(lancedb::function::FunctionErrorPolicy::Fail),
Some("skip") => Some(lancedb::function::FunctionErrorPolicy::Skip),
Some(other) => {
return Err(napi::Error::from_reason(format!(
"errorPolicy must be \"fail\" or \"skip\", got \"{other}\""
)));
}
};
options.max_failed_fraction = value.max_failed_fraction;
options.max_failed_rows = value
.max_failed_rows
.map(|rows| {
u64::try_from(rows).map_err(|_| {
napi::Error::from_reason("maxFailedRows must be a non-negative integer")
})
})
.transpose()?;
options.row_timeout_seconds = value.row_timeout_seconds;
options.init_timeout_seconds = value.init_timeout_seconds;
options.num_workers = value.num_workers.map(|workers| workers as usize);
options.max_executor_processes_per_worker = value.max_executor_processes_per_worker;
Ok(options)
}
}
/// One row a Function refresh skipped, as the server recorded it.
#[napi(object)]
#[derive(Clone, Debug)]
+14
View File
@@ -617,21 +617,35 @@ class Table:
self,
column: str,
*,
where: Optional[str] = None,
batch_size: Optional[int] = None,
gpu_class: Optional[str] = None,
gpu_count: Optional[int] = None,
retry_policy: Optional[str] = None,
error_policy: Optional[str] = None,
max_failed_fraction: Optional[float] = None,
max_failed_rows: Optional[int] = None,
row_timeout_seconds: Optional[float] = None,
init_timeout_seconds: Optional[float] = None,
num_workers: Optional[int] = None,
max_executor_processes_per_worker: Optional[int] = None,
) -> RefreshColumnResult: ...
async def refresh_column_async(
self,
column: str,
*,
where: Optional[str] = None,
batch_size: Optional[int] = None,
gpu_class: Optional[str] = None,
gpu_count: Optional[int] = None,
retry_policy: Optional[str] = None,
error_policy: Optional[str] = None,
max_failed_fraction: Optional[float] = None,
max_failed_rows: Optional[int] = None,
row_timeout_seconds: Optional[float] = None,
init_timeout_seconds: Optional[float] = None,
num_workers: Optional[int] = None,
max_executor_processes_per_worker: Optional[int] = None,
) -> Job: ...
async def function_errors(
self,
+28
View File
@@ -1138,20 +1138,34 @@ class RemoteTable(Table):
self,
column: str,
*,
where: Optional[str] = None,
batch_size: Optional[int] = None,
gpu_class: Optional[str] = None,
gpu_count: Optional[int] = None,
retry_policy: Optional[FunctionRetryPolicy] = None,
error_policy: Optional[Literal["fail", "skip"]] = None,
max_failed_fraction: Optional[float] = None,
max_failed_rows: Optional[int] = None,
row_timeout_seconds: Optional[float] = None,
init_timeout_seconds: Optional[float] = None,
num_workers: Optional[int] = None,
max_executor_processes_per_worker: Optional[int] = None,
):
return LOOP.run(
self._table.refresh_column(
column,
where=where,
batch_size=batch_size,
gpu_class=gpu_class,
gpu_count=gpu_count,
retry_policy=retry_policy,
error_policy=error_policy,
max_failed_fraction=max_failed_fraction,
max_failed_rows=max_failed_rows,
row_timeout_seconds=row_timeout_seconds,
init_timeout_seconds=init_timeout_seconds,
num_workers=num_workers,
max_executor_processes_per_worker=max_executor_processes_per_worker,
)
)
@@ -1159,21 +1173,35 @@ class RemoteTable(Table):
self,
column: str,
*,
where: Optional[str] = None,
batch_size: Optional[int] = None,
gpu_class: Optional[str] = None,
gpu_count: Optional[int] = None,
retry_policy: Optional[FunctionRetryPolicy] = None,
error_policy: Optional[Literal["fail", "skip"]] = None,
max_failed_fraction: Optional[float] = None,
max_failed_rows: Optional[int] = None,
row_timeout_seconds: Optional[float] = None,
init_timeout_seconds: Optional[float] = None,
num_workers: Optional[int] = None,
max_executor_processes_per_worker: Optional[int] = None,
) -> Job[RefreshColumnResult]:
return Job(
LOOP.run(
self._table.refresh_column_async(
column,
where=where,
batch_size=batch_size,
gpu_class=gpu_class,
gpu_count=gpu_count,
retry_policy=retry_policy,
error_policy=error_policy,
max_failed_fraction=max_failed_fraction,
max_failed_rows=max_failed_rows,
row_timeout_seconds=row_timeout_seconds,
init_timeout_seconds=init_timeout_seconds,
num_workers=num_workers,
max_executor_processes_per_worker=max_executor_processes_per_worker,
)
)
)
+150
View File
@@ -3164,11 +3164,18 @@ class Table(ABC):
self,
column: str,
*,
where: Optional[str] = None,
batch_size: Optional[int] = None,
gpu_class: Optional[str] = None,
gpu_count: Optional[int] = None,
retry_policy: Optional[FunctionRetryPolicy] = None,
error_policy: Optional[Literal["fail", "skip"]] = None,
max_failed_fraction: Optional[float] = None,
max_failed_rows: Optional[int] = None,
row_timeout_seconds: Optional[float] = None,
init_timeout_seconds: Optional[float] = None,
num_workers: Optional[int] = None,
max_executor_processes_per_worker: Optional[int] = None,
) -> "RefreshColumnResult":
"""
Fill the rows of a computed column that hold no value yet.
@@ -3189,6 +3196,10 @@ class Table(ABC):
----------
column: str
The name of the computed column to fill.
where: str, optional
SQL filter selecting the rows to refresh, as in ``REFRESH COLUMN
... WHERE``. Unfilled rows outside it stay unfilled. LanceDB
Enterprise only; local refreshes reject this option.
batch_size: int, optional
Maximum rows per Function call on the remote server. Must be
greater than zero. None uses the server default. Has no effect on
@@ -3209,6 +3220,22 @@ class Table(ABC):
error_policy: {"fail", "skip"}, optional
Action after failure or retry exhaustion. Skip records the error
and attempt count, subject to the server's failure limits.
max_failed_fraction: float, optional
Under a skip policy, fail the job once more than this share of
rows (0 to 1) has failed. LanceDB Enterprise only.
max_failed_rows: int, optional
Under a skip policy, fail the job once more than this many rows
have failed. LanceDB Enterprise only.
row_timeout_seconds: float, optional
Deadline for one Function call. LanceDB Enterprise only.
init_timeout_seconds: float, optional
Deadline for Function initialization. LanceDB Enterprise only.
num_workers: int, optional
Worker count for an expression column; registered Function
columns reject it. LanceDB Enterprise only.
max_executor_processes_per_worker: int, optional
Cap on executor processes per Function worker. LanceDB
Enterprise only.
Returns
-------
@@ -3222,11 +3249,18 @@ class Table(ABC):
self,
column: str,
*,
where: Optional[str] = None,
batch_size: Optional[int] = None,
gpu_class: Optional[str] = None,
gpu_count: Optional[int] = None,
retry_policy: Optional[FunctionRetryPolicy] = None,
error_policy: Optional[Literal["fail", "skip"]] = None,
max_failed_fraction: Optional[float] = None,
max_failed_rows: Optional[int] = None,
row_timeout_seconds: Optional[float] = None,
init_timeout_seconds: Optional[float] = None,
num_workers: Optional[int] = None,
max_executor_processes_per_worker: Optional[int] = None,
) -> Job[RefreshColumnJobResult]:
"""
Like :meth:`refresh_column`, but returns a handle to the refresh job
@@ -3245,6 +3279,10 @@ class Table(ABC):
----------
column: str
The name of the computed column to fill.
where: str, optional
SQL filter selecting the rows to refresh, as in ``REFRESH COLUMN
... WHERE``. Unfilled rows outside it stay unfilled. LanceDB
Enterprise only; local refreshes reject this option.
batch_size: int, optional
Maximum rows per Function call on the remote server. Must be
greater than zero. None uses the server default. Lower this for
@@ -3266,6 +3304,22 @@ class Table(ABC):
error_policy: {"fail", "skip"}, optional
Action after failure or retry exhaustion. Skip records the error
and attempt count, subject to the server's failure limits.
max_failed_fraction: float, optional
Under a skip policy, fail the job once more than this share of
rows (0 to 1) has failed. LanceDB Enterprise only.
max_failed_rows: int, optional
Under a skip policy, fail the job once more than this many rows
have failed. LanceDB Enterprise only.
row_timeout_seconds: float, optional
Deadline for one Function call. LanceDB Enterprise only.
init_timeout_seconds: float, optional
Deadline for Function initialization. LanceDB Enterprise only.
num_workers: int, optional
Worker count for an expression column; registered Function
columns reject it. LanceDB Enterprise only.
max_executor_processes_per_worker: int, optional
Cap on executor processes per Function worker. LanceDB
Enterprise only.
Returns
-------
@@ -5374,11 +5428,18 @@ class LanceTable(Table):
self,
column: str,
*,
where: Optional[str] = None,
batch_size: Optional[int] = None,
gpu_class: Optional[str] = None,
gpu_count: Optional[int] = None,
retry_policy: Optional[FunctionRetryPolicy] = None,
error_policy: Optional[Literal["fail", "skip"]] = None,
max_failed_fraction: Optional[float] = None,
max_failed_rows: Optional[int] = None,
row_timeout_seconds: Optional[float] = None,
init_timeout_seconds: Optional[float] = None,
num_workers: Optional[int] = None,
max_executor_processes_per_worker: Optional[int] = None,
) -> "RefreshColumnResult":
"""Fill a computed column's unfilled rows and recompute those whose
inputs changed. See
@@ -5386,11 +5447,18 @@ class LanceTable(Table):
return LOOP.run(
self._table.refresh_column(
column,
where=where,
batch_size=batch_size,
gpu_class=gpu_class,
gpu_count=gpu_count,
retry_policy=retry_policy,
error_policy=error_policy,
max_failed_fraction=max_failed_fraction,
max_failed_rows=max_failed_rows,
row_timeout_seconds=row_timeout_seconds,
init_timeout_seconds=init_timeout_seconds,
num_workers=num_workers,
max_executor_processes_per_worker=max_executor_processes_per_worker,
)
)
@@ -5398,11 +5466,18 @@ class LanceTable(Table):
self,
column: str,
*,
where: Optional[str] = None,
batch_size: Optional[int] = None,
gpu_class: Optional[str] = None,
gpu_count: Optional[int] = None,
retry_policy: Optional[FunctionRetryPolicy] = None,
error_policy: Optional[Literal["fail", "skip"]] = None,
max_failed_fraction: Optional[float] = None,
max_failed_rows: Optional[int] = None,
row_timeout_seconds: Optional[float] = None,
init_timeout_seconds: Optional[float] = None,
num_workers: Optional[int] = None,
max_executor_processes_per_worker: Optional[int] = None,
) -> Job[RefreshColumnJobResult]:
"""Fill a computed column's unfilled rows and recompute those whose
inputs changed, returning a handle to the refresh job. See
@@ -5412,11 +5487,18 @@ class LanceTable(Table):
LOOP.run(
self._table.refresh_column_async(
column,
where=where,
batch_size=batch_size,
gpu_class=gpu_class,
gpu_count=gpu_count,
retry_policy=retry_policy,
error_policy=error_policy,
max_failed_fraction=max_failed_fraction,
max_failed_rows=max_failed_rows,
row_timeout_seconds=row_timeout_seconds,
init_timeout_seconds=init_timeout_seconds,
num_workers=num_workers,
max_executor_processes_per_worker=max_executor_processes_per_worker,
)
)
)
@@ -7769,11 +7851,18 @@ class AsyncTable:
self,
column: str,
*,
where: Optional[str] = None,
batch_size: Optional[int] = None,
gpu_class: Optional[str] = None,
gpu_count: Optional[int] = None,
retry_policy: Optional[FunctionRetryPolicy] = None,
error_policy: Optional[Literal["fail", "skip"]] = None,
max_failed_fraction: Optional[float] = None,
max_failed_rows: Optional[int] = None,
row_timeout_seconds: Optional[float] = None,
init_timeout_seconds: Optional[float] = None,
num_workers: Optional[int] = None,
max_executor_processes_per_worker: Optional[int] = None,
) -> RefreshColumnResult:
"""
Fill the rows of a computed column that hold no value yet.
@@ -7794,6 +7883,10 @@ class AsyncTable:
----------
column: str
The name of the computed column to fill.
where: str, optional
SQL filter selecting the rows to refresh, as in ``REFRESH COLUMN
... WHERE``. Unfilled rows outside it stay unfilled. LanceDB
Enterprise only; local refreshes reject this option.
batch_size: int, optional
Maximum rows per Function call on the remote server. Must be
greater than zero. None uses the server default. Has no effect on
@@ -7814,6 +7907,22 @@ class AsyncTable:
error_policy: {"fail", "skip"}, optional
Action after failure or retry exhaustion. Skip records the error
and attempt count, subject to the server's failure limits.
max_failed_fraction: float, optional
Under a skip policy, fail the job once more than this share of
rows (0 to 1) has failed. LanceDB Enterprise only.
max_failed_rows: int, optional
Under a skip policy, fail the job once more than this many rows
have failed. LanceDB Enterprise only.
row_timeout_seconds: float, optional
Deadline for one Function call. LanceDB Enterprise only.
init_timeout_seconds: float, optional
Deadline for Function initialization. LanceDB Enterprise only.
num_workers: int, optional
Worker count for an expression column; registered Function
columns reject it. LanceDB Enterprise only.
max_executor_processes_per_worker: int, optional
Cap on executor processes per Function worker. LanceDB
Enterprise only.
Returns
-------
@@ -7822,6 +7931,7 @@ class AsyncTable:
"""
return await self._inner.refresh_column(
column,
where=where,
batch_size=batch_size,
gpu_class=gpu_class,
gpu_count=gpu_count,
@@ -7829,6 +7939,12 @@ class AsyncTable:
if retry_policy is None
else FunctionRetryPolicy.model_validate(retry_policy).model_dump_json(),
error_policy=error_policy,
max_failed_fraction=max_failed_fraction,
max_failed_rows=max_failed_rows,
row_timeout_seconds=row_timeout_seconds,
init_timeout_seconds=init_timeout_seconds,
num_workers=num_workers,
max_executor_processes_per_worker=max_executor_processes_per_worker,
)
async def function_errors(
@@ -7924,11 +8040,18 @@ class AsyncTable:
self,
column: str,
*,
where: Optional[str] = None,
batch_size: Optional[int] = None,
gpu_class: Optional[str] = None,
gpu_count: Optional[int] = None,
retry_policy: Optional[FunctionRetryPolicy] = None,
error_policy: Optional[Literal["fail", "skip"]] = None,
max_failed_fraction: Optional[float] = None,
max_failed_rows: Optional[int] = None,
row_timeout_seconds: Optional[float] = None,
init_timeout_seconds: Optional[float] = None,
num_workers: Optional[int] = None,
max_executor_processes_per_worker: Optional[int] = None,
) -> AsyncJob[RefreshColumnJobResult]:
"""
Like :meth:`refresh_column`, but returns a handle to the refresh job
@@ -7947,6 +8070,10 @@ class AsyncTable:
----------
column: str
The name of the computed column to fill.
where: str, optional
SQL filter selecting the rows to refresh, as in ``REFRESH COLUMN
... WHERE``. Unfilled rows outside it stay unfilled. LanceDB
Enterprise only; local refreshes reject this option.
batch_size: int, optional
Maximum rows per Function call on the remote server. Must be
greater than zero. None uses the server default. Lower this for
@@ -7968,6 +8095,22 @@ class AsyncTable:
error_policy: {"fail", "skip"}, optional
Action after failure or retry exhaustion. Skip records the error
and attempt count, subject to the server's failure limits.
max_failed_fraction: float, optional
Under a skip policy, fail the job once more than this share of
rows (0 to 1) has failed. LanceDB Enterprise only.
max_failed_rows: int, optional
Under a skip policy, fail the job once more than this many rows
have failed. LanceDB Enterprise only.
row_timeout_seconds: float, optional
Deadline for one Function call. LanceDB Enterprise only.
init_timeout_seconds: float, optional
Deadline for Function initialization. LanceDB Enterprise only.
num_workers: int, optional
Worker count for an expression column; registered Function
columns reject it. LanceDB Enterprise only.
max_executor_processes_per_worker: int, optional
Cap on executor processes per Function worker. LanceDB
Enterprise only.
Returns
-------
@@ -7993,6 +8136,7 @@ class AsyncTable:
return _typed_job(
await self._inner.refresh_column_async(
column,
where=where,
batch_size=batch_size,
gpu_class=gpu_class,
gpu_count=gpu_count,
@@ -8000,6 +8144,12 @@ class AsyncTable:
if retry_policy is None
else FunctionRetryPolicy.model_validate(retry_policy).model_dump_json(),
error_policy=error_policy,
max_failed_fraction=max_failed_fraction,
max_failed_rows=max_failed_rows,
row_timeout_seconds=row_timeout_seconds,
init_timeout_seconds=init_timeout_seconds,
num_workers=num_workers,
max_executor_processes_per_worker=max_executor_processes_per_worker,
),
RefreshColumnJobResult.from_json,
)
+31
View File
@@ -1868,6 +1868,16 @@ def test_remote_create_index_async_returns_job():
"error_policy": "skip",
},
{"retry_policy": {"max_attempts": 1}, "error_policy": "fail"},
{
"where": "id % 2 = 0",
"error_policy": "skip",
"max_failed_fraction": 0.01,
"max_failed_rows": 50,
"row_timeout_seconds": 30,
"init_timeout_seconds": 120,
"num_workers": 4,
"max_executor_processes_per_worker": 2,
},
],
)
async def test_remote_refresh_async_returns_typed_terminal_result(
@@ -1898,6 +1908,17 @@ async def test_remote_refresh_async_returns_typed_terminal_result(
).model_dump(mode="json")
if refresh_kwargs.get("error_policy") is not None:
expected["error_policy"] = refresh_kwargs["error_policy"]
for key in (
"where",
"max_failed_fraction",
"max_failed_rows",
"row_timeout_seconds",
"init_timeout_seconds",
"num_workers",
"max_executor_processes_per_worker",
):
if refresh_kwargs.get(key) is not None:
expected[key] = refresh_kwargs[key]
assert json.loads(body) == expected
request.send_response(202)
request.send_header("Content-Type", "application/json")
@@ -2061,6 +2082,16 @@ def test_remote_refresh_async_forwards_progress_from_submitted_handle():
({"gpu_count": -1}, "gpu_count"),
({"retry_policy": {"transient": True, "max_attempts": 0}}, "retry_policy"),
({"error_policy": "retry"}, "error_policy"),
({"where": " "}, "where"),
({"max_failed_fraction": 1.5}, "max_failed_fraction"),
({"max_failed_rows": -1}, "max_failed_rows"),
({"row_timeout_seconds": 0}, "row_timeout_seconds"),
({"init_timeout_seconds": -5}, "init_timeout_seconds"),
({"num_workers": 0}, "num_workers"),
(
{"max_executor_processes_per_worker": -2},
"max_executor_processes_per_worker",
),
],
)
@pytest.mark.asyncio
+21 -4
View File
@@ -5985,11 +5985,28 @@ def test_refresh_column_rejects_invalid_batch_size(tmp_path, method, batch_size)
@pytest.mark.parametrize("method", ["refresh_column", "refresh_column_async"])
def test_local_refresh_rejects_function_retry_policy(tmp_path, method):
@pytest.mark.parametrize(
"refresh_kwargs,option",
[
({"retry_policy": lancedb.FunctionRetryPolicy(transient=True)}, "retry_policy"),
({"where": "x > 0"}, "where"),
({"error_policy": "skip"}, "error_policy"),
({"max_failed_fraction": 0.5}, "max_failed_fraction"),
({"max_failed_rows": 10}, "max_failed_rows"),
({"row_timeout_seconds": 30}, "row_timeout_seconds"),
({"init_timeout_seconds": 60}, "init_timeout_seconds"),
({"num_workers": 2}, "num_workers"),
({"max_executor_processes_per_worker": 2}, "max_executor_processes_per_worker"),
],
)
def test_local_refresh_rejects_enterprise_options(
tmp_path, method, refresh_kwargs, option
):
table = lancedb.connect(tmp_path).create_table("rows", [{"x": 1}])
policy = lancedb.FunctionRetryPolicy(transient=True)
with pytest.raises(NotImplementedError, match="Enterprise Function refresh"):
getattr(table, method)("x", retry_policy=policy)
with pytest.raises(
NotImplementedError, match=f"{option} requires a LanceDB Enterprise refresh"
):
getattr(table, method)("x", **refresh_kwargs)
@pytest.mark.asyncio
+96 -8
View File
@@ -42,14 +42,62 @@ use pyo3::{
mod scannable;
fn refresh_column_options(
/// Keyword arguments shared by `refresh_column` and `refresh_column_async`.
struct RefreshColumnArgs {
where_clause: Option<String>,
batch_size: Option<i64>,
gpu_class: Option<String>,
gpu_count: Option<i64>,
retry_policy: Option<String>,
error_policy: Option<String>,
) -> PyResult<RefreshColumnOptions> {
max_failed_fraction: Option<f64>,
max_failed_rows: Option<i64>,
row_timeout_seconds: Option<f64>,
init_timeout_seconds: Option<f64>,
num_workers: Option<i64>,
max_executor_processes_per_worker: Option<i64>,
}
fn positive_count<T: TryFrom<i64>>(value: Option<i64>, name: &str) -> PyResult<Option<T>> {
value
.map(|v| {
T::try_from(v)
.ok()
.filter(|_| v > 0)
.ok_or_else(|| PyValueError::new_err(format!("{name} must be greater than zero")))
})
.transpose()
}
fn refresh_column_options(args: RefreshColumnArgs) -> PyResult<RefreshColumnOptions> {
let RefreshColumnArgs {
where_clause,
batch_size,
gpu_class,
gpu_count,
retry_policy,
error_policy,
max_failed_fraction,
max_failed_rows,
row_timeout_seconds,
init_timeout_seconds,
num_workers,
max_executor_processes_per_worker,
} = args;
let mut options = RefreshColumnOptions::default();
options.where_clause = where_clause;
options.max_failed_fraction = max_failed_fraction;
options.max_failed_rows = max_failed_rows
.map(u64::try_from)
.transpose()
.map_err(|_| PyValueError::new_err("max_failed_rows must be a non-negative integer"))?;
options.row_timeout_seconds = row_timeout_seconds;
options.init_timeout_seconds = init_timeout_seconds;
options.num_workers = positive_count(num_workers, "num_workers")?;
options.max_executor_processes_per_worker = positive_count(
max_executor_processes_per_worker,
"max_executor_processes_per_worker",
)?;
options.retry_policy = retry_policy
.map(|json| {
serde_json::from_str(&json)
@@ -2142,18 +2190,38 @@ impl Table {
})
}
#[pyo3(signature = (column, *, batch_size=None, gpu_class=None, gpu_count=None, retry_policy=None, error_policy=None))]
#[pyo3(signature = (column, *, r#where=None, batch_size=None, gpu_class=None, gpu_count=None, retry_policy=None, error_policy=None, max_failed_fraction=None, max_failed_rows=None, row_timeout_seconds=None, init_timeout_seconds=None, num_workers=None, max_executor_processes_per_worker=None))]
#[allow(clippy::too_many_arguments)]
pub fn refresh_column(
self_: PyRef<'_, Self>,
column: String,
r#where: Option<String>,
batch_size: Option<i64>,
gpu_class: Option<String>,
gpu_count: Option<i64>,
retry_policy: Option<String>,
error_policy: Option<String>,
max_failed_fraction: Option<f64>,
max_failed_rows: Option<i64>,
row_timeout_seconds: Option<f64>,
init_timeout_seconds: Option<f64>,
num_workers: Option<i64>,
max_executor_processes_per_worker: Option<i64>,
) -> PyResult<Bound<'_, PyAny>> {
let options =
refresh_column_options(batch_size, gpu_class, gpu_count, retry_policy, error_policy)?;
let options = refresh_column_options(RefreshColumnArgs {
where_clause: r#where,
batch_size,
gpu_class,
gpu_count,
retry_policy,
error_policy,
max_failed_fraction,
max_failed_rows,
row_timeout_seconds,
init_timeout_seconds,
num_workers,
max_executor_processes_per_worker,
})?;
let inner = self_.inner_ref()?.clone();
future_into_py(self_.py(), async move {
let result = inner
@@ -2164,18 +2232,38 @@ impl Table {
})
}
#[pyo3(signature = (column, *, batch_size=None, gpu_class=None, gpu_count=None, retry_policy=None, error_policy=None))]
#[pyo3(signature = (column, *, r#where=None, batch_size=None, gpu_class=None, gpu_count=None, retry_policy=None, error_policy=None, max_failed_fraction=None, max_failed_rows=None, row_timeout_seconds=None, init_timeout_seconds=None, num_workers=None, max_executor_processes_per_worker=None))]
#[allow(clippy::too_many_arguments)]
pub fn refresh_column_async(
self_: PyRef<'_, Self>,
column: String,
r#where: Option<String>,
batch_size: Option<i64>,
gpu_class: Option<String>,
gpu_count: Option<i64>,
retry_policy: Option<String>,
error_policy: Option<String>,
max_failed_fraction: Option<f64>,
max_failed_rows: Option<i64>,
row_timeout_seconds: Option<f64>,
init_timeout_seconds: Option<f64>,
num_workers: Option<i64>,
max_executor_processes_per_worker: Option<i64>,
) -> PyResult<Bound<'_, PyAny>> {
let options =
refresh_column_options(batch_size, gpu_class, gpu_count, retry_policy, error_policy)?;
let options = refresh_column_options(RefreshColumnArgs {
where_clause: r#where,
batch_size,
gpu_class,
gpu_count,
retry_policy,
error_policy,
max_failed_fraction,
max_failed_rows,
row_timeout_seconds,
init_timeout_seconds,
num_workers,
max_executor_processes_per_worker,
})?;
let inner = self_.inner_ref()?.clone();
future_into_py(self_.py(), async move {
let job = inner
+91 -16
View File
@@ -3969,22 +3969,10 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
) -> Result<Job<crate::function::RefreshColumnResult>> {
options.validate()?;
self.check_mutable().await?;
let mut body = serde_json::json!({ "column": column });
if let Some(batch_size) = options.batch_size {
body["batch_size"] = serde_json::json!(batch_size);
}
if let Some(gpu_class) = options.gpu_class {
body["gpu_class"] = serde_json::json!(gpu_class);
}
if let Some(gpu_count) = options.gpu_count {
body["gpu_count"] = serde_json::json!(gpu_count);
}
if let Some(policy) = options.retry_policy {
body["retry_policy"] = serde_json::json!(policy);
}
if let Some(policy) = options.error_policy {
body["error_policy"] = serde_json::json!(policy);
}
let mut body = serde_json::to_value(&options).map_err(|e| Error::Runtime {
message: format!("failed to serialize refresh options: {e}"),
})?;
body["column"] = serde_json::json!(column);
self.apply_branch_body(&mut body);
let request = self
.client
@@ -11021,6 +11009,93 @@ mod tests {
);
}
#[tokio::test]
async fn test_refresh_column_async_filter_and_failure_options() {
use crate::function::FunctionErrorPolicy;
use crate::table::RefreshColumnOptions;
let table = Table::new_with_handler("my_table", |request| {
assert_eq!(request.url().path(), "/v1/table/my_table/backfill_column");
assert_eq!(
request_body_json(&request),
serde_json::json!({
"column": "quality",
"where": "id % 2 = 0",
"error_policy": "skip",
"max_failed_fraction": 0.01,
"max_failed_rows": 50,
"row_timeout_seconds": 30.0,
"init_timeout_seconds": 120.0,
"num_workers": 4,
"max_executor_processes_per_worker": 2,
})
);
http::Response::builder()
.status(202)
.body(r#"{"job_id": "j-filtered"}"#)
.unwrap()
});
let options = RefreshColumnOptions::default()
.where_clause("id % 2 = 0")
.error_policy(FunctionErrorPolicy::Skip)
.max_failed_fraction(0.01)
.max_failed_rows(50)
.row_timeout_seconds(30.0)
.init_timeout_seconds(120.0)
.num_workers(4)
.max_executor_processes_per_worker(2);
let job = table
.refresh_column_async_with_options("quality", options)
.await
.unwrap();
assert_eq!(job.id(), Some("j-filtered"));
}
#[tokio::test]
async fn test_refresh_column_rejects_invalid_filter_and_limit_options_before_request() {
use crate::table::RefreshColumnOptions;
let table = Table::new_with_handler("my_table", |_| -> http::Response<String> {
panic!("invalid refresh options must not send a request")
});
for (options, expected) in [
(RefreshColumnOptions::default().where_clause(" "), "where"),
(
RefreshColumnOptions::default().max_failed_fraction(1.5),
"max_failed_fraction",
),
(
RefreshColumnOptions::default().max_failed_fraction(f64::NAN),
"max_failed_fraction",
),
(
RefreshColumnOptions::default().row_timeout_seconds(0.0),
"row_timeout_seconds",
),
(
RefreshColumnOptions::default().init_timeout_seconds(f64::INFINITY),
"init_timeout_seconds",
),
(
RefreshColumnOptions::default().num_workers(0),
"num_workers",
),
(
RefreshColumnOptions::default().max_executor_processes_per_worker(0),
"max_executor_processes_per_worker",
),
] {
let error = table
.refresh_column_async_with_options("quality", options)
.await
.unwrap_err();
assert!(
matches!(&error, Error::InvalidInput { message } if message.contains(expected)),
"{error:?}"
);
}
}
#[tokio::test]
async fn test_refresh_column_rejects_invalid_gpu_options_before_request() {
use crate::table::RefreshColumnOptions;
+6 -8
View File
@@ -883,18 +883,17 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
message: "this operation is only supported on LanceDB Embedded".into(),
})
}
/// Refresh with options; expression-backed local refreshes ignore the
/// Function batch size.
/// Refresh with options. A local refresh ignores the Function batch size
/// and GPU placement and rejects every other option.
async fn refresh_column_with_options(
&self,
column: &str,
options: RefreshColumnOptions,
) -> Result<RefreshColumnResult> {
options.validate()?;
if options.retry_policy.is_some() || options.error_policy.is_some() {
if let Some(name) = options.enterprise_only_option() {
return Err(Error::NotSupported {
message: "Function retry and error policies require an Enterprise Function refresh"
.into(),
message: format!("{name} requires a LanceDB Enterprise refresh"),
});
}
self.refresh_column(column).await
@@ -916,10 +915,9 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
options: RefreshColumnOptions,
) -> Result<Job<crate::function::RefreshColumnResult>> {
options.validate()?;
if options.retry_policy.is_some() || options.error_policy.is_some() {
if let Some(name) = options.enterprise_only_option() {
return Err(Error::NotSupported {
message: "Function retry and error policies require an Enterprise Function refresh"
.into(),
message: format!("{name} requires a LanceDB Enterprise refresh"),
});
}
self.refresh_column_async(column).await
+134 -1
View File
@@ -56,31 +56,107 @@ use crate::job::Job;
use crate::{Error, Result};
/// Options for refreshing a computed column.
#[derive(Debug, Clone, Default)]
///
/// Serializes to the Enterprise backfill request body, one key per set
/// option; the SQL `REFRESH COLUMN ... WHERE ... WITH (...)` form takes the
/// same names.
#[derive(Debug, Clone, Default, Serialize)]
#[non_exhaustive]
pub struct RefreshColumnOptions {
/// Only refresh the rows matching this SQL filter. Enterprise only.
#[serde(rename = "where", skip_serializing_if = "Option::is_none")]
pub where_clause: Option<String>,
/// Override the Function's default user-code retry policy. Enterprise only.
#[serde(skip_serializing_if = "Option::is_none")]
pub retry_policy: Option<crate::function::FunctionRetryPolicy>,
/// Action after user-code failure or retry exhaustion. Enterprise only.
#[serde(skip_serializing_if = "Option::is_none")]
pub error_policy: Option<crate::function::FunctionErrorPolicy>,
/// Under a skip policy, fail the job once more than this share of rows
/// has failed, 0 to 1. Enterprise only.
#[serde(skip_serializing_if = "Option::is_none")]
pub max_failed_fraction: Option<f64>,
/// Under a skip policy, fail the job once more than this many rows have
/// failed. Enterprise only.
#[serde(skip_serializing_if = "Option::is_none")]
pub max_failed_rows: Option<u64>,
/// Deadline for one Function call, in seconds. Enterprise only.
#[serde(skip_serializing_if = "Option::is_none")]
pub row_timeout_seconds: Option<f64>,
/// Deadline for Function initialization, in seconds. Enterprise only.
#[serde(skip_serializing_if = "Option::is_none")]
pub init_timeout_seconds: Option<f64>,
/// Worker count for an expression column in LanceDB Enterprise;
/// registered Function columns reject it.
#[serde(skip_serializing_if = "Option::is_none")]
pub num_workers: Option<usize>,
/// At most this many executor processes per Function worker. Enterprise
/// only.
#[serde(skip_serializing_if = "Option::is_none")]
pub max_executor_processes_per_worker: Option<u32>,
/// Maximum rows per Function call in LanceDB Enterprise.
///
/// Must be greater than zero. `None` uses the server default. This has no
/// effect on expression-backed columns in LanceDB Embedded.
#[serde(skip_serializing_if = "Option::is_none")]
pub batch_size: Option<usize>,
/// GPU class for Function workers in LanceDB Enterprise.
///
/// `None` keeps the Function's registered class, or the deployment default
/// when the Function does not declare one.
#[serde(skip_serializing_if = "Option::is_none")]
pub gpu_class: Option<String>,
/// Positive aggregate GPU budget for Function workers in LanceDB
/// Enterprise.
///
/// `None` leaves the server's configured budget unchanged.
#[serde(skip_serializing_if = "Option::is_none")]
pub gpu_count: Option<u32>,
}
impl RefreshColumnOptions {
/// Only refresh the rows matching `filter`.
pub fn where_clause(mut self, filter: impl Into<String>) -> Self {
self.where_clause = Some(filter.into());
self
}
/// Fail the job once more than `fraction` of its rows has failed.
pub fn max_failed_fraction(mut self, fraction: f64) -> Self {
self.max_failed_fraction = Some(fraction);
self
}
/// Fail the job once more than `rows` rows have failed.
pub fn max_failed_rows(mut self, rows: u64) -> Self {
self.max_failed_rows = Some(rows);
self
}
/// Give each Function call at most `seconds` to return.
pub fn row_timeout_seconds(mut self, seconds: f64) -> Self {
self.row_timeout_seconds = Some(seconds);
self
}
/// Give Function initialization at most `seconds` to complete.
pub fn init_timeout_seconds(mut self, seconds: f64) -> Self {
self.init_timeout_seconds = Some(seconds);
self
}
/// Request `num_workers` workers for an expression column refresh.
pub fn num_workers(mut self, num_workers: usize) -> Self {
self.num_workers = Some(num_workers);
self
}
/// Cap executor processes per Function worker at `processes`.
pub fn max_executor_processes_per_worker(mut self, processes: u32) -> Self {
self.max_executor_processes_per_worker = Some(processes);
self
}
/// Override retries of failed Function calls for this refresh.
pub fn retry_policy(mut self, policy: crate::function::FunctionRetryPolicy) -> Self {
self.retry_policy = Some(policy);
@@ -110,11 +186,68 @@ impl RefreshColumnOptions {
self
}
/// The first set option a LanceDB Embedded refresh cannot honor, if any.
pub fn enterprise_only_option(&self) -> Option<&'static str> {
[
("where", self.where_clause.is_some()),
("retry_policy", self.retry_policy.is_some()),
("error_policy", self.error_policy.is_some()),
("max_failed_fraction", self.max_failed_fraction.is_some()),
("max_failed_rows", self.max_failed_rows.is_some()),
("row_timeout_seconds", self.row_timeout_seconds.is_some()),
("init_timeout_seconds", self.init_timeout_seconds.is_some()),
("num_workers", self.num_workers.is_some()),
(
"max_executor_processes_per_worker",
self.max_executor_processes_per_worker.is_some(),
),
]
.into_iter()
.find_map(|(name, set)| set.then_some(name))
}
/// Validate client-side constraints before any request is sent.
pub fn validate(&self) -> Result<()> {
if self
.where_clause
.as_deref()
.is_some_and(|f| f.trim().is_empty())
{
return Err(Error::InvalidInput {
message: "where must not be empty".into(),
});
}
if let Some(policy) = &self.retry_policy {
policy.validate()?;
}
if self
.max_failed_fraction
.is_some_and(|f| !f.is_finite() || !(0.0..=1.0).contains(&f))
{
return Err(Error::InvalidInput {
message: "max_failed_fraction must be a finite number between 0 and 1".into(),
});
}
for (name, seconds) in [
("row_timeout_seconds", self.row_timeout_seconds),
("init_timeout_seconds", self.init_timeout_seconds),
] {
if seconds.is_some_and(|s| !s.is_finite() || s <= 0.0) {
return Err(Error::InvalidInput {
message: format!("{name} must be a positive finite number"),
});
}
}
if self.num_workers == Some(0) {
return Err(Error::InvalidInput {
message: "num_workers must be greater than zero".into(),
});
}
if self.max_executor_processes_per_worker == Some(0) {
return Err(Error::InvalidInput {
message: "max_executor_processes_per_worker must be greater than zero".into(),
});
}
if self.batch_size == Some(0) {
return Err(Error::InvalidInput {
message: "batch_size must be greater than zero".into(),