diff --git a/docs/web/feature-engineering/refresh-column.mdx b/docs/web/feature-engineering/refresh-column.mdx index f4550ed24..ce36daa32 100644 --- a/docs/web/feature-engineering/refresh-column.mdx +++ b/docs/web/feature-engineering/refresh-column.mdx @@ -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} Enterprise In SQL a refresh can be scoped to a diff --git a/docs/web/sql/ddl/table.mdx b/docs/web/sql/ddl/table.mdx index 792568220..11e7a3900 100644 --- a/docs/web/sql/ddl/table.mdx +++ b/docs/web/sql/ddl/table.mdx @@ -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) ``` -`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 (...)`. */} diff --git a/python/python/lancedb/_lancedb.pyi b/python/python/lancedb/_lancedb.pyi index 5a613b210..89c69e721 100644 --- a/python/python/lancedb/_lancedb.pyi +++ b/python/python/lancedb/_lancedb.pyi @@ -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: ... diff --git a/python/python/lancedb/remote/table.py b/python/python/lancedb/remote/table.py index eec958799..96b988911 100644 --- a/python/python/lancedb/remote/table.py +++ b/python/python/lancedb/remote/table.py @@ -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)) diff --git a/python/python/lancedb/table.py b/python/python/lancedb/table.py index 60404b73c..d5872e6f0 100644 --- a/python/python/lancedb/table.py +++ b/python/python/lancedb/table.py @@ -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): diff --git a/python/python/tests/test_first_class_function_slice1.py b/python/python/tests/test_first_class_function_slice1.py index b08dadf36..06730a5f1 100644 --- a/python/python/tests/test_first_class_function_slice1.py +++ b/python/python/tests/test_first_class_function_slice1.py @@ -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 diff --git a/python/python/tests/test_table.py b/python/python/tests/test_table.py index 7ec2003b7..755e94d7f 100644 --- a/python/python/tests/test_table.py +++ b/python/python/tests/test_table.py @@ -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 diff --git a/python/src/table.rs b/python/src/table.rs index 6cea168ec..c1d23d151 100644 --- a/python/src/table.rs +++ b/python/src/table.rs @@ -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, + auto_refresh_interval_seconds: Option, ) -> PyResult> { 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, + auto_refresh_enabled: Option, + auto_refresh_interval_seconds: Option, ) -> PyResult> { 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)) }) } diff --git a/rust/lancedb/src/remote/table.rs b/rust/lancedb/src/remote/table.rs index 3325f6943..1c569f6e5 100644 --- a/rust/lancedb/src/remote/table.rs +++ b/rust/lancedb/src/remote/table.rs @@ -3497,7 +3497,11 @@ impl BaseTable for RemoteTable { Ok(result) } - async fn add_computed_columns(&self, columns: &[(String, String)]) -> Result { + async fn add_computed_columns( + &self, + columns: &[(String, String)], + options: crate::table::computed_columns::ComputedColumnOptions, + ) -> Result { self.check_mutable().await?; crate::table::computed_columns::ensure_not_function_bound( self.schema().await?.as_ref(), @@ -3516,7 +3520,10 @@ impl BaseTable for RemoteTable { }, ) .collect::>(); - 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 BaseTable for RemoteTable { &self, application: &crate::function::FunctionApplication, output_name: Option<&str>, + options: crate::table::computed_columns::ComputedColumnOptions, ) -> Result { self.check_mutable().await?; let schema = self.schema().await?; @@ -3570,6 +3578,7 @@ impl BaseTable for RemoteTable { .collect::>(); 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 BaseTable for RemoteTable { "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) diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs index d99926ae1..61a710dcd 100644 --- a/rust/lancedb/src/table.rs +++ b/rust/lancedb/src/table.rs @@ -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 { 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 { 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 { - let result = schema_evolution::execute_declare(self, columns).await?; + async fn add_computed_columns( + &self, + columns: &[(String, String)], + options: computed_columns::ComputedColumnOptions, + ) -> Result { + let result = schema_evolution::execute_declare(self, columns, options).await?; self.bump_freshness(); Ok(result) } diff --git a/rust/lancedb/src/table/add_columns.rs b/rust/lancedb/src/table/add_columns.rs index 635a99bea..4f9cbb772 100644 --- a/rust/lancedb/src/table/add_columns.rs +++ b/rust/lancedb/src/table/add_columns.rs @@ -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)>, read_columns: Option>, + options: Option, } 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 { 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; diff --git a/rust/lancedb/src/table/computed_columns.rs b/rust/lancedb/src/table/computed_columns.rs index 9a3922b2b..e86b350ee 100644 --- a/rust/lancedb/src/table/computed_columns.rs +++ b/rust/lancedb/src/table/computed_columns.rs @@ -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, +} + +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) -> Result { + 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 { + 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::>(); + Ok(ArrowSchema::new_with_metadata( + fields, + schema.metadata().clone(), + )) + } + + /// Field metadata values shared by native and server declaration paths. + pub fn metadata(&self) -> HashMap { + 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"; diff --git a/rust/lancedb/src/table/schema_evolution.rs b/rust/lancedb/src/table/schema_evolution.rs index c541a9a43..1ad532d0a 100644 --- a/rust/lancedb/src/table/schema_evolution.rs +++ b/rust/lancedb/src/table/schema_evolution.rs @@ -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 { 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 }