feat: configure automatic refresh when declaring computed columns (#4607)

Add `auto_refresh_enabled` and `auto_refresh_interval_seconds` to the
Rust builder and Python sync/async add-column APIs for computed columns
and registered Function outputs.

Python users can configure refresh when declaring a column:

```python
table.add_columns(
    computed={"total": "price * quantity"},
    auto_refresh_enabled=True,
    auto_refresh_interval_seconds=300,
)
```

Omit the interval to let the server determine when to refresh:

```python
table.add_columns(computed={"discounted_total": "total * 0.9"})
```

The async API accepts the same options:

```python
await async_table.add_columns(
    computed={"total": "price * quantity"},
    auto_refresh_enabled=True,
    auto_refresh_interval_seconds=300,
)
```

For an existing registered `FunctionApplication`, the options also apply
to Function columns. Disabling automatic refresh still permits explicit
refresh:

```python
table.add_columns(
    {"features": function_application},
    auto_refresh_enabled=False,
    auto_refresh_interval_seconds=300,
)
table.refresh_column("features")
```

Options apply to every output declared by a call. Automatic refresh is
enabled by default; when the interval is omitted, the server determines
when to refresh. An explicit interval sets a minimum delay after a
successful refresh. The interval does not force recomputation of
unchanged inputs. Invalid intervals and options on one-time transforms
are rejected before declaration.

Native declarations persist `computed_column.auto_refresh_enabled` and
an explicitly supplied `computed_column.auto_refresh_interval_seconds`
in each output field’s metadata in the same commit as the declaration.
An omitted interval is neither stored nor sent in remote requests.
Remote declarations forward the settings to Cloud/Enterprise servers.
Local tables store these settings; automatic execution requires a
supporting server with scheduling enabled.
This commit is contained in:
Jack Ye
2026-10-21 20:27:11 -07:00
committed by GitHub
parent d97df0c82b
commit a4544c5472
13 changed files with 573 additions and 47 deletions
@@ -57,6 +57,54 @@ That is what makes the call idempotent and safe to retry. However, unlike a
spreadsheet formula, nothing is recomputed until you refresh, so a derived
column is stale between an update and the next refresh.
{/*
Draft for automatic computed-column refresh. Keep the released manual-refresh
instructions above until the server feature is available.
## Automatic refresh {#automatic}
Set scheduling options when declaring a computed column:
```python
table.add_columns(
computed={"doubled": "x * 2"},
auto_refresh_enabled=True,
auto_refresh_interval_seconds=300,
)
```
Both options apply to every computed output declared in that call, including
registered Function outputs. Automatic refresh is enabled by default. When
the interval is omitted, the server determines when to refresh. Local tables
persist only an explicitly configured interval; automatic execution requires a
Cloud or Enterprise server with its separate `column_refresh.enabled` switch
turned on. That server switch is disabled in the base configuration.
The interval is a positive minimum delay after the last successful refresh
of that output. A first refresh is immediately eligible. The scheduler still
checks input freshness and waits for computed ancestors to settle; the timer
does not make unchanged inputs require recomputation. Each Function output's
interval is independent, so refreshing a sibling does not postpone it.
Pacing uses retained live Job Registry history. If completed jobs are archived
before a configured interval expires, the scheduler can lose the last-success
timestamp and refresh earlier. Leave archival disabled, or retain live history
for at least the longest configured column interval. Archival is disabled in
the base configuration; its default retention threshold is seven days.
Change a stored setting through SQL:
```sql
ALTER TABLE my_table ALTER COLUMN doubled SET auto_refresh_enabled = false;
ALTER TABLE my_table ALTER COLUMN doubled SET auto_refresh_interval_seconds = 600;
ALTER TABLE my_table ALTER COLUMN doubled SET auto_refresh_enabled = true;
```
Disabling automatic refresh prevents new automatic jobs. Inflight jobs may
finish, and manual refresh remains available. Re-enabling wakes maintenance
to consider accumulated changes under the interval and dependency rules.
*/}
## Refreshing part of a table {#partial}
<Badge color="red">Enterprise</Badge> In SQL a refresh can be scoped to a
+42 -3
View File
@@ -51,6 +51,43 @@ The column is committed empty and the expression is stored, not evaluated:
declaring one on a billion-row table costs what declaring one on an empty table
costs. [`REFRESH COLUMN`](/sql/dml/refresh) fills it.
{/*
Draft for automatic computed-column refresh. Enable this section when the
server feature is released.
### Automatic refresh options {#automatic-refresh-options}
```sql
ALTER TABLE t ADD COLUMN doubled AS value * 2
WITH (auto_refresh_enabled = true, auto_refresh_interval_seconds = 300);
ALTER TABLE t ALTER COLUMN doubled SET auto_refresh_enabled = false;
ALTER TABLE t ALTER COLUMN doubled SET auto_refresh_interval_seconds = 600;
ALTER TABLE t ALTER COLUMN doubled SET auto_refresh_enabled = true;
```
Automatic refresh is enabled by default. An explicitly configured interval is
stored in each computed output's field metadata; when omitted, the server
determines when to refresh. The interval must be a positive integer
and is minimum pacing after a successful refresh of that output; unchanged
inputs do not force work. A new column is immediately eligible for its first
refresh. The server's separate automatic-refresh switch must also be enabled.
These options also work beside a Function binding's `version` and `branch`:
```sql
ALTER TABLE t ADD COLUMN (label, score) AS classify(text)
WITH (version = 'fv_1', auto_refresh_enabled = true,
auto_refresh_interval_seconds = 300);
ALTER TABLE t ALTER COLUMN score SET auto_refresh_enabled = false;
```
Declaration settings initialize every output in the group; later SET commands
change one output. Disabling automatic refresh prevents new submissions while
allowing inflight jobs to finish and explicit REFRESH COLUMN to run. Column
SUSPEND/RESUME syntax is not supported.
*/}
{/*
Draft for the Functions release.
@@ -81,12 +118,14 @@ ALTER TABLE t ADD COLUMN (label, score) AS classify(text) WITH (version = 'fv_1'
```sql
ALTER TABLE t ADD COLUMN c AS f(x) FILTER (WHERE keep) WITH (version = 'fv_1')
ALTER TABLE t ADD COLUMN c AS f(x) WITH (num_workers = '2')
REFRESH COLUMN c ON t WITH (num_workers = 2)
```
<Note>
`WITH` on a plain computed column accepts only `version`. Resource options such
as `num_workers` apply to function columns, where there are workers to size.
Declaration `WITH` accepts `auto_refresh_enabled` and
`auto_refresh_interval_seconds`. Function bindings also accept `version` and
`branch`. Worker resource options such as `num_workers` belong on
`REFRESH COLUMN ... WITH (...)`.
</Note>
*/}
+9 -2
View File
@@ -501,10 +501,17 @@ class Table:
async def delete(self, filter: Union[str, PyExpr]) -> DeleteResult: ...
async def add_columns(self, columns: list[tuple[str, str]]) -> AddColumnsResult: ...
async def add_computed_columns(
self, columns: list[tuple[str, str]]
self,
columns: list[tuple[str, str]],
auto_refresh_enabled: Optional[bool] = None,
auto_refresh_interval_seconds: Optional[int] = None,
) -> AddColumnsResult: ...
async def add_function_columns(
self, application_json: str, output_name: Optional[str]
self,
application_json: str,
output_name: Optional[str],
auto_refresh_enabled: Optional[bool] = None,
auto_refresh_interval_seconds: Optional[int] = None,
) -> AddColumnsResult: ...
async def refresh_column(self, column: str) -> RefreshColumnResult: ...
async def refresh_column_async(self, column: str) -> Job: ...
+10 -1
View File
@@ -1001,8 +1001,17 @@ class RemoteTable(Table):
| None = None,
*,
computed: Dict[str, str] | None = None,
auto_refresh_enabled: bool | None = None,
auto_refresh_interval_seconds: int | None = None,
) -> AddColumnsResult:
return LOOP.run(self._table.add_columns(transforms, computed=computed))
return LOOP.run(
self._table.add_columns(
transforms,
computed=computed,
auto_refresh_enabled=auto_refresh_enabled,
auto_refresh_interval_seconds=auto_refresh_interval_seconds,
)
)
def refresh_column(self, column: str):
return LOOP.run(self._table.refresh_column(column))
+70 -3
View File
@@ -2561,6 +2561,8 @@ class Table(ABC):
| None = None,
*,
computed: Dict[str, str] | None = None,
auto_refresh_enabled: bool | None = None,
auto_refresh_interval_seconds: int | None = None,
):
"""
Add new columns with defined values.
@@ -2605,6 +2607,15 @@ class Table(ABC):
[`refresh_column_async`][lancedb.table.Table.refresh_column_async].
Cannot be combined with ``transforms``.
auto_refresh_enabled: bool, optional
Enable automatic refresh for each declared computed output. Defaults
to True. Local tables store this option; scheduling requires a server.
auto_refresh_interval_seconds: int, optional
Positive minimum interval after a successful refresh, in seconds.
When omitted, the server determines when to refresh. Applies to every
computed output in this call.
Scheduling options require computed or Function columns.
Returns
-------
AddColumnsResult
@@ -4803,8 +4814,17 @@ class LanceTable(Table):
| None = None,
*,
computed: Dict[str, str] | None = None,
auto_refresh_enabled: bool | None = None,
auto_refresh_interval_seconds: int | None = None,
) -> AddColumnsResult:
return LOOP.run(self._table.add_columns(transforms, computed=computed))
return LOOP.run(
self._table.add_columns(
transforms,
computed=computed,
auto_refresh_enabled=auto_refresh_enabled,
auto_refresh_interval_seconds=auto_refresh_interval_seconds,
)
)
def refresh_column(self, column: str) -> "RefreshColumnResult":
"""Fill a computed column's unfilled rows and recompute those whose
@@ -6852,6 +6872,8 @@ class AsyncTable:
| None = None,
*,
computed: dict[str, str] | None = None,
auto_refresh_enabled: bool | None = None,
auto_refresh_interval_seconds: int | None = None,
) -> AddColumnsResult:
"""
Add new columns with defined values.
@@ -6891,12 +6913,46 @@ class AsyncTable:
On LanceDB Cloud and Enterprise the expression is planned by
the server. Cannot be combined with ``transforms``.
auto_refresh_enabled: bool, optional
Enable automatic refresh for each computed output declared by this
call. Defaults to True. Local tables store the setting; automatic
scheduling requires Cloud or Enterprise.
auto_refresh_interval_seconds: int, optional
Positive minimum seconds after a successful refresh before another
automatic refresh is eligible. When omitted, the server determines when
to refresh. Unchanged inputs do
not force recomputation. Applies to every declared computed output.
These options cannot be used with one-time transforms or ordinary
all-null columns.
Examples
--------
>>> await table.add_columns( # doctest: +SKIP
... computed={"total": "price * quantity"},
... auto_refresh_enabled=True,
... auto_refresh_interval_seconds=300,
... )
Returns
-------
AddColumnsResult
version: the new version number of the table after adding columns.
"""
if auto_refresh_enabled is not None and type(auto_refresh_enabled) is not bool:
raise ValueError("auto_refresh_enabled must be a boolean")
if auto_refresh_interval_seconds is not None and (
type(auto_refresh_interval_seconds) is not int
or not 0 < auto_refresh_interval_seconds <= (2**64 - 1) // 1000
):
raise ValueError(
"auto_refresh_interval_seconds must be a positive integer "
"no greater than u64::MAX / 1000"
)
options_supplied = (
auto_refresh_enabled is not None
or auto_refresh_interval_seconds is not None
)
function_application = None
function_output_name = None
if isinstance(transforms, FunctionApplication):
@@ -6920,7 +6976,10 @@ class AsyncTable:
)
function_application._ensure_declarable()
return await self._inner.add_function_columns(
function_application.to_canonical_json(), function_output_name
function_application.to_canonical_json(),
function_output_name,
auto_refresh_enabled,
auto_refresh_interval_seconds,
)
if isinstance(transforms, pa.Field):
@@ -6934,7 +6993,15 @@ class AsyncTable:
raise ValueError(
"add_columns cannot take both transforms and computed columns"
)
return await self._inner.add_computed_columns(list(computed.items()))
return await self._inner.add_computed_columns(
list(computed.items()),
auto_refresh_enabled,
auto_refresh_interval_seconds,
)
if options_supplied:
raise ValueError(
"automatic refresh options require computed or Function columns"
)
if transforms is None:
raise ValueError("add_columns requires transforms or computed columns")
if isinstance(transforms, pa.Schema):
@@ -398,8 +398,21 @@ class _FunctionDeclarationInner:
def __init__(self):
self.calls = []
async def add_function_columns(self, application_json, output_name):
self.calls.append((json.loads(application_json), output_name))
async def add_function_columns(
self,
application_json,
output_name,
auto_refresh_enabled,
auto_refresh_interval_seconds,
):
self.calls.append(
(
json.loads(application_json),
output_name,
auto_refresh_enabled,
auto_refresh_interval_seconds,
)
)
return "declared"
@@ -420,14 +433,18 @@ async def test_add_columns_routes_struct_as_one_and_multi_output_binding_atomica
)
assert result == "declared"
assert inner.calls[-1][1] == "features"
assert inner.calls[-1][2:] == (None, None)
bare = application._copy(update={"columns": {}}).rename(
columns={"normalized_text": "search_text"}
)
result = await table.add_columns(bare)
result = await table.add_columns(
bare, auto_refresh_enabled=False, auto_refresh_interval_seconds=300
)
assert result == "declared"
assert inner.calls[-1][1] is None
assert inner.calls[-1][0]["columns"] == {"normalized_text": "search_text"}
assert inner.calls[-1][2:] == (False, 300)
@pytest.mark.asyncio
+50
View File
@@ -4869,3 +4869,53 @@ async def test_refresh_column_async_job_async_table(tmp_path):
assert result.published_version == 4
assert await job.status() == "finished"
assert (await table.to_arrow())["tripled"].to_pylist() == [9]
@pytest.mark.parametrize("interval", [None, 300])
def test_computed_auto_refresh_options(tmp_path, interval):
db = lancedb.connect(tmp_path)
table = db.create_table("refresh_options", [{"x": 1}])
before = table.version
table.add_columns(
computed={"doubled": "x * 2", "tripled": "x * 3"},
auto_refresh_enabled=False,
auto_refresh_interval_seconds=interval,
)
assert table.version == before + 1
for column in ("doubled", "tripled"):
metadata = table.schema.field(column).metadata
assert metadata[b"computed_column.auto_refresh_enabled"] == b"false"
if interval is None:
assert b"computed_column.auto_refresh_interval_seconds" not in metadata
else:
assert metadata[b"computed_column.auto_refresh_interval_seconds"] == b"300"
table.refresh_column("doubled")
assert table.to_arrow()["doubled"].to_pylist() == [2]
@pytest.mark.parametrize("value", [0, -1, True, 1.5, "300", 2**64])
def test_computed_auto_refresh_rejects_invalid_interval(tmp_path, value):
table = lancedb.connect(tmp_path).create_table("invalid_options", [{"x": 1}])
with pytest.raises(ValueError, match="auto_refresh_interval_seconds"):
table.add_columns(
computed={"doubled": "x * 2"}, auto_refresh_interval_seconds=value
)
assert table.schema.names == ["x"]
def test_auto_refresh_options_require_computed_columns(tmp_path):
table = lancedb.connect(tmp_path).create_table("invalid_kind", [{"x": 1}])
with pytest.raises(ValueError, match="require computed"):
table.add_columns({"doubled": "x * 2"}, auto_refresh_enabled=True)
@pytest.mark.asyncio
async def test_async_computed_auto_refresh_uses_server_interval(tmp_path):
db = await lancedb.connect_async(tmp_path)
table = await db.create_table("server_interval", [{"x": 1}])
before = await table.version()
await table.add_columns(computed={"doubled": "x * 2"})
assert await table.version() == before + 1
metadata = (await table.schema()).field("doubled").metadata
assert metadata[b"computed_column.auto_refresh_enabled"] == b"true"
assert b"computed_column.auto_refresh_interval_seconds" not in metadata
+27 -2
View File
@@ -1,5 +1,6 @@
// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
use lancedb::table::computed_columns::ComputedColumnOptions;
use std::{
collections::HashMap,
sync::{
@@ -1803,35 +1804,59 @@ impl Table {
})
}
#[pyo3(signature = (columns, auto_refresh_enabled=None, auto_refresh_interval_seconds=None))]
pub fn add_computed_columns(
self_: PyRef<'_, Self>,
columns: Vec<(String, String)>,
auto_refresh_enabled: Option<bool>,
auto_refresh_interval_seconds: Option<u64>,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner_ref()?.clone();
let options = ComputedColumnOptions {
auto_refresh_enabled: auto_refresh_enabled.unwrap_or(true),
auto_refresh_interval_seconds,
};
options.validate().infer_error()?;
future_into_py(self_.py(), async move {
let mut builder = inner.add_columns();
for (name, expression) in columns {
builder = builder.computed(name, expression);
}
let result = builder.execute().await.infer_error()?;
let result = builder
.computed_options(options)
.execute()
.await
.infer_error()?;
Ok(AddColumnsResult::from(result))
})
}
#[pyo3(signature = (application_json, output_name, auto_refresh_enabled=None, auto_refresh_interval_seconds=None))]
pub fn add_function_columns(
self_: PyRef<'_, Self>,
application_json: String,
output_name: Option<String>,
auto_refresh_enabled: Option<bool>,
auto_refresh_interval_seconds: Option<u64>,
) -> PyResult<Bound<'_, PyAny>> {
let application =
lancedb::function::FunctionApplication::from_json(&application_json).infer_error()?;
let inner = self_.inner_ref()?.clone();
let options = ComputedColumnOptions {
auto_refresh_enabled: auto_refresh_enabled.unwrap_or(true),
auto_refresh_interval_seconds,
};
options.validate().infer_error()?;
future_into_py(self_.py(), async move {
let builder = match output_name {
Some(name) => inner.add_columns().function_as(name, application),
None => inner.add_columns().function(application),
};
let result = builder.execute().await.infer_error()?;
let result = builder
.computed_options(options)
.execute()
.await
.infer_error()?;
Ok(AddColumnsResult::from(result))
})
}
+57 -29
View File
@@ -3497,7 +3497,11 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
Ok(result)
}
async fn add_computed_columns(&self, columns: &[(String, String)]) -> Result<AddColumnsResult> {
async fn add_computed_columns(
&self,
columns: &[(String, String)],
options: crate::table::computed_columns::ComputedColumnOptions,
) -> Result<AddColumnsResult> {
self.check_mutable().await?;
crate::table::computed_columns::ensure_not_function_bound(
self.schema().await?.as_ref(),
@@ -3516,7 +3520,10 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
},
)
.collect::<Vec<_>>();
let mut body = serde_json::json!({ "new_columns": entries });
let mut body = serde_json::json!({ "new_columns": entries, "auto_refresh_enabled": options.auto_refresh_enabled });
if let Some(interval) = options.auto_refresh_interval_seconds {
body["auto_refresh_interval_seconds"] = serde_json::json!(interval);
}
self.apply_branch_body(&mut body);
let request = self
.client
@@ -3550,6 +3557,7 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
&self,
application: &crate::function::FunctionApplication,
output_name: Option<&str>,
options: crate::table::computed_columns::ComputedColumnOptions,
) -> Result<AddColumnsResult> {
self.check_mutable().await?;
let schema = self.schema().await?;
@@ -3570,6 +3578,7 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
.collect::<Vec<_>>();
let mut body = serde_json::json!({
"new_columns": new_columns,
"auto_refresh_enabled": options.auto_refresh_enabled,
"function": {
"application": plan.application,
"binding_metadata_version": plan.binding_metadata_version,
@@ -3579,6 +3588,9 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
"outputs": plan.outputs,
},
});
if let Some(interval) = options.auto_refresh_interval_seconds {
body["auto_refresh_interval_seconds"] = serde_json::json!(interval);
}
self.apply_branch_body(&mut body);
let request = self
.client
@@ -9263,31 +9275,44 @@ mod tests {
/// client never types the expression itself.
#[tokio::test]
async fn test_add_computed_columns_sends_the_expression() {
let table = Table::new_with_handler("my_table", |request| match request.url().path() {
"/v1/table/my_table/describe/" => simple_describe_response(),
"/v1/table/my_table/add_columns/" => {
assert_eq!(request.method(), "POST");
let body = request.body().unwrap().as_bytes().unwrap();
let value: serde_json::Value = serde_json::from_slice(body).unwrap();
assert_eq!(
value["new_columns"],
serde_json::json!([{"name": "doubled", "computed": "x * 2"}])
);
http::Response::builder()
.status(200)
.body(r#"{"version": 7}"#.to_string())
.unwrap()
}
path => panic!("Unexpected path: {path}"),
});
for interval in [None, Some(300)] {
let table =
Table::new_with_handler("my_table", move |request| match request.url().path() {
"/v1/table/my_table/describe/" => simple_describe_response(),
"/v1/table/my_table/add_columns/" => {
assert_eq!(request.method(), "POST");
let body = request.body().unwrap().as_bytes().unwrap();
let value: serde_json::Value = serde_json::from_slice(body).unwrap();
assert_eq!(
value["new_columns"],
serde_json::json!([{"name": "doubled", "computed": "x * 2"}])
);
assert_eq!(value["auto_refresh_enabled"], false);
if let Some(interval) = interval {
assert_eq!(value["auto_refresh_interval_seconds"], interval);
} else {
assert!(value.get("auto_refresh_interval_seconds").is_none());
}
http::Response::builder()
.status(200)
.body(r#"{"version": 7}"#.to_string())
.unwrap()
}
path => panic!("Unexpected path: {path}"),
});
let result = table
.add_columns()
.computed("doubled", "x * 2")
.execute()
.await
.unwrap();
assert_eq!(result.version, 7);
let result = table
.add_columns()
.computed("doubled", "x * 2")
.computed_options(crate::table::computed_columns::ComputedColumnOptions {
auto_refresh_enabled: false,
auto_refresh_interval_seconds: interval,
})
.execute()
.await
.unwrap();
assert_eq!(result.version, 7);
}
}
#[tokio::test]
@@ -9305,10 +9330,11 @@ mod tests {
request.body().unwrap().as_bytes().unwrap(),
)
.unwrap();
let expected: serde_json::Value = serde_json::from_str(include_str!(
let mut expected: serde_json::Value = serde_json::from_str(include_str!(
"../../tests/fixtures/first_class_functions/v1/remote_scalar_declaration_request.json"
))
.unwrap();
expected["auto_refresh_enabled"] = serde_json::json!(true);
assert_eq!(actual, expected);
http::Response::builder()
.status(200)
@@ -9504,10 +9530,11 @@ mod tests {
request.body().unwrap().as_bytes().unwrap(),
)
.unwrap();
let expected: serde_json::Value = serde_json::from_str(include_str!(
let mut expected: serde_json::Value = serde_json::from_str(include_str!(
"../../tests/fixtures/first_class_functions/v1/remote_fixed_size_declaration_request.json"
))
.unwrap();
expected["auto_refresh_enabled"] = serde_json::json!(true);
assert_eq!(actual, expected);
http::Response::builder()
.status(200)
@@ -9550,10 +9577,11 @@ mod tests {
"/v1/table/my_table/add_columns/" => {
let actual: serde_json::Value =
serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()).unwrap();
let expected: serde_json::Value = serde_json::from_str(include_str!(
let mut expected: serde_json::Value = serde_json::from_str(include_str!(
"../../tests/fixtures/first_class_functions/v1/remote_multi_output_declaration_request.json"
))
.unwrap();
expected["auto_refresh_enabled"] = serde_json::json!(true);
assert_eq!(actual, expected);
http::Response::builder()
.status(200)
+8 -2
View File
@@ -814,6 +814,7 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
async fn add_computed_columns(
&self,
_columns: &[(String, String)],
_options: computed_columns::ComputedColumnOptions,
) -> Result<AddColumnsResult> {
Err(Error::NotSupported {
message: "computed columns are not supported on this table type".into(),
@@ -824,6 +825,7 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
&self,
_application: &crate::function::FunctionApplication,
_output_name: Option<&str>,
_options: computed_columns::ComputedColumnOptions,
) -> Result<AddColumnsResult> {
Err(Error::NotSupported {
message: "Function columns are supported only on LanceDB Cloud and Enterprise".into(),
@@ -3768,8 +3770,12 @@ impl BaseTable for NativeTable {
Ok(result)
}
async fn add_computed_columns(&self, columns: &[(String, String)]) -> Result<AddColumnsResult> {
let result = schema_evolution::execute_declare(self, columns).await?;
async fn add_computed_columns(
&self,
columns: &[(String, String)],
options: computed_columns::ComputedColumnOptions,
) -> Result<AddColumnsResult> {
let result = schema_evolution::execute_declare(self, columns, options).await?;
self.bump_freshness();
Ok(result)
}
+126 -2
View File
@@ -8,6 +8,7 @@ use std::sync::Arc;
use lance::dataset::NewColumnTransform;
use super::BaseTable;
use super::computed_columns::ComputedColumnOptions;
use super::schema_evolution::AddColumnsResult;
use crate::function::FunctionApplication;
use crate::{Error, Result};
@@ -19,6 +20,7 @@ pub struct AddColumnsBuilder {
computed: Vec<(String, String)>,
function: Option<(FunctionApplication, Option<String>)>,
read_columns: Option<Vec<String>>,
options: Option<ComputedColumnOptions>,
}
impl std::fmt::Debug for AddColumnsBuilder {
@@ -41,6 +43,7 @@ impl AddColumnsBuilder {
computed: Vec::new(),
function: None,
read_columns: None,
options: None,
}
}
@@ -139,6 +142,26 @@ impl AddColumnsBuilder {
self
}
/// Configure automatic refresh for every output declared in this call.
/// Settings are stored with the declaration. They require computed or
/// Function columns; local tables persist them without starting a scheduler.
///
/// ```
/// # use lancedb::{Table, table::computed_columns::ComputedColumnOptions};
/// # async fn declare(table: &Table) -> lancedb::Result<()> {
/// table.add_columns().computed("doubled", "x * 2")
/// .computed_options(ComputedColumnOptions {
/// auto_refresh_enabled: true,
/// auto_refresh_interval_seconds: Some(300),
/// }).execute().await?;
/// # Ok(())
/// # }
/// ```
pub fn computed_options(mut self, options: ComputedColumnOptions) -> Self {
self.options = Some(options);
self
}
/// Add the columns.
pub async fn execute(self) -> Result<AddColumnsResult> {
let Self {
@@ -147,7 +170,18 @@ impl AddColumnsBuilder {
computed,
function,
read_columns,
options,
} = self;
if let Some(options) = options {
options.validate()?;
if computed.is_empty() && function.is_none() {
return Err(Error::InvalidInput {
message: "automatic refresh options require computed or Function columns"
.into(),
});
}
}
let options = options.unwrap_or_default();
let declaration_count = usize::from(!computed.is_empty()) + usize::from(function.is_some());
if transform.is_some() && declaration_count != 0 || declaration_count > 1 {
@@ -179,7 +213,7 @@ impl AddColumnsBuilder {
.into(),
});
}
parent.add_computed_columns(&computed).await
parent.add_computed_columns(&computed, options).await
}
(None, true, Some((application, output_name))) => {
if read_columns.is_some() {
@@ -189,7 +223,7 @@ impl AddColumnsBuilder {
});
}
parent
.add_function_columns(&application, output_name.as_deref())
.add_function_columns(&application, output_name.as_deref(), options)
.await
}
_ => unreachable!("mixed add_columns modes were rejected above"),
@@ -201,6 +235,9 @@ impl AddColumnsBuilder {
mod tests {
use std::sync::Arc;
use super::ComputedColumnOptions;
use crate::table::computed_columns::AUTO_REFRESH_INTERVAL_META_KEY;
use arrow_array::{Int32Array, RecordBatch, record_batch};
use arrow_schema::{DataType, Field, Schema};
use lance::dataset::{BatchUDF, NewColumnTransform};
@@ -214,6 +251,93 @@ mod tests {
conn.create_table(name, batch).execute().await.unwrap()
}
#[tokio::test]
async fn computed_options_commit_with_declaration() {
let table = table_with_two_columns("options").await;
let version = table.version().await.unwrap();
let options = ComputedColumnOptions {
auto_refresh_enabled: false,
auto_refresh_interval_seconds: Some(300),
};
let result = table
.add_columns()
.computed("doubled", "x * 2")
.computed_options(options)
.execute()
.await
.unwrap();
assert_eq!(result.version, version + 1);
let schema = table.schema().await.unwrap();
assert_eq!(
ComputedColumnOptions::from_metadata(
schema.field_with_name("doubled").unwrap().metadata()
)
.unwrap(),
options
);
for interval in [0, u64::MAX] {
let invalid = ComputedColumnOptions {
auto_refresh_interval_seconds: Some(interval),
..options
};
assert!(
table
.add_columns()
.computed("bad", "x")
.computed_options(invalid)
.execute()
.await
.is_err()
);
}
assert!(
table
.add_columns()
.transform(NewColumnTransform::SqlExpressions(vec![(
"once".into(),
"x".into()
)]))
.computed_options(options)
.execute()
.await
.is_err()
);
assert!(
table
.schema()
.await
.unwrap()
.field_with_name("bad")
.is_err()
);
}
#[tokio::test]
async fn omitted_refresh_interval_stays_unset() {
let table = table_with_two_columns("server_interval").await;
let version = table.version().await.unwrap();
let result = table
.add_columns()
.computed("doubled", "x * 2")
.execute()
.await
.unwrap();
assert_eq!(result.version, version + 1);
let schema = table.schema().await.unwrap();
let metadata = schema.field_with_name("doubled").unwrap().metadata();
assert!(!metadata.contains_key(AUTO_REFRESH_INTERVAL_META_KEY));
assert_eq!(
ComputedColumnOptions::from_metadata(metadata).unwrap(),
ComputedColumnOptions::default()
);
assert!(
serde_json::to_value(ComputedColumnOptions::default())
.unwrap()
.get("auto_refresh_interval_seconds")
.is_none()
);
}
#[tokio::test]
async fn test_requires_a_transform() {
let table = table_with_two_columns("no_transform").await;
+100
View File
@@ -55,6 +55,106 @@ pub const EXPRESSION_META_KEY: &str = "computed_column.expression";
/// Field metadata key holding the column's inputs, as a JSON array of names.
pub const INPUTS_META_KEY: &str = "computed_column.inputs";
/// Whether the server may automatically refresh this output.
pub const AUTO_REFRESH_ENABLED_META_KEY: &str = "computed_column.auto_refresh_enabled";
/// Minimum interval between successful automatic refreshes, in seconds.
pub const AUTO_REFRESH_INTERVAL_META_KEY: &str = "computed_column.auto_refresh_interval_seconds";
/// Scheduling settings persisted on each computed output. Local tables store
/// these settings; automatic scheduling requires a Cloud or Enterprise server.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct ComputedColumnOptions {
/// Permit automatic refresh. Manual refresh is always available.
pub auto_refresh_enabled: bool,
/// Minimum seconds after a successful refresh before another is eligible.
/// When omitted, the server determines when to refresh.
#[serde(skip_serializing_if = "Option::is_none")]
pub auto_refresh_interval_seconds: Option<u64>,
}
impl Default for ComputedColumnOptions {
fn default() -> Self {
Self {
auto_refresh_enabled: true,
auto_refresh_interval_seconds: None,
}
}
}
impl ComputedColumnOptions {
/// Reject intervals that cannot be represented by the server scheduler.
pub fn validate(&self) -> Result<()> {
if self
.auto_refresh_interval_seconds
.is_some_and(|interval| interval == 0 || interval.checked_mul(1000).is_none())
{
return Err(Error::InvalidInput {
message: "auto_refresh_interval_seconds must be a positive integer no greater than u64::MAX / 1000".into(),
});
}
Ok(())
}
/// Read scheduling settings, using defaults for omitted properties.
pub fn from_metadata(metadata: &HashMap<String, String>) -> Result<Self> {
let mut options = Self::default();
if let Some(value) = metadata.get(AUTO_REFRESH_ENABLED_META_KEY) {
options.auto_refresh_enabled = value.parse().map_err(|_| Error::InvalidInput {
message: "auto_refresh_enabled must be true or false".into(),
})?;
}
if let Some(value) = metadata.get(AUTO_REFRESH_INTERVAL_META_KEY) {
options.auto_refresh_interval_seconds =
Some(value.parse().map_err(|_| Error::InvalidInput {
message: "auto_refresh_interval_seconds must be a positive integer".into(),
})?);
}
options.validate()?;
Ok(options)
}
/// Attach validated settings to newly declared computed output fields.
pub fn apply_to_schema(&self, schema: &ArrowSchema) -> Result<ArrowSchema> {
self.validate()?;
let fields = schema
.fields()
.iter()
.map(|field| {
if computed_column_from_field(field).is_none()
|| field
.metadata()
.get(FUNCTION_OUTPUT_ORDINAL_META_KEY)
.is_some_and(|ordinal| {
ordinal == &FUNCTION_ASSIGNMENT_OUTPUT_ORDINAL.to_string()
})
{
return field.clone();
}
let mut metadata = field.metadata().clone();
metadata.extend(self.metadata());
Arc::new(field.as_ref().clone().with_metadata(metadata))
})
.collect::<Vec<_>>();
Ok(ArrowSchema::new_with_metadata(
fields,
schema.metadata().clone(),
))
}
/// Field metadata values shared by native and server declaration paths.
pub fn metadata(&self) -> HashMap<String, String> {
let mut metadata = HashMap::from([(
AUTO_REFRESH_ENABLED_META_KEY.into(),
self.auto_refresh_enabled.to_string(),
)]);
if let Some(interval) = self.auto_refresh_interval_seconds {
metadata.insert(AUTO_REFRESH_INTERVAL_META_KEY.into(), interval.to_string());
}
metadata
}
}
/// Field metadata key holding the Function binding identity.
pub const FUNCTION_BINDING_ID_META_KEY: &str = "computed_column.function.binding_id";
@@ -13,6 +13,7 @@ use lance::dataset::transaction::{Operation, Transaction, UpdateMap, UpdateMapEn
use lance::dataset::{ColumnAlteration, CommitBuilder, Dataset, NewColumnTransform};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use super::computed_columns;
use super::{BaseTable, NativeTable};
@@ -128,6 +129,7 @@ pub(crate) async fn execute_add_columns(
pub(crate) async fn execute_declare(
table: &NativeTable,
columns: &[(String, String)],
options: computed_columns::ComputedColumnOptions,
) -> Result<AddColumnsResult> {
use lance::dataset::mem_wal::DatasetMemWalExt;
@@ -157,6 +159,10 @@ pub(crate) async fn execute_declare(
});
}
let transform = computed_columns::declare(table.schema().await?, columns)?;
let NewColumnTransform::AllNulls(schema) = transform else {
unreachable!("declarations are all-null")
};
let transform = NewColumnTransform::AllNulls(Arc::new(options.apply_to_schema(&schema)?));
commit_add_columns(table, transform, None).await
}