From e9774ed168d3e82a50ba8cee19f602ca2ac41cfa Mon Sep 17 00:00:00 2001 From: Jack Ye Date: Wed, 11 Nov 2026 07:34:54 -0800 Subject: [PATCH] feat(nodejs): expose Function refresh publication intervals (#4957) Expose `publicationIntervalRows` on TypeScript refresh options and forward it as `publication_interval_rows`. Accept positive safe integers and preserve whole-fragment publication when omitted. ```typescript const job = await table.refreshColumnAsync("embedding", { publicationIntervalRows: 10000, }); await job.wait(); ``` Document the threshold semantics and add coverage for serialization, omission, invalid values, safe-integer boundaries, and Embedded rejection. Closes ENT-3755. --- docs/src/js/classes/Table.md | 6 +++ .../src/js/interfaces/RefreshColumnOptions.md | 13 ++++++ .../feature-engineering/refresh-column.mdx | 15 ++++-- nodejs/__test__/remote.test.ts | 46 +++++++++++++++++++ nodejs/__test__/table.test.ts | 8 ++++ nodejs/lancedb/table.ts | 6 +++ nodejs/src/table.rs | 15 ++++-- 7 files changed, 103 insertions(+), 6 deletions(-) diff --git a/docs/src/js/classes/Table.md b/docs/src/js/classes/Table.md index aba747a7c..431796843 100644 --- a/docs/src/js/classes/Table.md +++ b/docs/src/js/classes/Table.md @@ -1197,6 +1197,11 @@ LanceDB Enterprise it is the server's backfill job. `resourceBudget` for registered Functions. All are LanceDB Enterprise only; a LanceDB Embedded refresh rejects every option other than `batchSize`, `gpuClass` and `gpuCount`. + `publicationIntervalRows` opts into publishing completed Function split + ranges once their accumulated unpublished rows reach the threshold. + It must be a positive safe integer and requires non-Blob Function outputs. + Omit it for whole-fragment publication. Publications can exceed the + threshold, and remaining rows publish at completion. #### Returns @@ -1212,6 +1217,7 @@ const job = await table.refreshColumnAsync("embedding", { memoryBytes: 16 * 1024 ** 3, }, maxConcurrency: 4, + publicationIntervalRows: 10000, errorPolicy: "skip", maxFailedFraction: 0.01, rowTimeoutSeconds: 30, diff --git a/docs/src/js/interfaces/RefreshColumnOptions.md b/docs/src/js/interfaces/RefreshColumnOptions.md index 6cc8f7f8f..c534e0fac 100644 --- a/docs/src/js/interfaces/RefreshColumnOptions.md +++ b/docs/src/js/interfaces/RefreshColumnOptions.md @@ -117,6 +117,19 @@ Worker count for an expression column; Function columns reject it. *** +### publicationIntervalRows? + +```ts +optional publicationIntervalRows: number; +``` + +Row-count threshold for publishing completed Function split ranges. +Must be a positive safe integer. Omit for whole-fragment publication. +Actual publications can be larger; remaining rows publish at completion. +Requires LanceDB Enterprise with non-Blob Function outputs. + +*** + ### resourceBudget? ```ts diff --git a/docs/web/feature-engineering/refresh-column.mdx b/docs/web/feature-engineering/refresh-column.mdx index ed4838497..1b809b91d 100644 --- a/docs/web/feature-engineering/refresh-column.mdx +++ b/docs/web/feature-engineering/refresh-column.mdx @@ -414,8 +414,8 @@ and `gpuCount` in the `refreshColumnAsync` options. Completed Function outputs publish with their fragment by default. To opt into -a row-count publication threshold, set `publication_interval_rows` in Python or -`publication_interval_rows()` in Rust: +a row-count publication threshold, set `publication_interval_rows` in Python, +`publicationIntervalRows` in TypeScript, or `publication_interval_rows()` in Rust: ```python Python icon="python" @@ -426,6 +426,13 @@ job = table.refresh_column_async( job.wait() ``` +```typescript TypeScript icon="square-js" +const job = await table.refreshColumnAsync("embedding", { + publicationIntervalRows: 60, +}); +await job.wait(); +``` + ```rust Rust icon="rust" use lancedb::table::RefreshColumnOptions; @@ -443,7 +450,9 @@ individual publication can differ from this value depending on completed work and Function batching; this is not an exact-N contract. Omit it to keep whole-fragment publication. -The TypeScript SDK does not expose the publication threshold option yet. +In TypeScript, the threshold must be a positive safe integer, at most +`Number.MAX_SAFE_INTEGER`. This option requires a supporting LanceDB Enterprise +server and Function outputs without Blob types; Blob inputs are allowed. 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 diff --git a/nodejs/__test__/remote.test.ts b/nodejs/__test__/remote.test.ts index f87c70091..4ac08b4c4 100644 --- a/nodejs/__test__/remote.test.ts +++ b/nodejs/__test__/remote.test.ts @@ -776,6 +776,7 @@ describe("remote connection", () => { const job = await table.refreshColumnAsync("embedding", { where: "id % 2 = 0", batchSize: 16, + publicationIntervalRows: 60, maxConcurrency: 3, gpuClass: "a10g", gpuCount: 2, @@ -788,6 +789,18 @@ describe("remote connection", () => { }); expect(job.id).toBe("j-9"); await table.refreshColumnAsync("embedding"); + await table.refreshColumnAsync("embedding", { + publicationIntervalRows: undefined, + }); + for (const publicationIntervalRows of [ + 1, + 2 ** 32 + 1, + Number.MAX_SAFE_INTEGER, + ]) { + await table.refreshColumnAsync("embedding", { + publicationIntervalRows, + }); + } await table.refreshColumnAsync("embedding", { resourceBudget: { cpuMillis: 8_000, @@ -799,6 +812,14 @@ describe("remote connection", () => { maxConcurrency: true as unknown as number, }), ).rejects.toThrow(); + for (const publicationIntervalRows of [true, "60", 60n]) { + await expect( + table.refreshColumnAsync("embedding", { + publicationIntervalRows: + publicationIntervalRows as unknown as number, + }), + ).rejects.toThrow(); + } // Invalid options reject before any request is sent. for (const [options, message] of [ [{ maxFailedRows: -1 }, "maxFailedRows"], @@ -806,6 +827,22 @@ describe("remote connection", () => { [{ where: " " }, "where"], [{ maxFailedFraction: 2 }, "max_failed_fraction"], [{ rowTimeoutSeconds: 0 }, "row_timeout_seconds"], + [{ publicationIntervalRows: 0 }, "publicationIntervalRows"], + [{ publicationIntervalRows: -1 }, "publicationIntervalRows"], + [{ publicationIntervalRows: 1.5 }, "publicationIntervalRows"], + [{ publicationIntervalRows: Number.NaN }, "publicationIntervalRows"], + [ + { publicationIntervalRows: Number.POSITIVE_INFINITY }, + "publicationIntervalRows", + ], + [ + { publicationIntervalRows: Number.NEGATIVE_INFINITY }, + "publicationIntervalRows", + ], + [ + { publicationIntervalRows: Number.MAX_SAFE_INTEGER + 1 }, + "publicationIntervalRows", + ], [{ maxConcurrency: 1.5 }, "maxConcurrency"], [{ maxConcurrency: 0 }, "maxConcurrency"], [{ maxConcurrency: -1 }, "maxConcurrency"], @@ -893,6 +930,7 @@ describe("remote connection", () => { expect(bodies).toEqual([ JSON.parse( `{"column": "embedding", "where": "id % 2 = 0", "batch_size": 16, + "publication_interval_rows": 60, "max_concurrency": 3, "gpu_class": "a10g", "gpu_count": 2, "error_policy": "skip", "max_failed_fraction": 0.01, "max_failed_rows": 50, @@ -900,6 +938,14 @@ describe("remote connection", () => { "max_executor_processes_per_worker": 2}`, ), { column: "embedding" }, + { column: "embedding" }, + JSON.parse('{"column": "embedding", "publication_interval_rows": 1}'), + JSON.parse( + '{"column": "embedding", "publication_interval_rows": 4294967297}', + ), + JSON.parse( + '{"column": "embedding", "publication_interval_rows": 9007199254740991}', + ), JSON.parse( `{"column": "embedding", "resource_budget": { diff --git a/nodejs/__test__/table.test.ts b/nodejs/__test__/table.test.ts index 80d8296e2..b803e9043 100644 --- a/nodejs/__test__/table.test.ts +++ b/nodejs/__test__/table.test.ts @@ -5686,6 +5686,14 @@ describe("computed columns", () => { await expect( table.refreshColumnAsync("x", { errorPolicy: "skip" }), ).rejects.toThrow("error_policy requires a LanceDB Enterprise refresh"); + for (const refresh of [ + () => table.refreshColumn("x", { publicationIntervalRows: 60 }), + () => table.refreshColumnAsync("x", { publicationIntervalRows: 60 }), + ]) { + await expect(refresh()).rejects.toThrow( + "publication_interval_rows requires a LanceDB Enterprise refresh", + ); + } await expect( table.refreshColumnAsync("x", { maxConcurrency: 4 }), ).rejects.toThrow("max_concurrency requires a LanceDB Enterprise refresh"); diff --git a/nodejs/lancedb/table.ts b/nodejs/lancedb/table.ts index 703fc7926..b38c2381c 100644 --- a/nodejs/lancedb/table.ts +++ b/nodejs/lancedb/table.ts @@ -808,6 +808,11 @@ export abstract class Table { * `resourceBudget` for registered Functions. All are LanceDB Enterprise * only; a LanceDB Embedded refresh rejects every option other than * `batchSize`, `gpuClass` and `gpuCount`. + * `publicationIntervalRows` opts into publishing completed Function split + * ranges once their accumulated unpublished rows reach the threshold. + * It must be a positive safe integer and requires non-Blob Function outputs. + * Omit it for whole-fragment publication. Publications can exceed the + * threshold, and remaining rows publish at completion. * @example * ```ts * const job = await table.refreshColumnAsync("embedding", { @@ -817,6 +822,7 @@ export abstract class Table { * memoryBytes: 16 * 1024 ** 3, * }, * maxConcurrency: 4, + * publicationIntervalRows: 10000, * errorPolicy: "skip", * maxFailedFraction: 0.01, * rowTimeoutSeconds: 30, diff --git a/nodejs/src/table.rs b/nodejs/src/table.rs index 62fe811ad..e3bd22c0b 100644 --- a/nodejs/src/table.rs +++ b/nodejs/src/table.rs @@ -1649,7 +1649,7 @@ pub struct FunctionWorkerResourceBudget { pub gpu_count: Option, } -fn resource_budget_integer(value: f64, name: &str) -> napi::Result { +fn positive_safe_integer(value: f64, name: &str) -> napi::Result { const MAX_SAFE_INTEGER: f64 = 9_007_199_254_740_991.0; if !value.is_finite() || value.fract() != 0.0 || !(1.0..=MAX_SAFE_INTEGER).contains(&value) { return Err(napi::Error::from_reason(format!( @@ -1699,6 +1699,11 @@ pub struct RefreshColumnOptions { pub where_clause: Option, /// Maximum rows per Function call; must be greater than zero. pub batch_size: Option, + /// Row-count threshold for publishing completed Function split ranges. + /// Must be a positive safe integer. Omit for whole-fragment publication. + /// Actual publications can be larger; remaining rows publish at completion. + /// Requires LanceDB Enterprise with non-Blob Function outputs. + pub publication_interval_rows: Option, /// Override the registered Function's maximum concurrent user-code calls /// for this logical execution. pub max_concurrency: Option, @@ -1735,6 +1740,10 @@ impl TryFrom for lancedb::table::RefreshColumnOptions { let mut options = Self::default(); options.where_clause = value.where_clause; options.batch_size = value.batch_size.map(|size| size as usize); + options.publication_interval_rows = value + .publication_interval_rows + .map(|rows| positive_safe_integer(rows, "publicationIntervalRows")) + .transpose()?; options.max_concurrency = positive_u32(value.max_concurrency, "maxConcurrency")?; options.gpu_class = value.gpu_class; options.resource_budget = value @@ -1742,8 +1751,8 @@ impl TryFrom for lancedb::table::RefreshColumnOptions { .map(|budget| { Ok::<_, napi::Error>( LanceFunctionWorkerResourceBudget::new( - resource_budget_integer(budget.cpu_millis, "resourceBudget.cpuMillis")?, - resource_budget_integer(budget.memory_bytes, "resourceBudget.memoryBytes")?, + positive_safe_integer(budget.cpu_millis, "resourceBudget.cpuMillis")?, + positive_safe_integer(budget.memory_bytes, "resourceBudget.memoryBytes")?, ) .with_gpu_count(resource_budget_gpu_count(budget.gpu_count)?), )