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.
This commit is contained in:
Jack Ye
2026-11-11 07:34:54 -08:00
committed by GitHub
parent c1fee36045
commit e9774ed168
7 changed files with 103 additions and 6 deletions
+6
View File
@@ -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,
@@ -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
@@ -414,8 +414,8 @@ and `gpuCount` in the `refreshColumnAsync` options.
</Note>
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:
<CodeGroup>
```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
+46
View File
@@ -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": {
+8
View File
@@ -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");
+6
View File
@@ -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,
+12 -3
View File
@@ -1649,7 +1649,7 @@ pub struct FunctionWorkerResourceBudget {
pub gpu_count: Option<f64>,
}
fn resource_budget_integer(value: f64, name: &str) -> napi::Result<u64> {
fn positive_safe_integer(value: f64, name: &str) -> napi::Result<u64> {
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<String>,
/// Maximum rows per Function call; must be greater than zero.
pub batch_size: Option<u32>,
/// 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<f64>,
/// Override the registered Function's maximum concurrent user-code calls
/// for this logical execution.
pub max_concurrency: Option<f64>,
@@ -1735,6 +1740,10 @@ impl TryFrom<RefreshColumnOptions> 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<RefreshColumnOptions> 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)?),
)