diff --git a/docs/src/js/classes/Table.md b/docs/src/js/classes/Table.md index 1226d5a1c..8bdc6b5f8 100644 --- a/docs/src/js/classes/Table.md +++ b/docs/src/js/classes/Table.md @@ -1082,7 +1082,7 @@ for await (const batch of table.query()) { ### refreshColumn() ```ts -abstract refreshColumn(column): Promise +abstract refreshColumn(column, options?): Promise ``` 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`<[`RefreshColumnResult`](../interfaces/RefreshColumnResult.md)> @@ -1110,7 +1114,7 @@ number of rows filled and the new version number of the table. ### refreshColumnAsync() ```ts -abstract refreshColumnAsync(column): Promise +abstract refreshColumnAsync(column, options?): Promise ``` 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`<[`Job`](Job.md)> @@ -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" ``` diff --git a/docs/src/js/globals.md b/docs/src/js/globals.md index d6437219e..46f8db833 100644 --- a/docs/src/js/globals.md +++ b/docs/src/js/globals.md @@ -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) diff --git a/docs/src/js/interfaces/RefreshColumnOptions.md b/docs/src/js/interfaces/RefreshColumnOptions.md new file mode 100644 index 000000000..840012040 --- /dev/null +++ b/docs/src/js/interfaces/RefreshColumnOptions.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. diff --git a/docs/web/feature-engineering/refresh-column.mdx b/docs/web/feature-engineering/refresh-column.mdx index 24d050362..8f5726307 100644 --- a/docs/web/feature-engineering/refresh-column.mdx +++ b/docs/web/feature-engineering/refresh-column.mdx @@ -241,10 +241,59 @@ GPU class discovery. 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. +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: + + +```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?; +``` + + +`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 diff --git a/nodejs/__test__/remote.test.ts b/nodejs/__test__/remote.test.ts index b11a79a67..242be6101 100644 --- a/nodejs/__test__/remote.test.ts +++ b/nodejs/__test__/remote.test.ts @@ -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) => { diff --git a/nodejs/__test__/table.test.ts b/nodejs/__test__/table.test.ts index bbe1d6a71..af66ebb60 100644 --- a/nodejs/__test__/table.test.ts +++ b/nodejs/__test__/table.test.ts @@ -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 }]); diff --git a/nodejs/lancedb/index.ts b/nodejs/lancedb/index.ts index 58f687e1a..0e054d70e 100644 --- a/nodejs/lancedb/index.ts +++ b/nodejs/lancedb/index.ts @@ -56,6 +56,7 @@ export { MergeResult, AddResult, AddColumnsResult, + RefreshColumnOptions, RefreshColumnResult, FunctionErrors, FunctionErrorsOptions, diff --git a/nodejs/lancedb/table.ts b/nodejs/lancedb/table.ts index e84704803..de22ef0f8 100644 --- a/nodejs/lancedb/table.ts +++ b/nodejs/lancedb/table.ts @@ -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} A promise that resolves to the * number of rows filled and the new version number of the table. */ - abstract refreshColumn(column: string): Promise; + abstract refreshColumn( + column: string, + options?: RefreshColumnOptions, + ): Promise; /** * 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; + abstract refreshColumnAsync( + column: string, + options?: RefreshColumnOptions, + ): Promise; /** * 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 { - return await this.inner.refreshColumn(column); + async refreshColumn( + column: string, + options?: RefreshColumnOptions, + ): Promise { + return await this.inner.refreshColumn(column, options); } - async refreshColumnAsync(column: string): Promise { - return new Job(await this.inner.refreshColumnAsync(column)); + async refreshColumnAsync( + column: string, + options?: RefreshColumnOptions, + ): Promise { + return new Job(await this.inner.refreshColumnAsync(column, options)); } async functionErrors( diff --git a/nodejs/src/table.rs b/nodejs/src/table.rs index c480ba042..5c3805f51 100644 --- a/nodejs/src/table.rs +++ b/nodejs/src/table.rs @@ -421,20 +421,30 @@ impl Table { } #[napi(catch_unwind)] - pub async fn refresh_column(&self, column: String) -> napi::Result { + pub async fn refresh_column( + &self, + column: String, + options: Option, + ) -> napi::Result { + 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 { + pub async fn refresh_column_async( + &self, + column: String, + options: Option, + ) -> napi::Result { + 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, } +/// 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, + /// Maximum rows per Function call; must be greater than zero. + pub batch_size: Option, + /// GPU class for Function workers. + pub gpu_class: Option, + /// Positive aggregate GPU budget for Function workers. + pub gpu_count: Option, + /// `"fail"` (default) stops on a row error; `"skip"` writes null and + /// records the error, subject to the failure limits. + pub error_policy: Option, + /// Under a skip policy, fail the job once more than this share of rows + /// (0 to 1) has failed. + pub max_failed_fraction: Option, + /// Under a skip policy, fail the job once more than this many rows have + /// failed. + pub max_failed_rows: Option, + /// Deadline for one Function call, in seconds. + pub row_timeout_seconds: Option, + /// Deadline for Function initialization, in seconds. + pub init_timeout_seconds: Option, + /// Worker count for an expression column; Function columns reject it. + pub num_workers: Option, + /// Cap on executor processes per Function worker. + pub max_executor_processes_per_worker: Option, +} + +impl TryFrom for lancedb::table::RefreshColumnOptions { + type Error = napi::Error; + + fn try_from(value: RefreshColumnOptions) -> napi::Result { + 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)] diff --git a/python/python/lancedb/_lancedb.pyi b/python/python/lancedb/_lancedb.pyi index f0f6027ad..a1a893f8b 100644 --- a/python/python/lancedb/_lancedb.pyi +++ b/python/python/lancedb/_lancedb.pyi @@ -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, diff --git a/python/python/lancedb/remote/table.py b/python/python/lancedb/remote/table.py index c916fcb92..6bef59922 100644 --- a/python/python/lancedb/remote/table.py +++ b/python/python/lancedb/remote/table.py @@ -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, ) ) ) diff --git a/python/python/lancedb/table.py b/python/python/lancedb/table.py index 559552cee..f43df8b40 100644 --- a/python/python/lancedb/table.py +++ b/python/python/lancedb/table.py @@ -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, ) diff --git a/python/python/tests/test_remote_db.py b/python/python/tests/test_remote_db.py index 8e6d74e86..8a65a2a1c 100644 --- a/python/python/tests/test_remote_db.py +++ b/python/python/tests/test_remote_db.py @@ -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 diff --git a/python/python/tests/test_table.py b/python/python/tests/test_table.py index 45ce9276e..2a502fe62 100644 --- a/python/python/tests/test_table.py +++ b/python/python/tests/test_table.py @@ -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 diff --git a/python/src/table.rs b/python/src/table.rs index 4c5a919d2..6371d94c0 100644 --- a/python/src/table.rs +++ b/python/src/table.rs @@ -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, batch_size: Option, gpu_class: Option, gpu_count: Option, retry_policy: Option, error_policy: Option, -) -> PyResult { + max_failed_fraction: Option, + max_failed_rows: Option, + row_timeout_seconds: Option, + init_timeout_seconds: Option, + num_workers: Option, + max_executor_processes_per_worker: Option, +} + +fn positive_count>(value: Option, name: &str) -> PyResult> { + 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 { + 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, batch_size: Option, gpu_class: Option, gpu_count: Option, retry_policy: Option, error_policy: Option, + max_failed_fraction: Option, + max_failed_rows: Option, + row_timeout_seconds: Option, + init_timeout_seconds: Option, + num_workers: Option, + max_executor_processes_per_worker: Option, ) -> PyResult> { - 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, batch_size: Option, gpu_class: Option, gpu_count: Option, retry_policy: Option, error_policy: Option, + max_failed_fraction: Option, + max_failed_rows: Option, + row_timeout_seconds: Option, + init_timeout_seconds: Option, + num_workers: Option, + max_executor_processes_per_worker: Option, ) -> PyResult> { - 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 diff --git a/rust/lancedb/src/remote/table.rs b/rust/lancedb/src/remote/table.rs index abba7ff94..b781531fe 100644 --- a/rust/lancedb/src/remote/table.rs +++ b/rust/lancedb/src/remote/table.rs @@ -3969,22 +3969,10 @@ impl BaseTable for RemoteTable { ) -> Result> { 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 { + 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; diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs index 94a934c43..aefc0ac91 100644 --- a/rust/lancedb/src/table.rs +++ b/rust/lancedb/src/table.rs @@ -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 { 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> { 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 diff --git a/rust/lancedb/src/table/refresh.rs b/rust/lancedb/src/table/refresh.rs index a9e69452d..a8ae3c286 100644 --- a/rust/lancedb/src/table/refresh.rs +++ b/rust/lancedb/src/table/refresh.rs @@ -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, /// Override the Function's default user-code retry policy. Enterprise only. + #[serde(skip_serializing_if = "Option::is_none")] pub retry_policy: Option, /// Action after user-code failure or retry exhaustion. Enterprise only. + #[serde(skip_serializing_if = "Option::is_none")] pub error_policy: Option, + /// 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, + /// 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, + /// Deadline for one Function call, in seconds. Enterprise only. + #[serde(skip_serializing_if = "Option::is_none")] + pub row_timeout_seconds: Option, + /// Deadline for Function initialization, in seconds. Enterprise only. + #[serde(skip_serializing_if = "Option::is_none")] + pub init_timeout_seconds: Option, + /// 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, + /// 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, /// 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, /// 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, /// 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, } impl RefreshColumnOptions { + /// Only refresh the rows matching `filter`. + pub fn where_clause(mut self, filter: impl Into) -> 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(),