diff --git a/rust/lancedb/src/error.rs b/rust/lancedb/src/error.rs index 4a6e6d8d9..6bd1ffa2b 100644 --- a/rust/lancedb/src/error.rs +++ b/rust/lancedb/src/error.rs @@ -71,6 +71,14 @@ pub enum Error { IndexNotFound { name: String }, #[snafu(display("Embedding function '{name}' was not found. : {reason}"))] EmbeddingFunctionNotFound { name: String, reason: String }, + #[snafu(display("Column '{name}' was not found"))] + ColumnNotFound { name: String }, + #[snafu(display("Column '{name}' already exists"))] + ColumnAlreadyExists { name: String }, + #[snafu(display("Column '{name}' is not a computed column"))] + NotAComputedColumn { name: String }, + #[snafu(display("Invalid expression for column '{column}': {message}"))] + InvalidExpression { column: String, message: String }, #[snafu(display("Table '{name}' already exists"))] TableAlreadyExists { name: String }, diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs index d03ac823f..24cd151df 100644 --- a/rust/lancedb/src/table.rs +++ b/rust/lancedb/src/table.rs @@ -69,6 +69,7 @@ pub mod add_columns; mod add_data; pub mod branch_merge; pub mod checkpoint; +pub mod computed_columns; mod create_index; pub mod datafusion; pub(crate) mod dataset; @@ -91,6 +92,7 @@ pub use branch_merge::{ MergeBranchResult, MergeBranchStatus, MergePreview, RowCountSummary, }; pub use chrono::Duration; +pub use computed_columns::{ComputedColumn, computed_column_from_field, computed_columns}; pub use delete::DeleteResult; use futures::future::join_all; pub use lance::dataset::refs::{BranchContents, Ref, TagContents, Tags as LanceTags}; diff --git a/rust/lancedb/src/table/add_columns.rs b/rust/lancedb/src/table/add_columns.rs index 0d410cd04..8c95ef669 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; use super::schema_evolution::AddColumnsResult; use crate::{Error, Result}; @@ -15,6 +16,7 @@ use crate::{Error, Result}; pub struct AddColumnsBuilder { parent: Arc, transform: Option, + computed: Vec<(String, String)>, read_columns: Option>, } @@ -23,6 +25,7 @@ impl std::fmt::Debug for AddColumnsBuilder { f.debug_struct("AddColumnsBuilder") .field("parent", &self.parent) .field("has_transform", &self.transform.is_some()) + .field("computed", &self.computed) .field("read_columns", &self.read_columns) .finish() } @@ -33,19 +36,29 @@ impl AddColumnsBuilder { Self { parent, transform: None, + computed: Vec::new(), read_columns: None, } } - /// Set how the new columns' values are produced. Required. + /// Set how the new columns' values are produced. pub fn transform(mut self, transform: NewColumnTransform) -> Self { self.transform = Some(transform); self } + /// Add a column defined by `expression`, evaluated by a later refresh + /// rather than by this commit. Its type and inputs are derived from the + /// expression. + pub fn computed(mut self, name: impl Into, expression: impl Into) -> Self { + self.computed.push((name.into(), expression.into())); + self + } + /// Limit which existing columns a [`NewColumnTransform::BatchUDF`] mapper - /// receives. Every other transform determines what it reads, so setting - /// this alongside one is an error rather than a silent no-op. + /// receives. Every other transform, and a computed column, determines what + /// it reads, so setting this alongside one is an error rather than a silent + /// no-op. pub fn read_columns(mut self, columns: impl IntoIterator>) -> Self { self.read_columns = Some(columns.into_iter().map(Into::into).collect()); self @@ -56,24 +69,43 @@ impl AddColumnsBuilder { let Self { parent, transform, + computed, read_columns, } = self; - let Some(transform) = transform else { - return Err(Error::InvalidInput { - message: "add_columns requires a transform".into(), - }); - }; - - if read_columns.is_some() && !matches!(transform, NewColumnTransform::BatchUDF(_)) { - return Err(Error::InvalidInput { - message: "read_columns applies only to a BatchUDF transform; \ - every other transform determines what it reads" + match (transform, computed.is_empty()) { + (None, true) => Err(Error::InvalidInput { + message: "add_columns requires a transform or a computed column".into(), + }), + // The two commit through different transforms, so one call covering + // both would be two commits and could half-apply. + (Some(_), false) => Err(Error::InvalidInput { + message: "add_columns cannot mix a transform with computed columns; \ + they cannot be added atomically in one call" .into(), - }); + }), + (Some(transform), true) => { + if read_columns.is_some() && !matches!(transform, NewColumnTransform::BatchUDF(_)) { + return Err(Error::InvalidInput { + message: "read_columns applies only to a BatchUDF transform; \ + every other transform determines what it reads" + .into(), + }); + } + parent.add_columns(transform, read_columns).await + } + (None, false) => { + if read_columns.is_some() { + return Err(Error::InvalidInput { + message: "read_columns applies only to a BatchUDF transform; \ + a computed column's inputs come from its expression" + .into(), + }); + } + let transform = computed_columns::declare(parent.schema().await?, &computed)?; + parent.add_columns(transform, None).await + } } - - parent.add_columns(transform, read_columns).await } } @@ -85,8 +117,8 @@ mod tests { use arrow_schema::{DataType, Field, Schema}; use lance::dataset::{BatchUDF, NewColumnTransform}; - use crate::Table; use crate::connect; + use crate::{Error, Table}; async fn table_with_two_columns(name: &str) -> Table { let conn = connect("memory://").execute().await.unwrap(); @@ -98,10 +130,7 @@ mod tests { async fn test_requires_a_transform() { let table = table_with_two_columns("no_transform").await; let err = table.add_columns().execute().await.unwrap_err(); - assert!( - err.to_string().contains("requires a transform"), - "got: {err}" - ); + assert!(matches!(err, Error::InvalidInput { .. })); } #[tokio::test] @@ -117,7 +146,7 @@ mod tests { .execute() .await .unwrap_err(); - assert!(err.to_string().contains("BatchUDF"), "got: {err}"); + assert!(matches!(err, Error::InvalidInput { .. })); let schema = table.schema().await.unwrap(); assert!( @@ -126,6 +155,47 @@ mod tests { ); } + #[tokio::test] + async fn test_mixing_transform_and_computed_is_rejected() { + let table = table_with_two_columns("mixed_add").await; + let err = table + .add_columns() + .transform(NewColumnTransform::SqlExpressions(vec![( + "eager".into(), + "x * 2".into(), + )])) + .computed("lazy", "x * 3") + .execute() + .await + .unwrap_err(); + assert!(matches!(err, Error::InvalidInput { .. })); + + let schema = table.schema().await.unwrap(); + assert!(schema.field_with_name("eager").is_err()); + assert!(schema.field_with_name("lazy").is_err()); + } + + #[tokio::test] + async fn test_read_columns_with_computed_is_rejected() { + let table = table_with_two_columns("read_cols_computed").await; + let err = table + .add_columns() + .computed("doubled", "x * 2") + .read_columns(["x"]) + .execute() + .await + .unwrap_err(); + assert!(matches!(err, Error::InvalidInput { .. })); + assert!( + table + .schema() + .await + .unwrap() + .field_with_name("doubled") + .is_err() + ); + } + #[tokio::test] async fn test_read_columns_limits_what_a_batch_udf_sees() { let table = table_with_two_columns("read_cols_udf").await; diff --git a/rust/lancedb/src/table/computed_columns.rs b/rust/lancedb/src/table/computed_columns.rs new file mode 100644 index 000000000..eb80d2d1c --- /dev/null +++ b/rust/lancedb/src/table/computed_columns.rs @@ -0,0 +1,491 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The LanceDB Authors + +//! Expression-backed computed columns. +//! +//! A computed column is defined by a SQL expression rather than by values +//! supplied at write time. Declaring one commits the column carrying its +//! expression in field metadata but no data, so the cost does not scale with +//! the table; a later refresh fills the rows. +//! +//! The expression is the whole definition: both the result type and the input +//! columns are derived from it, so a caller writes neither. +//! +//! [`computed_columns`] and [`computed_column_from_field`] read declarations +//! back off a schema. + +use std::collections::HashMap; +use std::sync::Arc; + +use arrow_schema::{Field as ArrowField, Schema as ArrowSchema, SchemaRef}; +use lance::dataset::NewColumnTransform; +use lance_datafusion::planner::Planner; + +use crate::{Error, Result}; + +/// Field metadata key marking a column as computed. The value is `"true"`. +pub const COMPUTED_COLUMN_META_KEY: &str = "computed_column"; + +/// Field metadata key holding the SQL expression that defines the column. +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"; + +/// A computed column's declaration, as read back from field metadata. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ComputedColumn { + /// Name of the computed column. + pub name: String, + /// The SQL expression that defines it. + pub expression: String, + /// Columns the expression reads, parsed from it at declaration time. + pub inputs: Vec, +} + +/// Build the field metadata recording a binding. +fn computed_column_metadata(expression: &str, inputs: &[String]) -> HashMap { + HashMap::from([ + (COMPUTED_COLUMN_META_KEY.to_string(), "true".to_string()), + (EXPRESSION_META_KEY.to_string(), expression.to_string()), + ( + INPUTS_META_KEY.to_string(), + serde_json::to_string(inputs).unwrap_or_else(|_| "[]".to_string()), + ), + ]) +} + +/// Read a field's computed-column declaration, if it carries one. +/// +/// A field flagged computed but missing its expression is not a computed +/// column here: without the binding there is nothing to refresh from, so it is +/// reported as absent rather than as a half-formed declaration. +pub fn computed_column_from_field(field: &ArrowField) -> Option { + let metadata = field.metadata(); + if metadata.get(COMPUTED_COLUMN_META_KEY).map(String::as_str) != Some("true") { + return None; + } + let expression = metadata.get(EXPRESSION_META_KEY)?; + let inputs = metadata + .get(INPUTS_META_KEY) + .and_then(|raw| serde_json::from_str::>(raw).ok()) + .unwrap_or_default(); + Some(ComputedColumn { + name: field.name().clone(), + expression: expression.clone(), + inputs, + }) +} + +/// Read every computed-column declaration carried by `schema`, in field order. +/// +/// Introspection is a pure read of the schema the caller already holds, the +/// way a SQL catalog reports a generation expression as another column of +/// `information_schema.columns`. +pub fn computed_columns(schema: &ArrowSchema) -> Vec { + schema + .fields() + .iter() + .filter_map(|field| computed_column_from_field(field)) + .collect() +} + +/// Resolve `(name, expression)` pairs against `schema` into fields carrying +/// their bindings. +/// +/// Everything that can be known statically is checked here rather than at +/// refresh time: that the expression parses, that every column it reads +/// exists, and that the target name is free. A declaration that survives this +/// is one a refresh can always act on. +pub(crate) fn plan(schema: SchemaRef, columns: &[(String, String)]) -> Result> { + if columns.is_empty() { + return Err(Error::InvalidInput { + message: "at least one computed column is required".into(), + }); + } + + let planner = Planner::new(schema.clone()); + let mut fields = Vec::with_capacity(columns.len()); + let mut declared: Vec<&str> = Vec::with_capacity(columns.len()); + + for (name, expression) in columns { + if schema.field_with_name(name).is_ok() || declared.contains(&name.as_str()) { + return Err(Error::ColumnAlreadyExists { name: name.clone() }); + } + + let expr = planner + .parse_expr(expression) + .and_then(|expr| planner.optimize_expr(expr)) + .map_err(|e| Error::InvalidExpression { + column: name.clone(), + message: e.to_string(), + })?; + + let mut inputs = Planner::column_names_in_expr(&expr); + inputs.sort(); + inputs.dedup(); + + // Resolved here rather than left to the planner so an unknown column + // names itself in the error instead of surfacing as a plan failure. + let mut indices = Vec::with_capacity(inputs.len()); + for input in &inputs { + let index = schema + .index_of(input) + .map_err(|_| Error::InvalidExpression { + column: name.clone(), + message: format!("unknown column '{input}'"), + })?; + indices.push(index); + } + + // Physical expressions address columns by position, so the planner + // that types the expression has to be built on the projected schema + // the refresh will actually read. + let read_schema = + Arc::new( + schema + .project(&indices) + .map_err(|e| Error::InvalidExpression { + column: name.clone(), + message: e.to_string(), + })?, + ); + let physical = Planner::new(read_schema.clone()) + .create_physical_expr(&expr) + .map_err(|e| Error::InvalidExpression { + column: name.clone(), + message: e.to_string(), + })?; + let data_type = + physical + .data_type(read_schema.as_ref()) + .map_err(|e| Error::InvalidExpression { + column: name.clone(), + message: e.to_string(), + })?; + + // Declared columns start entirely null, so nullability is a property + // of the declaration rather than of what the expression yields. + fields.push( + ArrowField::new(name, data_type, true) + .with_metadata(computed_column_metadata(expression, &inputs)), + ); + declared.push(name); + } + + Ok(fields) +} + +/// Build the transform that declares `columns` against `schema`. +/// +/// An all-null column is how a binding with no values yet is carried into a +/// commit; that it is spelled `AllNulls` is a detail of the commit, not of the +/// column, which is why this is internal and +/// [`AddColumnsBuilder::computed`](super::AddColumnsBuilder::computed) is the +/// public way in. +pub(crate) fn declare( + schema: SchemaRef, + columns: &[(String, String)], +) -> Result { + let fields = plan(schema, columns)?; + Ok(NewColumnTransform::AllNulls(Arc::new(ArrowSchema::new( + fields, + )))) +} + +#[cfg(test)] +mod tests { + use arrow_array::record_batch; + use arrow_schema::DataType; + use futures::TryStreamExt; + + use super::*; + use crate::connect; + use crate::query::{ExecutableQuery, QueryBase, Select}; + use crate::{Error, Table}; + + async fn table_with_ints(name: &str) -> Table { + let conn = connect("memory://").execute().await.unwrap(); + let batch = record_batch!(("x", Int32, [1, 2, 3])).unwrap(); + conn.create_table(name, batch).execute().await.unwrap() + } + + /// Declare `columns` the way a caller would: plan the expressions, then + /// add them through the ordinary column API. + async fn add_computed(table: &Table, columns: &[(String, String)]) -> Result { + let mut builder = table.add_columns(); + for (name, expression) in columns { + builder = builder.computed(name, expression); + } + Ok(builder.execute().await?.version) + } + + async fn declared(table: &Table) -> Vec { + computed_columns(table.schema().await.unwrap().as_ref()) + } + + #[tokio::test] + async fn test_declare_infers_type_and_inputs() { + let table = table_with_ints("declare_infers").await; + let initial = table.version().await.unwrap(); + + let version = add_computed(&table, &[("doubled".into(), "x * 2".into())]) + .await + .unwrap(); + assert!(version > initial); + + let schema = table.schema().await.unwrap(); + let field = schema.field_with_name("doubled").unwrap(); + assert_eq!(field.data_type(), &DataType::Int32); + assert!(field.is_nullable()); + + assert_eq!( + declared(&table).await, + vec![ComputedColumn { + name: "doubled".into(), + expression: "x * 2".into(), + inputs: vec!["x".into()], + }] + ); + } + + /// The binding reaches the schema only if `AllNulls` carries per-field + /// metadata through the commit. The whole representation rests on it. + #[tokio::test] + async fn test_all_nulls_preserves_field_metadata() { + let table = table_with_ints("metadata_survives").await; + add_computed(&table, &[("doubled".into(), "x * 2".into())]) + .await + .unwrap(); + + let schema = table.schema().await.unwrap(); + let metadata = schema.field_with_name("doubled").unwrap().metadata(); + assert_eq!( + metadata.get(COMPUTED_COLUMN_META_KEY).map(String::as_str), + Some("true") + ); + assert_eq!( + metadata.get(EXPRESSION_META_KEY).map(String::as_str), + Some("x * 2") + ); + assert_eq!( + metadata.get(INPUTS_META_KEY).map(String::as_str), + Some(r#"["x"]"#) + ); + } + + #[tokio::test] + async fn test_declared_column_is_all_null() { + let table = table_with_ints("declare_is_null").await; + add_computed(&table, &[("doubled".into(), "x * 2".into())]) + .await + .unwrap(); + + let batches = table + .query() + .select(Select::columns(&["doubled"])) + .execute() + .await + .unwrap() + .try_collect::>() + .await + .unwrap(); + + let total: usize = batches.iter().map(|b| b.num_rows()).sum(); + assert_eq!(total, 3); + for batch in &batches { + assert_eq!(batch["doubled"].null_count(), batch.num_rows()); + } + } + + #[tokio::test] + async fn test_unknown_column_fails_at_declare_time() { + let table = table_with_ints("unknown_input").await; + let err = add_computed(&table, &[("bad".into(), "missing + 1".into())]) + .await + .unwrap_err(); + assert!(matches!(err, Error::InvalidExpression { column, .. } if column == "bad")); + + let schema = table.schema().await.unwrap(); + assert!(schema.field_with_name("bad").is_err()); + } + + #[tokio::test] + async fn test_unparsable_expression_fails_at_declare_time() { + let table = table_with_ints("bad_syntax").await; + let err = add_computed(&table, &[("bad".into(), "x *".into())]) + .await + .unwrap_err(); + assert!(matches!(err, Error::InvalidExpression { column, .. } if column == "bad")); + assert!( + table + .schema() + .await + .unwrap() + .field_with_name("bad") + .is_err() + ); + } + + /// A user-defined function is an expression like any other; only its + /// resolution is missing. When a registry-aware planner exists this + /// becomes a supported declaration rather than a new API. + #[tokio::test] + async fn test_unregistered_function_is_rejected_for_now() { + let table = table_with_ints("udf_not_yet").await; + let err = add_computed(&table, &[("vec".into(), "embed(x)".into())]) + .await + .unwrap_err(); + assert!(matches!(err, Error::InvalidExpression { column, .. } if column == "vec")); + assert!( + table + .schema() + .await + .unwrap() + .field_with_name("vec") + .is_err() + ); + } + + #[tokio::test] + async fn test_existing_column_name_is_rejected() { + let table = table_with_ints("name_taken").await; + let err = add_computed(&table, &[("x".into(), "x * 2".into())]) + .await + .unwrap_err(); + assert!(matches!(err, Error::ColumnAlreadyExists { name } if name == "x")); + assert!(declared(&table).await.is_empty()); + } + + #[tokio::test] + async fn test_constant_expression_needs_no_inputs() { + let table = table_with_ints("constant").await; + add_computed(&table, &[("answer".into(), "42".into())]) + .await + .unwrap(); + + let declared = declared(&table).await; + assert_eq!(declared.len(), 1); + assert!(declared[0].inputs.is_empty()); + } + + #[tokio::test] + async fn test_multiple_columns_in_one_commit() { + let table = table_with_ints("multi").await; + let initial = table.version().await.unwrap(); + + add_computed( + &table, + &[ + ("plus".into(), "x + 1".into()), + ("squared".into(), "x * x".into()), + ], + ) + .await + .unwrap(); + + assert_eq!(table.version().await.unwrap(), initial + 1); + let declared = declared(&table).await; + assert_eq!(declared.len(), 2); + assert_eq!(declared[0].name, "plus"); + assert_eq!(declared[1].name, "squared"); + } + + #[tokio::test] + async fn test_duplicate_declaration_in_one_call_is_rejected() { + let table = table_with_ints("dupe").await; + let err = add_computed( + &table, + &[ + ("dup".into(), "x + 1".into()), + ("dup".into(), "x + 2".into()), + ], + ) + .await + .unwrap_err(); + assert!(matches!(err, Error::ColumnAlreadyExists { name } if name == "dup")); + assert!(declared(&table).await.is_empty()); + } + + /// A column added by an ordinary transform is materialized, not bound, so + /// it carries no declaration to report. + #[tokio::test] + async fn test_ordinary_columns_are_not_reported_as_computed() { + let table = table_with_ints("plain").await; + assert!(declared(&table).await.is_empty()); + + table + .add_columns() + .transform(NewColumnTransform::SqlExpressions(vec![( + "eager".into(), + "x * 2".into(), + )])) + .execute() + .await + .unwrap(); + assert!(declared(&table).await.is_empty()); + } + + /// Built-in functions type the column the same way an operator does. + #[tokio::test] + async fn test_builtin_function_inference() { + let conn = connect("memory://").execute().await.unwrap(); + let batch = record_batch!(("name", Utf8, ["ada", "grace"]), ("n", Int32, [-1, 2])).unwrap(); + let table = conn + .create_table("builtins", batch) + .execute() + .await + .unwrap(); + + add_computed( + &table, + &[ + ("shout".into(), "upper(name)".into()), + ("width".into(), "length(name)".into()), + ("magnitude".into(), "abs(n)".into()), + ], + ) + .await + .unwrap(); + + let schema = table.schema().await.unwrap(); + assert_eq!( + schema.field_with_name("shout").unwrap().data_type(), + &DataType::Utf8 + ); + assert_eq!( + schema.field_with_name("magnitude").unwrap().data_type(), + &DataType::Int32 + ); + // length() returns a width-dependent integer type; assert it is one + // rather than pinning which. + assert!( + schema + .field_with_name("width") + .unwrap() + .data_type() + .is_integer() + ); + + let declared = declared(&table).await; + assert_eq!(declared.len(), 3); + assert_eq!(declared[0].inputs, vec!["name".to_string()]); + assert_eq!(declared[2].inputs, vec!["n".to_string()]); + } + + #[tokio::test] + async fn test_inputs_are_deduplicated_and_sorted() { + let conn = connect("memory://").execute().await.unwrap(); + let batch = record_batch!(("b", Int32, [1, 2]), ("a", Int32, [3, 4])).unwrap(); + let table = conn.create_table("dedupe", batch).execute().await.unwrap(); + + add_computed(&table, &[("total".into(), "b + a + b".into())]) + .await + .unwrap(); + + assert_eq!( + declared(&table).await[0].inputs, + vec!["a".to_string(), "b".to_string()] + ); + } +}