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

@@ -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<Vec<String>>,
     ) -> Result<AddColumnsResult>;
+    /// Fill a computed column's unfilled rows.
+    ///
+    /// The default returns `NotSupported`; Lance-backed tables override it.
+    async fn refresh_column(&self, _column: &str) -> Result<RefreshColumnResult> {
+        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<AlterColumnsResult>;
     /// 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<str>) -> Result<RefreshColumnResult> {
+        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<RefreshColumnResult> {
+        let result = refresh::execute_refresh_column(self, column).await?;
+        self.bump_freshness();
+        Ok(result)
+    }
+
     async fn alter_columns(&self, alterations: &[ColumnAlteration]) -> Result<AlterColumnsResult> {
         let result = schema_evolution::execute_alter_columns(self, alterations).await?;
         self.bump_freshness();

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<RefreshColumnResult> {
+    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<i32>) -> 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<u64> {
+        Ok(table
+            .add_columns()
+            .computed("doubled", "x * 2")
+            .execute()
+            .await?
+            .version)
+    }
+
+    async fn read(table: &Table, column: &str) -> Vec<Option<i32>> {
+        let batches = table
+            .query()
+            .select(Select::columns(&[column]))
+            .execute()
+            .await
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+        let mut values: Vec<Option<i32>> = batches
+            .iter()
+            .flat_map(|batch| {
+                batch[column]
+                    .as_any()
+                    .downcast_ref::<Int32Array>()
+                    .unwrap()
+                    .iter()
+                    .collect::<Vec<_>>()
+            })
+            .collect();
+        values.sort();
+        values
+    }
+
+    async fn append(table: &Table, values: Vec<i32>) {
+        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"));
+    }
+}