From e40e073a9d69f7f247075ccd7457fcc678080599 Mon Sep 17 00:00:00 2001 From: Wyatt Alt Date: Tue, 4 Aug 2026 11:25:45 -0700 Subject: [PATCH] feat(rust): fill computed columns with refresh_column Declaring a computed column stores its expression but computes nothing, so until now the column stayed null with no way to fill it. refresh_column evaluates the expression over the rows that still hold no value and commits the results: table.refresh_column("doubled").await? Rows without a value are the ones to fill, which also makes the operation idempotent and resumable after a failure: refreshing again picks up whatever did not land. A row whose expression evaluates to null is indistinguishable from an unfilled one and is recomputed, which costs work but cannot change the result. Values written after a refresh are reachable by the next one, which is the case that matters -- an expression column populated once and then appended to would otherwise read null for every later row forever. --- rust/lancedb/src/table.rs | 21 +++ rust/lancedb/src/table/refresh.rs | 205 ++++++++++++++++++++++++++++++ 2 files changed, 226 insertions(+) create mode 100644 rust/lancedb/src/table/refresh.rs diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs index 24cd151df..1cc511035 100644 --- a/rust/lancedb/src/table.rs +++ b/rust/lancedb/src/table.rs @@ -79,6 +79,7 @@ pub mod merge; pub mod optimize; mod primary_key; pub mod query; +pub mod refresh; pub mod schema_evolution; pub mod update; pub mod write_progress; @@ -100,6 +101,7 @@ pub use lance::dataset::scanner::DatasetRecordBatchStream; pub use lance_index::optimize::OptimizeOptions; pub use lsm_stats::{BucketStats, GenerationStats, LsmStats, MemtableStats}; pub use optimize::{CompactionOptions, OptimizeAction, OptimizeStats}; +pub use refresh::RefreshColumnResult; pub use schema_evolution::{ AddColumnsResult, AlterColumnsResult, DropColumnsResult, FieldMetadataUpdate, UpdateFieldMetadataResult, @@ -784,6 +786,14 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync { transforms: NewColumnTransform, read_columns: Option>, ) -> Result; + /// Fill a computed column's unfilled rows. + /// + /// The default returns `NotSupported`; Lance-backed tables override it. + async fn refresh_column(&self, _column: &str) -> Result { + Err(Error::NotSupported { + message: "refresh_column is not supported on this table type".into(), + }) + } /// Alter columns in the table. async fn alter_columns(&self, alterations: &[ColumnAlteration]) -> Result; /// Drop columns from the table. @@ -1676,6 +1686,11 @@ impl Table { AddColumnsBuilder::new(self.inner.clone()) } + /// Compute and store values for a computed column's unfilled rows. + pub async fn refresh_column(&self, column: impl AsRef) -> Result { + self.inner.refresh_column(column.as_ref()).await + } + /// Change a column's name or nullability. pub async fn alter_columns( &self, @@ -3343,6 +3358,12 @@ impl BaseTable for NativeTable { Ok(result) } + async fn refresh_column(&self, column: &str) -> Result { + let result = refresh::execute_refresh_column(self, column).await?; + self.bump_freshness(); + Ok(result) + } + async fn alter_columns(&self, alterations: &[ColumnAlteration]) -> Result { let result = schema_evolution::execute_alter_columns(self, alterations).await?; self.bump_freshness(); diff --git a/rust/lancedb/src/table/refresh.rs b/rust/lancedb/src/table/refresh.rs new file mode 100644 index 000000000..fcaeb0917 --- /dev/null +++ b/rust/lancedb/src/table/refresh.rs @@ -0,0 +1,205 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The LanceDB Authors + +//! Filling computed columns. + +use arrow_schema::Schema as ArrowSchema; +use lance::dataset::UpdateBuilder as LanceUpdateBuilder; +use serde::{Deserialize, Serialize}; + +use super::NativeTable; +use super::computed_columns::computed_column_from_field; +use crate::{Error, Result}; + +/// The result of refreshing a computed column. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)] +pub struct RefreshColumnResult { + /// Rows that had a value computed. + #[serde(default)] + pub rows_filled: u64, + /// The commit version associated with the operation. + #[serde(default)] + pub version: u64, +} + +/// Internal implementation of the refresh logic. +pub(crate) async fn execute_refresh_column( + table: &NativeTable, + column: &str, +) -> Result { + table.dataset.ensure_mutable()?; + let dataset = table.dataset.get().await?; + + let schema = ArrowSchema::from(dataset.schema()); + let field = schema + .field_with_name(column) + .map_err(|_| Error::ColumnNotFound { + name: column.to_string(), + })?; + let declaration = + computed_column_from_field(field).ok_or_else(|| Error::NotAComputedColumn { + name: column.to_string(), + })?; + + // Rows still holding no value are the ones to fill. A row whose expression + // evaluates to null is indistinguishable from an unfilled one and is + // recomputed, which costs work but cannot change the result. + let builder = LanceUpdateBuilder::new(dataset) + .update_where(&format!("{column} IS NULL"))? + .set(column, &declaration.expression)?; + let result = builder.build()?.execute().await?; + + let version = result.new_dataset.version().version; + table.dataset.update(result.new_dataset.as_ref().clone()); + Ok(RefreshColumnResult { + rows_filled: result.rows_updated, + version, + }) +} + +#[cfg(test)] +mod tests { + use arrow_array::{Int32Array, record_batch}; + use futures::TryStreamExt; + + use crate::connect; + use crate::query::{ExecutableQuery, QueryBase, Select}; + use crate::{Error, Result, Table}; + + async fn table_with(name: &str, values: Vec) -> Table { + let conn = connect("memory://").execute().await.unwrap(); + let batch = record_batch!(("x", Int32, values)).unwrap(); + conn.create_table(name, batch).execute().await.unwrap() + } + + async fn declare_doubled(table: &Table) -> Result { + Ok(table + .add_columns() + .computed("doubled", "x * 2") + .execute() + .await? + .version) + } + + async fn read(table: &Table, column: &str) -> Vec> { + let batches = table + .query() + .select(Select::columns(&[column])) + .execute() + .await + .unwrap() + .try_collect::>() + .await + .unwrap(); + let mut values: Vec> = batches + .iter() + .flat_map(|batch| { + batch[column] + .as_any() + .downcast_ref::() + .unwrap() + .iter() + .collect::>() + }) + .collect(); + values.sort(); + values + } + + async fn append(table: &Table, values: Vec) { + let batch = record_batch!(("x", Int32, values)).unwrap(); + table.add(batch).execute().await.unwrap(); + } + + #[tokio::test] + async fn test_refresh_fills_a_declared_column() { + let table = table_with("refresh_fills", vec![1, 2, 3]).await; + let declared = declare_doubled(&table).await.unwrap(); + assert_eq!(read(&table, "doubled").await, vec![None, None, None]); + + let result = table.refresh_column("doubled").await.unwrap(); + assert!(result.version > declared); + assert_eq!(result.rows_filled, 3); + assert_eq!( + read(&table, "doubled").await, + vec![Some(2), Some(4), Some(6)] + ); + } + + /// Values written after the last refresh must be reachable by another one. + #[tokio::test] + async fn test_refresh_fills_rows_appended_since_the_last_refresh() { + let table = table_with("refresh_appended", vec![1, 2]).await; + declare_doubled(&table).await.unwrap(); + table.refresh_column("doubled").await.unwrap(); + + append(&table, vec![5, 6]).await; + assert_eq!( + read(&table, "doubled").await, + vec![None, None, Some(2), Some(4)] + ); + + let result = table.refresh_column("doubled").await.unwrap(); + assert_eq!(result.rows_filled, 2); + assert_eq!( + read(&table, "doubled").await, + vec![Some(2), Some(4), Some(10), Some(12)] + ); + } + + #[tokio::test] + async fn test_refresh_with_nothing_to_fill() { + let table = table_with("refresh_noop", vec![1, 2, 3]).await; + declare_doubled(&table).await.unwrap(); + table.refresh_column("doubled").await.unwrap(); + + let again = table.refresh_column("doubled").await.unwrap(); + assert_eq!(again.rows_filled, 0); + assert_eq!( + read(&table, "doubled").await, + vec![Some(2), Some(4), Some(6)] + ); + } + + #[tokio::test] + async fn test_refresh_leaves_deleted_rows_alone() { + let table = table_with("refresh_deleted", vec![1, 2, 3, 4]).await; + declare_doubled(&table).await.unwrap(); + table.delete("x = 2").await.unwrap(); + + let result = table.refresh_column("doubled").await.unwrap(); + assert_eq!(result.rows_filled, 3); + assert_eq!( + read(&table, "doubled").await, + vec![Some(2), Some(6), Some(8)] + ); + } + + #[tokio::test] + async fn test_refresh_a_constant_expression() { + let table = table_with("refresh_constant", vec![1, 2, 3]).await; + table + .add_columns() + .computed("answer", "42") + .execute() + .await + .unwrap(); + + let result = table.refresh_column("answer").await.unwrap(); + assert_eq!(result.rows_filled, 3); + } + + #[tokio::test] + async fn test_refresh_rejects_a_plain_column() { + let table = table_with("refresh_plain", vec![1, 2, 3]).await; + let err = table.refresh_column("x").await.unwrap_err(); + assert!(matches!(err, Error::NotAComputedColumn { name } if name == "x")); + } + + #[tokio::test] + async fn test_refresh_rejects_an_unknown_column() { + let table = table_with("refresh_missing", vec![1, 2, 3]).await; + let err = table.refresh_column("nope").await.unwrap_err(); + assert!(matches!(err, Error::ColumnNotFound { name } if name == "nope")); + } +}