diff --git a/rust/lancedb/src/database/namespace.rs b/rust/lancedb/src/database/namespace.rs index 78641a8b3..9447b27ea 100644 --- a/rust/lancedb/src/database/namespace.rs +++ b/rust/lancedb/src/database/namespace.rs @@ -3,6 +3,7 @@ //! Namespace-based database implementation that delegates table management to lance-namespace +use lance_datafusion::utils::StreamingWriteSource; use std::collections::{HashMap, HashSet}; use std::sync::{Arc, Mutex}; @@ -304,6 +305,10 @@ impl Database for LanceNamespaceDatabase { } async fn create_table(&self, request: DbCreateTableRequest) -> Result> { + // Refuse a bad declaration before the namespace records a table. + crate::table::computed_columns::ensure_declarations_are_planned( + &request.data.arrow_schema(), + )?; let mut table_id = request.namespace_path.clone(); table_id.push(request.name.clone()); let mut existing_table = None; diff --git a/rust/lancedb/src/materialized_view.rs b/rust/lancedb/src/materialized_view.rs index ec77c1181..80546b4a4 100644 --- a/rust/lancedb/src/materialized_view.rs +++ b/rust/lancedb/src/materialized_view.rs @@ -27,7 +27,12 @@ use crate::connection::Connection; use crate::database::listing::OPT_NEW_TABLE_ENABLE_STABLE_ROW_IDS; use crate::database::{CreateTableRequest, Database, OpenTableRequest}; use crate::embeddings::EmbeddingDefinition; +use crate::function::FunctionBinding; use crate::table::Table; +use crate::table::computed_columns::{ + FUNCTION_BINDINGS_META_KEY, computed_column_from_field, computed_columns, + ensure_declarations_are_planned, function_bindings_metadata, +}; use crate::table::refresh::quote_identifier; use crate::table::{ColumnDefinition, ColumnKind}; use crate::{Error, Result}; @@ -119,6 +124,16 @@ pub struct MaterializedViewDefinition { pub inputs: Vec, } +/// Prefix of the internal columns holding source columns a computed column +/// reads without the view projecting them; see +/// [`PreparedDeclaration::input_column`]. +pub const INPUT_COLUMN_PREFIX: &str = "__input_"; + +/// The internal view column holding a copy of `source_column`. +pub fn input_column_name(source_column: &str) -> String { + format!("{INPUT_COLUMN_PREFIX}{source_column}") +} + /// A view definition as read back from schema metadata. Non-exhaustive so a /// kind added later is additive. #[derive(Debug, Clone, PartialEq, Eq)] @@ -192,7 +207,7 @@ pub(crate) fn plan( source_schema: SchemaRef, source_table: &str, source_namespace: &[String], - projections: &[(String, String)], + projections: Option<&[(String, String)]>, filter: Option<&str>, limit: Option, ) -> Result<(MaterializedViewDefinition, Vec, Lineage)> { @@ -205,17 +220,16 @@ pub(crate) fn plan( }, err => err, })?; - let projections: Vec<(String, String)> = if projections.is_empty() { - source_schema + let projections: Vec<(String, String)> = match projections { + Some(projections) => projections.to_vec(), + // `SELECT *`. A source that is itself a view carries its own + // provenance column; the new view records its own, not a copy. + None => source_schema .fields() .iter() - // A source that is itself a view carries its own provenance - // column; the new view records its own, not a copy. .filter(|f| f.name() != SOURCE_ROW_ID_COLUMN) .map(|f| (f.name().clone(), quote_identifier(f.name()))) - .collect() - } else { - projections.to_vec() + .collect(), }; // A scan takes the cap as i64. Rejecting it here keeps creation and @@ -288,9 +302,16 @@ pub(crate) fn plan( message: e.to_string(), })?; - // Always nullable: what a refresh appends must fit the declared field - // whatever nullability the evaluator reports for a given batch. - let mut field = ArrowField::new(output, data_type, true); + // A projected column keeps its nullability; a computed value is + // nullable whatever the evaluator reports for a given batch. + let nullable = match projected_path(&expr).as_deref() { + Some([column]) => source_schema + .field_with_name(column) + .map(|f| f.is_nullable()) + .unwrap_or(true), + _ => true, + }; + let mut field = ArrowField::new(output, data_type, nullable); // Identity projections keep descriptive field metadata (blob markers); // computed values carry none. Structural declarations never come along. if let Some(source_field) = projected_field(&expr, &source_schema) { @@ -627,6 +648,12 @@ fn project_schema(schema: &ArrowSchema, columns: &[String]) -> SchemaRef { pub struct PreparedDeclaration { schema: SchemaRef, definition: MaterializedViewDefinition, + /// The source schema and the projection lineage, for placing a computed + /// column's inputs; `internal_inputs` counts the projections + /// [`PreparedDeclaration::input_column`] added after the declared ones. + source_schema: SchemaRef, + lineage: Lineage, + internal_inputs: usize, /// The source's own database: the only place /// [`PreparedDeclaration::create`] will put the view, because refresh /// resolves the recorded source coordinate through the view's database. @@ -647,6 +674,196 @@ impl PreparedDeclaration { &self.definition } + /// The schema the view will have: the declared columns in order, any + /// internal projections added by [`PreparedDeclaration::input_column`], + /// then [`SOURCE_ROW_ID_COLUMN`]. + pub fn schema(&self) -> &SchemaRef { + &self.schema + } + + /// The view column that holds `source_column` for a computed column to + /// read: the column the view projects it to, if any, otherwise an + /// internal projection added here, named by [`input_column_name`]. + pub fn input_column(&mut self, source_column: &str) -> Result { + if let Some(output) = self.lineage.get(source_column).and_then(|o| o.first()) { + return Ok(output.clone()); + } + let name = input_column_name(source_column); + let field = self + .source_schema + .field_with_name(source_column) + .map_err(|_| Error::InvalidInput { + message: format!("the source has no column '{source_column}' to read"), + })?; + if self.schema.field_with_name(&name).is_ok() { + return Err(Error::ColumnAlreadyExists { name }); + } + let row_id = self.row_id_index()?; + let mut fields: Vec = self + .schema + .fields() + .iter() + .map(|f| f.as_ref().clone()) + .collect(); + fields.insert( + row_id, + without_declarations(&field.as_ref().clone().with_name(name.clone())), + ); + self.definition.projections.push(ViewProjection { + output: name.clone(), + expression: quote_identifier(source_column), + }); + self.definition.inputs.push(source_column.to_string()); + self.definition.inputs.sort(); + self.definition.inputs.dedup(); + self.lineage + .entry(source_column.to_string()) + .or_default() + .push(name.clone()); + self.internal_inputs += 1; + let mut metadata = self.schema.metadata().clone(); + rewrite_column_definitions(&mut metadata, self.schema.as_ref(), &fields)?; + metadata.insert( + DEFINITION_META_KEY.to_string(), + definition_to_metadata(&self.definition)?, + ); + self.schema = Arc::new(ArrowSchema::new_with_metadata(fields, metadata)); + Ok(name) + } + + /// Add computed columns, each at its position among the declared + /// columns, with the bindings any of them name. + /// + /// Refresh never computes such a column: every row it writes carries + /// NULL there, and the declaration's owner fills it, `refresh_column` + /// for a SQL declaration. A commit that fills only computed columns is + /// the one commit on a view refresh does not treat as drift. Declarations + /// are validated over the assembled schema, and read only columns the + /// view holds (see [`PreparedDeclaration::input_column`]). + /// + /// ```no_run + /// # #![recursion_limit = "256"] + /// # use std::collections::HashMap; + /// # use arrow_schema::{DataType, Field}; + /// # use lancedb::materialized_view::prepare_declaration; + /// # use lancedb::table::computed_columns::{ + /// # COMPUTED_COLUMN_META_KEY, EXPRESSION_META_KEY, INPUTS_META_KEY, KIND_META_KEY, SQL_KIND, + /// # }; + /// # async fn declare(source: &lancedb::Table) -> Result<(), Box> { + /// let mut prepared = prepare_declaration( + /// source, + /// Some(&[("id".into(), "id".into())]), + /// None, + /// None, + /// ) + /// .await?; + /// // `text` is not projected; the view holds it internally for the column to read. + /// let text = prepared.input_column("text")?; + /// let length = Field::new("length", DataType::Int32, true).with_metadata(HashMap::from([ + /// (COMPUTED_COLUMN_META_KEY.into(), "true".into()), + /// (KIND_META_KEY.into(), SQL_KIND.into()), + /// (EXPRESSION_META_KEY.into(), format!("length({text})")), + /// (INPUTS_META_KEY.into(), format!("[\"{text}\"]")), + /// ])); + /// let view = prepared + /// .with_computed_columns(vec![(1, length)], &[])? + /// .create("lengths") + /// .await?; + /// view.refresh().execute().await?; // rows land with `length` NULL + /// view.table().refresh_column("length").await?; // filled + /// # Ok(()) + /// # } + /// ``` + pub fn with_computed_columns( + mut self, + columns: Vec<(usize, ArrowField)>, + bindings: &[FunctionBinding], + ) -> Result { + let invalid = |message: String| Error::InvalidInput { message }; + if columns.is_empty() { + return Err(invalid("at least one computed column is needed".into())); + } + if !computed_columns(&self.schema).is_empty() { + return Err(invalid( + "computed columns were already declared on this view".into(), + )); + } + if self.definition.projections.is_empty() { + return Err(invalid( + "a view of computed columns alone must read at least one source column".into(), + )); + } + let visible_count = self.visible_count(); + let mut fields: Vec = self + .schema + .fields() + .iter() + .map(|f| f.as_ref().clone()) + .collect(); + let mut columns = columns; + columns.sort_by_key(|(position, _)| *position); + for (inserted, (position, field)) in columns.iter().enumerate() { + let name = field.name().as_str(); + if name == SOURCE_ROW_ID_COLUMN + || name == ROW_ID + || name.starts_with(INPUT_COLUMN_PREFIX) + { + return Err(invalid(format!("view column name '{name}' is reserved"))); + } + if fields.iter().any(|f| f.name() == name) { + return Err(Error::ColumnAlreadyExists { + name: name.to_string(), + }); + } + if !field.is_nullable() { + return Err(invalid(format!( + "computed column '{name}' must be nullable until a refresh fills it" + ))); + } + if computed_column_from_field(field).is_none() { + return Err(invalid(format!( + "column '{name}' does not carry a computed-column declaration" + ))); + } + let limit = visible_count + inserted; + if *position > limit { + return Err(invalid(format!( + "computed column '{name}' is placed at {position}, past the view's {limit} columns" + ))); + } + // Positions index the select list, which counts the computed + // columns already inserted before this one. + fields.insert(*position, field.clone()); + } + let mut metadata = self.schema.metadata().clone(); + if !bindings.is_empty() { + metadata.insert( + FUNCTION_BINDINGS_META_KEY.to_string(), + function_bindings_metadata(bindings)?, + ); + } + rewrite_column_definitions(&mut metadata, self.schema.as_ref(), &fields)?; + let schema = ArrowSchema::new_with_metadata(fields, metadata); + ensure_declarations_are_planned(&schema)?; + self.schema = Arc::new(schema); + Ok(self) + } + + fn row_id_index(&self) -> Result { + self.schema + .index_of(SOURCE_ROW_ID_COLUMN) + .map_err(|e| Error::Runtime { + message: e.to_string(), + }) + } + + /// Columns the declaration lists: everything before the internal + /// projections and the provenance column. + fn visible_count(&self) -> usize { + self.definition.projections.len() - self.internal_inputs + + computed_columns(&self.schema).len() + } + /// Create the view table and verify it, consuming the declaration. /// /// The view goes at the root of the source's own database, where refresh @@ -717,11 +934,49 @@ impl PreparedDeclaration { } } -/// Validate a view declaration against its live source and hold what its -/// creation needs. The declaration is canonicalized through the coordinate a -/// refresh will resolve -- name and namespace both -- so a handle that does -/// not resolve back to itself is rejected. Same creation-time checks as -/// [`Connection::create_materialized_view`]. +/// Column definitions are positional over the view schema: carry each +/// field's entry to its place in `fields`, physical for a field that had none. +fn rewrite_column_definitions( + metadata: &mut HashMap, + previous: &ArrowSchema, + fields: &[ArrowField], +) -> Result<()> { + let Some(raw) = metadata.get(COLUMN_DEFINITIONS_META_KEY).cloned() else { + return Ok(()); + }; + let definitions: Vec = + serde_json::from_str(&raw).map_err(|e| Error::Runtime { + message: format!("unreadable column definitions on the view: {e}"), + })?; + let by_name: HashMap<&str, &ColumnDefinition> = previous + .fields() + .iter() + .zip(&definitions) + .map(|(field, definition)| (field.name().as_str(), definition)) + .collect(); + let rewritten: Vec = fields + .iter() + .map(|field| { + by_name + .get(field.name().as_str()) + .map(|d| (*d).clone()) + .unwrap_or(ColumnDefinition { + kind: ColumnKind::Physical, + }) + }) + .collect(); + metadata.insert( + COLUMN_DEFINITIONS_META_KEY.to_string(), + serde_json::to_string(&rewritten).map_err(|e| Error::Runtime { + message: format!("failed to serialize column definitions: {e}"), + })?, + ); + Ok(()) +} + +/// `projections` of `None` selects every source column, as `SELECT *`; +/// `Some(&[])` declares no projected column, for a view of function +/// columns alone. /// /// ```no_run /// # #![recursion_limit = "256"] @@ -729,7 +984,7 @@ impl PreparedDeclaration { /// # async fn declare(source: &lancedb::Table) -> Result<(), Box> { /// let prepared = prepare_declaration( /// source, -/// &[("id".into(), "id".into()), ("double".into(), "value * 2".into())], +/// Some(&[("id".into(), "id".into()), ("double".into(), "value * 2".into())]), /// Some("value > 0"), /// None, /// ) @@ -740,7 +995,7 @@ impl PreparedDeclaration { /// ``` pub async fn prepare_declaration( source: &Table, - projections: &[(String, String)], + projections: Option<&[(String, String)]>, filter: Option<&str>, limit: Option, ) -> Result { @@ -806,6 +1061,17 @@ pub async fn prepare_declaration( resolved.name(), ) .await?; + // The internal-input prefix belongs to the declaration alone; the + // replan at refresh sees those projections and must accept them. + if let Some(reserved) = projections + .unwrap_or_default() + .iter() + .find(|(output, _)| output.starts_with(INPUT_COLUMN_PREFIX)) + { + return Err(Error::InvalidInput { + message: format!("view column name '{}' is reserved", reserved.0), + }); + } let source_schema = resolved.schema().await?; let source_metadata = source_schema.metadata().clone(); let (definition, mut fields, lineage) = plan( @@ -841,6 +1107,9 @@ pub async fn prepare_declaration( Ok(PreparedDeclaration { schema: Arc::new(ArrowSchema::new_with_metadata(fields, metadata)), definition, + source_schema, + lineage, + internal_inputs: 0, database, }) } @@ -944,7 +1213,7 @@ impl CreateMaterializedViewBuilder { .await?; let prepared = prepare_declaration( &source, - &self.projections, + (!self.projections.is_empty()).then_some(self.projections.as_slice()), self.filter.as_deref(), self.limit, ) @@ -2076,7 +2345,7 @@ mod tests { ("id".to_string(), "id".to_string()), ("double".to_string(), "value * 2".to_string()), ]; - let prepared = prepare_declaration(&source, &projections, Some("value > 0"), None) + let prepared = prepare_declaration(&source, Some(&projections), Some("value > 0"), None) .await .unwrap(); assert_eq!(prepared.definition().source_table, "src"); @@ -2091,7 +2360,7 @@ mod tests { // external creation path cannot skip the check. conn.create_table("plain", batch).execute().await.unwrap(); let plain = conn.open_table("plain").execute().await.unwrap(); - let err = prepare_declaration(&plain, &[], None, None) + let err = prepare_declaration(&plain, None, None, None) .await .unwrap_err(); assert!(err.to_string().contains("stable row ids"), "{err}"); @@ -2113,7 +2382,7 @@ mod tests { .execute() .await .unwrap(); - let err = prepare_declaration(&masquerade, &[], None, None) + let err = prepare_declaration(&masquerade, None, None, None) .await .unwrap_err(); assert!( @@ -2134,7 +2403,7 @@ mod tests { .execute() .await .unwrap(); - let err = prepare_declaration(&custom, &[], None, None) + let err = prepare_declaration(&custom, None, None, None) .await .unwrap_err(); assert!(err.to_string().contains("custom_loc"), "{err}"); @@ -2272,4 +2541,664 @@ mod tests { ); } } + + /// A binding as the server records it: one Utf8 input over `input` + /// bound to a nullable parameter, one Int32 output named `output`, with + /// the exact schemas the durable contract requires. + pub fn test_binding(binding_id: &str, input: &str, output: &str) -> FunctionBinding { + let input_schema = lance_namespace::schema::arrow_schema_to_json(&ArrowSchema::new(vec![ + ArrowField::new("text", DataType::Utf8, true), + ])) + .unwrap(); + let output_schema = lance_namespace::schema::arrow_schema_to_json(&ArrowSchema::new(vec![ + ArrowField::new(output, DataType::Int32, true), + ])) + .unwrap(); + let input_type = input_schema.fields[0].r#type.r#type.clone(); + let output_type = output_schema.fields[0].r#type.r#type.clone(); + FunctionBinding::from_json( + &serde_json::json!({ + "binding_id": binding_id, + "function": {"name": "embed", "version": "fv_test"}, + "inputs": [{ + "parameter": "text", "field_id": -1, "field_path": input, + "arrow_type": input_type, "nullable": true, + }], + "outputs": [{ + "result_field": "$value", "output_name": output, "output_field_id": -1, + "output_ordinal": 0, "arrow_type": output_type, "nullable": false, + }], + "input_schema": serde_json::to_value(input_schema).unwrap(), + "output_schema": serde_json::to_value(output_schema).unwrap(), + }) + .to_string(), + ) + .unwrap() + } + + /// A computed column as the server declares it on a table: bound to a + /// registered Function. + pub fn computed_field(name: &str, binding_id: &str, input: &str) -> ArrowField { + ArrowField::new(name, DataType::Int32, true).with_metadata( + crate::table::computed_columns::function_computed_column_metadata( + binding_id, + 0, + &[input.to_string()], + ), + ) + } + + pub async fn people(conn: &Connection) -> Table { + let batch = + record_batch!(("id", Int32, [1, 2, 3]), ("name", Utf8, ["a", "b", "c"])).unwrap(); + conn.create_table("people", batch) + .write_options(stable_row_ids()) + .execute() + .await + .unwrap() + } + + /// `people` with both columns non-nullable, for nullability cases. + pub async fn strict_people(conn: &Connection) -> Table { + let schema = Arc::new(ArrowSchema::new(vec![ + ArrowField::new("id", DataType::Int32, false), + ArrowField::new("name", DataType::Utf8, false), + ])); + let batch = arrow_array::RecordBatch::try_new( + schema, + vec![ + Arc::new(arrow_array::Int32Array::from(vec![1, 2, 3])), + Arc::new(arrow_array::StringArray::from(vec!["a", "b", "c"])), + ], + ) + .unwrap(); + conn.create_table("people", batch) + .write_options(stable_row_ids()) + .execute() + .await + .unwrap() + } + + async fn prepared_people(conn: &Connection) -> PreparedDeclaration { + let source = people(conn).await; + prepare_declaration( + &source, + Some(&[ + ("id".to_string(), "id".to_string()), + ("name".to_string(), "name".to_string()), + ]), + None, + None, + ) + .await + .unwrap() + } + + #[tokio::test] + async fn a_computed_column_is_declared_null_with_its_binding() { + let conn = connect("memory://").execute().await.unwrap(); + let view = prepared_people(&conn) + .await + .with_computed_columns( + vec![(2, computed_field("emb", "fb_1", "name"))], + &[test_binding("fb_1", "name", "emb")], + ) + .unwrap() + .create("v") + .await + .unwrap(); + + let schema = view.table().schema().await.unwrap(); + let names: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect(); + assert_eq!(names, ["id", "name", "emb", SOURCE_ROW_ID_COLUMN]); + let declared: Vec = computed_columns(&schema) + .into_iter() + .map(|c| c.name) + .collect(); + assert_eq!(declared, ["emb"]); + let bindings = crate::table::computed_columns::function_bindings(&schema).unwrap(); + assert_eq!(bindings.len(), 1); + assert_eq!(bindings[0].binding_id(), "fb_1"); + // The stored definition is the plain select it always was. + let stored: serde_json::Value = + serde_json::from_str(&schema.metadata()[DEFINITION_META_KEY]).unwrap(); + assert_eq!(stored["kind"], SELECT_KIND); + assert_eq!(view.definition().projections.len(), 2); + assert_eq!(view.table().count_rows(None).await.unwrap(), 0); + assert_eq!(conn.open_materialized_view("v").await.unwrap().name(), "v"); + } + + #[tokio::test] + async fn computed_column_declarations_are_validated() { + let conn = connect("memory://").execute().await.unwrap(); + let prepared = prepared_people(&conn).await; + let binding = test_binding("fb_1", "name", "emb"); + let fails = |prepared: PreparedDeclaration, + columns: Vec<(usize, ArrowField)>, + bindings: &[FunctionBinding]| { + prepared + .with_computed_columns(columns, bindings) + .err() + .map(|e| e.to_string()) + .expect("the declaration should be refused") + }; + let emb = |binding_id: &str| computed_field("emb", binding_id, "name"); + + let err = fails( + prepared.clone(), + vec![(2, emb("fb_1").with_nullable(false))], + std::slice::from_ref(&binding), + ); + assert!(err.contains("must be nullable"), "{err}"); + + let plain = ArrowField::new("emb", DataType::Int32, true); + let err = fails( + prepared.clone(), + vec![(2, plain)], + std::slice::from_ref(&binding), + ); + assert!( + err.contains("does not carry a computed-column declaration"), + "{err}" + ); + + // The rest is the computed-column contract: a binding the field does + // not name, an output the binding does not map to this field, an + // input the view does not hold. + let err = fails( + prepared.clone(), + vec![(2, emb("fb_other"))], + std::slice::from_ref(&binding), + ); + assert!(err.contains("does not match binding 'fb_1'"), "{err}"); + let err = fails( + prepared.clone(), + vec![(2, emb("fb_1"))], + &[test_binding("fb_1", "name", "different_output")], + ); + assert!(err.contains("different_output"), "{err}"); + let err = fails( + prepared.clone(), + vec![(2, emb("fb_1"))], + &[test_binding("fb_1", "bio", "emb")], + ); + assert!(err.contains("'bio'"), "{err}"); + + let err = fails( + prepared.clone(), + vec![(2, computed_field("name", "fb_1", "name"))], + &[test_binding("fb_1", "name", "name")], + ); + assert!(err.contains("already exists"), "{err}"); + let err = fails( + prepared.clone(), + vec![(2, computed_field(SOURCE_ROW_ID_COLUMN, "fb_1", "name"))], + &[test_binding("fb_1", "name", SOURCE_ROW_ID_COLUMN)], + ); + assert!(err.contains("reserved"), "{err}"); + let err = fails( + prepared.clone(), + vec![(7, emb("fb_1"))], + std::slice::from_ref(&binding), + ); + assert!( + err.contains("placed at 7, past the view's 2 columns"), + "{err}" + ); + let err = fails(prepared, Vec::new(), std::slice::from_ref(&binding)); + assert!(err.contains("at least one computed column"), "{err}"); + } + + /// A source column a computed column reads without the view projecting + /// it becomes an internal projection before the provenance column, with + /// the source's nullability; a projected column is read from its + /// projection. + #[tokio::test] + async fn an_unprojected_input_becomes_an_internal_projection() { + let conn = connect("memory://").execute().await.unwrap(); + let source = strict_people(&conn).await; + let mut prepared = prepare_declaration( + &source, + Some(&[("key".to_string(), "id".to_string())]), + None, + None, + ) + .await + .unwrap(); + assert_eq!(prepared.input_column("id").unwrap(), "key"); + assert_eq!(prepared.input_column("name").unwrap(), "__input_name"); + assert_eq!(prepared.input_column("name").unwrap(), "__input_name"); + let err = prepared.input_column("missing").unwrap_err().to_string(); + assert!(err.contains("no column 'missing'"), "{err}"); + + let view = prepared + .with_computed_columns( + vec![(1, computed_field("emb", "fb_1", "__input_name"))], + &[test_binding("fb_1", "__input_name", "emb")], + ) + .unwrap() + .create("v") + .await + .unwrap(); + let schema = view.table().schema().await.unwrap(); + let names: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect(); + assert_eq!(names, ["key", "emb", "__input_name", SOURCE_ROW_ID_COLUMN]); + let input = schema.field_with_name("__input_name").unwrap(); + assert_eq!(input.data_type(), &DataType::Utf8); + assert!( + !input.is_nullable(), + "the copy keeps the source's nullability" + ); + assert!(!schema.field_with_name("key").unwrap().is_nullable()); + let projections: Vec<(&str, &str)> = view + .definition() + .projections + .iter() + .map(|p| (p.output.as_str(), p.expression.as_str())) + .collect(); + assert_eq!(projections, [("key", "id"), ("__input_name", "`name`")]); + assert_eq!(view.definition().inputs, ["id", "name"]); + } + + /// Two outputs of one binding land at consecutive positions: each + /// insertion widens the range the next may take. + #[tokio::test] + async fn sibling_computed_columns_take_consecutive_positions() { + let conn = connect("memory://").execute().await.unwrap(); + let source = people(&conn).await; + let prepared = prepare_declaration( + &source, + Some(&[("id".to_string(), "id".to_string())]), + None, + None, + ) + .await + .unwrap(); + let metadata = |ordinal: u32| { + crate::table::computed_columns::function_computed_column_metadata( + "fb_pair", + ordinal, + &["id".to_string()], + ) + }; + let left = ArrowField::new("left", DataType::Int32, true).with_metadata(metadata(0)); + let right = ArrowField::new("right", DataType::Int32, true).with_metadata(metadata(1)); + let input_schema = lance_namespace::schema::arrow_schema_to_json(&ArrowSchema::new(vec![ + ArrowField::new("value", DataType::Int32, true), + ])) + .unwrap(); + let output_schema = lance_namespace::schema::arrow_schema_to_json(&ArrowSchema::new(vec![ + ArrowField::new("left", DataType::Int32, true), + ArrowField::new("right", DataType::Int32, true), + ])) + .unwrap(); + let int = input_schema.fields[0].r#type.r#type.clone(); + let binding = FunctionBinding::from_json( + &serde_json::json!({ + "binding_id": "fb_pair", + "function": {"name": "pair", "version": "fv_test"}, + "inputs": [{"parameter": "value", "field_id": -1, "field_path": "id", + "arrow_type": int, "nullable": true}], + "outputs": [ + {"result_field": "left", "output_name": "left", "output_field_id": -1, + "output_ordinal": 0, "arrow_type": int, "nullable": false}, + {"result_field": "right", "output_name": "right", "output_field_id": -1, + "output_ordinal": 1, "arrow_type": int, "nullable": false}, + ], + "input_schema": serde_json::to_value(input_schema).unwrap(), + "output_schema": serde_json::to_value(output_schema).unwrap(), + }) + .to_string(), + ) + .unwrap(); + let view = prepared + .with_computed_columns(vec![(1, left), (2, right)], &[binding]) + .unwrap() + .create("v") + .await + .unwrap(); + let names: Vec = view + .table() + .schema() + .await + .unwrap() + .fields() + .iter() + .map(|f| f.name().clone()) + .collect(); + assert_eq!(names, ["id", "left", "right", SOURCE_ROW_ID_COLUMN]); + } + + /// A SQL declaration as `add_columns().computed()` records it. + pub fn sql_field( + name: &str, + data_type: DataType, + expression: &str, + inputs: &str, + ) -> ArrowField { + use crate::table::computed_columns::{ + COMPUTED_COLUMN_META_KEY, EXPRESSION_META_KEY, INPUTS_META_KEY, KIND_META_KEY, SQL_KIND, + }; + ArrowField::new(name, data_type, true).with_metadata(HashMap::from([ + (COMPUTED_COLUMN_META_KEY.to_string(), "true".to_string()), + (KIND_META_KEY.to_string(), SQL_KIND.to_string()), + (EXPRESSION_META_KEY.to_string(), expression.to_string()), + (INPUTS_META_KEY.to_string(), inputs.to_string()), + ])) + } + + /// A SQL declaration is re-planned at admission: it must parse against + /// the view, yield the declared type, and read the inputs it declares. + #[tokio::test] + async fn a_sql_declaration_is_planned_at_admission() { + let conn = connect("memory://").execute().await.unwrap(); + let fails = |prepared: PreparedDeclaration, field: ArrowField| { + prepared + .with_computed_columns(vec![(1, field)], &[]) + .err() + .map(|e| e.to_string()) + .expect("the declaration should be refused") + }; + let prepared = prepared_people(&conn).await; + let err = fails( + prepared.clone(), + sql_field("bad", DataType::Int32, "missing + 1", r#"["missing"]"#), + ); + assert!(err.contains("missing"), "{err}"); + let err = fails( + prepared.clone(), + sql_field("wide", DataType::Int64, "id + 1", r#"["id"]"#), + ); + assert!( + err.contains("declared as Int64 but its expression yields Int32"), + "{err}" + ); + let err = fails( + prepared.clone(), + sql_field("lying", DataType::Int32, "id + 1", r#"["name"]"#), + ); + assert!(err.contains("declares inputs"), "{err}"); + + let view = prepared + .with_computed_columns( + vec![(1, sql_field("next", DataType::Int32, "id + 1", r#"["id"]"#))], + &[], + ) + .unwrap() + .create("v") + .await + .unwrap(); + let names: Vec = view + .table() + .schema() + .await + .unwrap() + .fields() + .iter() + .map(|f| f.name().clone()) + .collect(); + assert_eq!(names, ["id", "next", "name", SOURCE_ROW_ID_COLUMN]); + } + + /// Creation persists a declaration only when it re-plans and the data + /// carries no values for it, whichever door created the table. + #[tokio::test] + async fn a_created_table_cannot_carry_computed_values() { + let conn = connect("memory://").execute().await.unwrap(); + let schema = Arc::new(ArrowSchema::new(vec![ + ArrowField::new("x", DataType::Int32, false), + sql_field("forged", DataType::Int32, "x + 1", r#"["x"]"#), + ])); + let filled = arrow_array::RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new(arrow_array::Int32Array::from(vec![1])), + Arc::new(arrow_array::Int32Array::from(vec![999])), + ], + ) + .unwrap(); + let err = conn + .create_table("forged", filled) + .execute() + .await + .unwrap_err() + .to_string(); + assert!(err.contains("cannot be written directly"), "{err}"); + assert!( + !conn + .table_names() + .execute() + .await + .unwrap() + .contains(&"forged".to_string()) + ); + + let unfilled = arrow_array::RecordBatch::try_new( + schema, + vec![ + Arc::new(arrow_array::Int32Array::from(vec![1])), + Arc::new(arrow_array::Int32Array::new_null(1)), + ], + ) + .unwrap(); + let table = conn + .create_table("declared", unfilled) + .execute() + .await + .unwrap(); + assert_eq!(table.refresh_column("forged").await.unwrap().rows_filled, 1); + + // A declaration with only its marker is broken, not absent. + let half = ArrowField::new("half", DataType::Int32, true).with_metadata(HashMap::from([( + crate::table::computed_columns::COMPUTED_COLUMN_META_KEY.to_string(), + "true".to_string(), + )])); + let batch = arrow_array::RecordBatch::try_new( + Arc::new(ArrowSchema::new(vec![ + ArrowField::new("x", DataType::Int32, false), + half, + ])), + vec![ + Arc::new(arrow_array::Int32Array::from(vec![1])), + Arc::new(arrow_array::Int32Array::from(vec![999])), + ], + ) + .unwrap(); + let err = conn + .create_table("half", batch) + .execute() + .await + .unwrap_err() + .to_string(); + assert!( + err.contains("incomplete computed-column declaration"), + "{err}" + ); + + let bogus = Arc::new(ArrowSchema::new(vec![ + ArrowField::new("x", DataType::Int32, false), + sql_field("bad", DataType::Int32, "missing + 1", r#"["missing"]"#), + ])); + let batch = arrow_array::RecordBatch::new_empty(bogus); + let err = conn + .create_table("bogus", batch) + .execute() + .await + .unwrap_err() + .to_string(); + assert!(err.contains("missing"), "{err}"); + } + + /// The internal-input prefix is reserved for the declaration, like the + /// provenance column, so an alias cannot masquerade as an internal input. + #[tokio::test] + async fn the_internal_input_prefix_is_reserved() { + let conn = connect("memory://").execute().await.unwrap(); + let source = people(&conn).await; + let err = prepare_declaration( + &source, + Some(&[("__input_x".to_string(), "name".to_string())]), + None, + None, + ) + .await + .unwrap_err() + .to_string(); + assert!(err.contains("'__input_x' is reserved"), "{err}"); + } + + /// A computed column may not read another, through any path: a + /// Function bound to a child of a computed struct is refused like a SQL + /// declaration over it. + #[tokio::test] + async fn a_computed_column_cannot_read_a_computed_root() { + let conn = connect("memory://").execute().await.unwrap(); + let prepared = prepared_people(&conn).await; + let payload = sql_field( + "payload", + DataType::Struct(vec![ArrowField::new("value", DataType::Utf8, true)].into()), + "named_struct('value', name)", + r#"["name"]"#, + ); + let err = prepared + .with_computed_columns( + vec![ + (2, payload), + (3, computed_field("emb", "fb_dependent", "payload.value")), + ], + &[test_binding("fb_dependent", "payload.value", "emb")], + ) + .err() + .map(|e| e.to_string()) + .expect("a computed root as a Function input should be refused"); + assert!(err.contains("reads computed column 'payload'"), "{err}"); + } + + /// The root check uses the canonical path parser: a quoted top-level + /// name containing a dot is one root, not two segments. + #[tokio::test] + async fn a_quoted_computed_root_is_still_refused() { + let conn = connect("memory://").execute().await.unwrap(); + let prepared = prepared_people(&conn).await; + let payload = sql_field( + "payload.dot", + DataType::Struct(vec![ArrowField::new("value", DataType::Utf8, true)].into()), + "named_struct('value', name)", + r#"["name"]"#, + ); + let input = "`payload.dot`.value"; + let err = prepared + .with_computed_columns( + vec![ + (2, payload), + (3, computed_field("emb", "fb_dependent", input)), + ], + &[test_binding("fb_dependent", input, "emb")], + ) + .err() + .map(|e| e.to_string()) + .expect("a quoted computed root should be refused"); + assert!(err.contains("reads computed column 'payload.dot'"), "{err}"); + } + + /// Namespace-backed creation admits declarations by the same rule, and + /// refuses before the namespace records the table. + #[tokio::test] + async fn a_namespace_created_table_cannot_carry_computed_values() { + let tmp = tempfile::tempdir().unwrap(); + let mut properties = std::collections::HashMap::new(); + properties.insert("root".to_string(), tmp.path().to_str().unwrap().to_string()); + let conn = crate::connect_namespace("dir", properties) + .execute() + .await + .unwrap(); + let half = ArrowField::new("half", DataType::Int32, true).with_metadata(HashMap::from([( + crate::table::computed_columns::COMPUTED_COLUMN_META_KEY.to_string(), + "true".to_string(), + )])); + let batch = arrow_array::RecordBatch::try_new( + Arc::new(ArrowSchema::new(vec![ + ArrowField::new("id", DataType::Int32, false), + half, + ])), + vec![ + Arc::new(arrow_array::Int32Array::from(vec![1])), + Arc::new(arrow_array::Int32Array::from(vec![999])), + ], + ) + .unwrap(); + let err = conn + .create_table("malformed", batch) + .execute() + .await + .unwrap_err() + .to_string(); + assert!( + err.contains("incomplete computed-column declaration"), + "{err}" + ); + assert!(conn.table_names().execute().await.unwrap().is_empty()); + + let schema = Arc::new(ArrowSchema::new(vec![ + ArrowField::new("id", DataType::Int32, false), + sql_field("next", DataType::Int32, "id + 1", r#"["id"]"#), + ])); + let filled = arrow_array::RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new(arrow_array::Int32Array::from(vec![1])), + Arc::new(arrow_array::Int32Array::from(vec![999])), + ], + ) + .unwrap(); + let err = conn + .create_table("forged", filled) + .execute() + .await + .unwrap_err() + .to_string(); + assert!(err.contains("cannot be written directly"), "{err}"); + + let unfilled = arrow_array::RecordBatch::try_new( + schema, + vec![ + Arc::new(arrow_array::Int32Array::from(vec![1])), + Arc::new(arrow_array::Int32Array::new_null(1)), + ], + ) + .unwrap(); + let table = conn + .create_table("declared", unfilled) + .execute() + .await + .unwrap(); + assert_eq!(table.refresh_column("next").await.unwrap().rows_filled, 1); + } + + /// A projected column keeps its nullability; a computed value is + /// nullable. + #[tokio::test] + async fn an_identity_projection_keeps_source_nullability() { + let conn = connect("memory://").execute().await.unwrap(); + let source = strict_people(&conn).await; + let prepared = prepare_declaration( + &source, + Some(&[ + ("id".to_string(), "id".to_string()), + ("n".to_string(), "name".to_string()), + ("next".to_string(), "id + 1".to_string()), + ]), + None, + None, + ) + .await + .unwrap(); + let nullable: Vec = prepared + .schema() + .fields() + .iter() + .map(|f| f.is_nullable()) + .collect(); + assert_eq!(nullable, [false, false, true, false]); + } } diff --git a/rust/lancedb/src/materialized_view/refresh.rs b/rust/lancedb/src/materialized_view/refresh.rs index efddd1c7b..23bb51566 100644 --- a/rust/lancedb/src/materialized_view/refresh.rs +++ b/rust/lancedb/src/materialized_view/refresh.rs @@ -24,8 +24,8 @@ use std::time::{SystemTime, UNIX_EPOCH}; use arrow_array::cast::AsArray; use arrow_array::types::UInt64Type; -use arrow_array::{RecordBatch, UInt64Array}; -use arrow_schema::{Schema as ArrowSchema, SchemaRef}; +use arrow_array::{RecordBatch, UInt64Array, new_null_array}; +use arrow_schema::{FieldRef, Schema as ArrowSchema, SchemaRef}; use datafusion::common::ScalarValue; use datafusion::error::DataFusionError; use datafusion::physical_plan::SendableRecordBatchStream; @@ -34,7 +34,7 @@ use datafusion::prelude::{col, lit}; use futures::{StreamExt, TryStreamExt}; use lance::Dataset; use lance::dataset::mem_wal::DatasetMemWalExt; -use lance::dataset::transaction::{Operation, Transaction}; +use lance::dataset::transaction::{Operation, Transaction, UpdateMode}; use lance::dataset::write::delete::DeleteBuilder; use lance::dataset::write::merge_insert::inserted_rows::{ KeyExistenceFilter, KeyExistenceFilterBuilder, KeyValue, @@ -51,6 +51,9 @@ use super::{ definition_to_metadata, }; use crate::database::OpenTableRequest; +use crate::table::computed_columns::{ + computed_column_from_field, computed_columns, ensure_declarations_are_planned, +}; use crate::table::{NativeTable, NativeTableExt, Table}; use crate::{Error, Result}; @@ -167,30 +170,52 @@ pub(crate) async fn execute_refresh( .map(|p| (p.output.clone(), p.expression.clone())) .collect(); validate_inputs(&source_ds, definition)?; - let (replanned, mut planned_fields, _renames) = super::plan( + let (replanned, planned_fields, _renames) = super::plan( source_schema, &definition.source_table, &definition.source_namespace, - &projections, + Some(&projections), definition.filter.as_deref(), definition.limit, )?; + let mut planned_fields = planned_fields; planned_fields.push(arrow_schema::Field::new( SOURCE_ROW_ID_COLUMN, arrow_schema::DataType::UInt64, false, )); + // A computed column is not planned from the source: refresh writes it + // NULL and its declaration's owner fills it. Its declaration must still + // be complete, and it must be able to hold NULL. let physical = ArrowSchema::from(view_ds.schema()); - let planned_shape: Vec<_> = planned_fields - .iter() - .map(|f| (f.name().clone(), f.data_type().clone(), f.is_nullable())) - .collect(); - let physical_shape: Vec<_> = physical + let mut computed = computed_columns(&physical).into_iter().map(|c| c.name); + if let Some(name) = computed.by_ref().find(|name| { + physical + .field_with_name(name) + .is_ok_and(|f| !f.is_nullable()) + }) { + return Err(Error::Schema { + message: format!( + "computed column '{name}' of view '{}' cannot hold NULL; recreate the view", + view.name() + ), + }); + } + ensure_declarations_are_planned(&physical)?; + let physical_planned: Vec<&FieldRef> = physical .fields() .iter() - .map(|f| (f.name().clone(), f.data_type().clone(), f.is_nullable())) + .filter(|f| computed_column_from_field(f).is_none()) .collect(); - if planned_shape != physical_shape { + // A projected column that became nullable at the source still fits the + // view's nullable field; the reverse would not. + let matches = planned_fields.len() == physical_planned.len() + && planned_fields.iter().zip(&physical_planned).all(|(e, p)| { + e.name() == p.name() + && e.data_type() == p.data_type() + && (p.is_nullable() || !e.is_nullable()) + }); + if !matches { return Err(Error::Schema { message: format!( "the stored definition of view '{}' does not produce this \ @@ -229,11 +254,18 @@ pub(crate) async fn execute_refresh( .get(SOURCE_VERSION_TS_META_KEY) .and_then(|raw| raw.parse().ok()); // The watermark speaks only for the view state its refresh left behind; - // any other commit on the view since then is drift. - let view_intact = metadata + // any other commit on the view since then is drift, except a fill of its + // computed columns, which rewrites nothing refresh certifies. + let recorded_view_version = metadata .get(VIEW_VERSION_META_KEY) - .and_then(|raw| raw.parse::().ok()) - == Some(view_ds.version().version); + .and_then(|raw| raw.parse::().ok()); + let view_intact = match recorded_view_version { + Some(recorded) if recorded == view_ds.version().version => true, + Some(recorded) if recorded < view_ds.version().version => { + only_computed_rewrites_since(&view_ds, recorded).await? + } + _ => false, + }; if !full && watermark == Some(source_version) && view_intact && recorded_ts == Some(source_ts) { return Ok(RefreshMaterializedViewResult { @@ -1090,6 +1122,69 @@ struct RowScope { limit: Option, } +/// Whether every commit on the view after `recorded` is a fill of its +/// computed columns: a column rewrite or data replacement touching only +/// those fields and neither adding nor removing rows. A version whose +/// transaction cannot be read is not proven, so it counts as drift. +async fn only_computed_rewrites_since(view_ds: &Dataset, recorded: u64) -> Result { + // A fill may write any field under a computed column, so the whole + // subtree counts, not only the root. + let physical = ArrowSchema::from(view_ds.schema()); + fn subtree(field: &lance_core::datatypes::Field, ids: &mut Vec) { + ids.push(field.id as u32); + for child in &field.children { + subtree(child, ids); + } + } + let mut computed_fields = Vec::new(); + for column in computed_columns(&physical) { + if let Some(field) = view_ds.schema().field(&column.name) { + subtree(field, &mut computed_fields); + } + } + if computed_fields.is_empty() { + return Ok(false); + } + for version in recorded + 1..=view_ds.version().version { + let Some(transaction) = view_ds.read_transaction_by_version(version).await? else { + return Ok(false); + }; + let fill = match &transaction.operation { + Operation::Update { + removed_fragment_ids, + new_fragments, + fields_modified, + update_mode: Some(UpdateMode::RewriteColumns), + .. + } => { + removed_fragment_ids.is_empty() + && new_fragments.is_empty() + && !fields_modified.is_empty() + && fields_modified + .iter() + .all(|field| computed_fields.contains(field)) + } + // What `refresh_column` commits for a SQL declaration. + Operation::DataReplacement { replacements } => { + !replacements.is_empty() + && replacements.iter().all(|group| { + !group.1.fields.is_empty() + && group + .1 + .fields + .iter() + .all(|field| computed_fields.contains(&(*field as u32))) + }) + } + _ => false, + }; + if !fill { + return Ok(false); + } + } + Ok(true) +} + async fn compute_stream( source: &Dataset, definition: &MaterializedViewDefinition, @@ -1158,6 +1253,10 @@ async fn compute_stream( let batch = batch.map_err(|e| DataFusionError::External(Box::new(e)))?; let mut columns = Vec::with_capacity(out_schema.fields().len()); for field in out_schema.fields() { + if computed_column_from_field(field).is_some() { + columns.push(new_null_array(field.data_type(), batch.num_rows())); + continue; + } let name = if field.name() == SOURCE_ROW_ID_COLUMN { ROW_ID } else { @@ -2768,7 +2867,7 @@ mod tests { let (conn, source) = db_with_source(vec![1]).await; let prepared = crate::materialized_view::prepare_declaration( &source, - &[("x".into(), "x".into()), ("twice".into(), "x * 2".into())], + Some(&[("x".into(), "x".into()), ("twice".into(), "x * 2".into())]), None, None, ) @@ -3132,4 +3231,424 @@ mod tests { let err = view.refresh().execute().await.unwrap_err(); assert!(err.to_string().contains("source table 'src'"), "{err}"); } + + /// A view with a computed column, declared over `people` and refreshed. + async fn refreshed_computed_view(conn: &Connection) -> MaterializedView { + use crate::materialized_view::tests::{computed_field, people, test_binding}; + let source = people(conn).await; + let view = crate::materialized_view::prepare_declaration( + &source, + Some(&[ + ("id".to_string(), "id".to_string()), + ("name".to_string(), "name".to_string()), + ]), + None, + None, + ) + .await + .unwrap() + .with_computed_columns( + vec![(2, computed_field("emb", "fb_1", "name"))], + &[test_binding("fb_1", "name", "emb")], + ) + .unwrap() + .create("v") + .await + .unwrap(); + let result = view.refresh().execute().await.unwrap(); + assert_eq!(result.mode, RefreshMode::Rebuild); + view + } + + async fn unfilled(view: &MaterializedView) -> usize { + view.table() + .count_rows(Some("emb IS NULL".to_string())) + .await + .unwrap() + } + + async fn append_people(conn: &Connection, ids: Vec, names: Vec<&str>) { + let batch = record_batch!(("id", Int32, ids), ("name", Utf8, names)).unwrap(); + conn.open_table("people") + .execute() + .await + .unwrap() + .add(batch) + .execute() + .await + .unwrap(); + } + + /// Commit the fill job's shape on the view: a column rewrite of + /// `fields`, touching no rows. The data is left as it is; what matters + /// here is how the next refresh classifies the commit. + async fn commit_column_rewrite(view: &MaterializedView, fields: &[&str]) { + let native = view.table().as_native().unwrap(); + native.dataset.reload().await.unwrap(); + let dataset = native.dataset.get().await.unwrap().as_ref().clone(); + let fields_modified = fields + .iter() + .map(|name| dataset.schema().field(name).unwrap().id as u32) + .collect(); + let updated_fragments = dataset + .get_fragments() + .iter() + .map(|fragment| fragment.metadata().clone()) + .collect(); + let operation = Operation::Update { + removed_fragment_ids: Vec::new(), + updated_fragments, + new_fragments: Vec::new(), + fields_modified, + compacted_sstables: Vec::new(), + fields_for_preserving_frag_bitmap: Vec::new(), + update_mode: Some(UpdateMode::RewriteColumns), + inserted_rows_filter: None, + updated_fragment_offsets: None, + }; + let read_version = dataset.version().version; + CommitBuilder::new(WriteDestination::Dataset(Arc::new(dataset))) + .execute(Transaction::new(read_version, operation, None)) + .await + .unwrap(); + } + + /// Refresh never computes a computed column: every row it writes, on a + /// rebuild, an append and a rewrite, carries NULL there, and the + /// declaration survives all three. + #[tokio::test] + async fn test_computed_columns_are_written_null_and_kept() { + let conn = connect("memory://").execute().await.unwrap(); + let view = refreshed_computed_view(&conn).await; + assert_eq!(unfilled(&view).await, 3); + + append_people(&conn, vec![4, 5], vec!["d", "e"]).await; + let result = view.refresh().execute().await.unwrap(); + assert_eq!(result.mode, RefreshMode::Incremental); + assert_eq!(unfilled(&view).await, 5); + + conn.open_table("people") + .execute() + .await + .unwrap() + .update() + .column("name", "'z'") + .only_if("id = 1") + .execute() + .await + .unwrap(); + view.refresh().execute().await.unwrap(); + assert_eq!(unfilled(&view).await, 5); + assert_eq!(read(view.table(), "id").await, vec![1, 2, 3, 4, 5]); + + let schema = view.table().schema().await.unwrap(); + assert!( + crate::table::computed_columns::function_bindings(&schema) + .unwrap() + .iter() + .any(|b| b.binding_id() == "fb_1"), + "the binding envelope was lost" + ); + assert!( + computed_column_from_field(schema.field_with_name("emb").unwrap()).is_some(), + "the declaration was lost" + ); + assert_eq!( + view.refresh().execute().await.unwrap().mode, + RefreshMode::NoOp + ); + } + + /// The fill job's commit rewrites only computed columns. It is the one + /// commit on a view that is not drift: the next refresh carries on from + /// its watermark instead of rebuilding, which would null what the fill + /// just wrote. + #[tokio::test] + async fn test_a_computed_column_fill_is_not_drift() { + let conn = connect("memory://").execute().await.unwrap(); + let view = refreshed_computed_view(&conn).await; + + commit_column_rewrite(&view, &["emb"]).await; + assert_eq!( + view.refresh().execute().await.unwrap().mode, + RefreshMode::NoOp + ); + + commit_column_rewrite(&view, &["emb"]).await; + append_people(&conn, vec![4], vec!["d"]).await; + let result = view.refresh().execute().await.unwrap(); + assert_eq!(result.mode, RefreshMode::Incremental); + assert_eq!(result.rows_written, 1); + assert_eq!(read(view.table(), "id").await, vec![1, 2, 3, 4]); + } + + /// A column rewrite that reaches a projected column is drift like any + /// other write: refresh certifies those columns and must recompute them. + #[tokio::test] + async fn test_a_rewrite_of_a_projected_column_is_drift() { + let conn = connect("memory://").execute().await.unwrap(); + let view = refreshed_computed_view(&conn).await; + + commit_column_rewrite(&view, &["emb", "name"]).await; + assert_eq!( + view.refresh().execute().await.unwrap().mode, + RefreshMode::Rebuild + ); + } + + /// The declaration contract is checked before any refresh mutation: a + /// missing binding envelope and a column that lost its declaration both + /// fail closed. + #[tokio::test] + async fn test_a_broken_declaration_is_refused_before_refresh() { + let conn = connect("memory://").execute().await.unwrap(); + let view = refreshed_computed_view(&conn).await; + let native = view.table().as_native().unwrap(); + let mut dataset = native.dataset.get().await.unwrap().as_ref().clone(); + dataset + .update_schema_metadata(vec![( + crate::table::computed_columns::FUNCTION_BINDINGS_META_KEY.to_string(), + None, + )]) + .await + .unwrap(); + let err = view.refresh().execute().await.unwrap_err().to_string(); + assert!(err.contains("references missing binding 'fb_1'"), "{err}"); + + let conn = connect("memory://").execute().await.unwrap(); + let view = refreshed_computed_view(&conn).await; + let native = view.table().as_native().unwrap(); + let mut dataset = native.dataset.get().await.unwrap().as_ref().clone(); + dataset + .replace_field_metadata(vec![( + dataset.schema().field("emb").unwrap().id as u32, + HashMap::new(), + )]) + .await + .unwrap(); + let err = view.refresh().execute().await.unwrap_err().to_string(); + assert!(err.contains("does not match binding 'fb_1'"), "{err}"); + } + + /// An input the view does not project is materialized on every refresh + /// path, before the provenance column, with the source's values. + #[tokio::test] + async fn test_internal_inputs_are_materialized_and_refreshed() { + use crate::materialized_view::tests::{computed_field, strict_people, test_binding}; + let conn = connect("memory://").execute().await.unwrap(); + let source = strict_people(&conn).await; + let mut prepared = crate::materialized_view::prepare_declaration( + &source, + Some(&[("id".to_string(), "id".to_string())]), + None, + None, + ) + .await + .unwrap(); + let input = prepared.input_column("name").unwrap(); + let view = prepared + .with_computed_columns( + vec![(1, computed_field("emb", "fb_1", &input))], + &[test_binding("fb_1", &input, "emb")], + ) + .unwrap() + .create("v") + .await + .unwrap(); + let names: Vec = view + .table() + .schema() + .await + .unwrap() + .fields() + .iter() + .map(|f| f.name().clone()) + .collect(); + assert_eq!(names, ["id", "emb", "__input_name", SOURCE_ROW_ID_COLUMN]); + + let unfilled_inputs = || async { + view.table() + .count_rows(Some("__input_name IS NULL".to_string())) + .await + .unwrap() + }; + assert_eq!( + view.refresh().execute().await.unwrap().mode, + RefreshMode::Rebuild + ); + assert_eq!(view.table().count_rows(None).await.unwrap(), 3); + assert_eq!(unfilled_inputs().await, 0); + + let more = arrow_array::RecordBatch::try_new( + source.schema().await.unwrap(), + vec![ + Arc::new(Int32Array::from(vec![4])), + Arc::new(arrow_array::StringArray::from(vec!["d"])), + ], + ) + .unwrap(); + source.add(more).execute().await.unwrap(); + assert_eq!( + view.refresh().execute().await.unwrap().mode, + RefreshMode::Incremental + ); + assert_eq!(unfilled_inputs().await, 0); + assert_eq!( + view.table() + .count_rows(Some("__input_name = 'd'".to_string())) + .await + .unwrap(), + 1 + ); + + source + .update() + .column("name", "'z'") + .only_if("id = 1") + .execute() + .await + .unwrap(); + view.refresh().execute().await.unwrap(); + assert_eq!( + view.table() + .count_rows(Some("__input_name = 'z'".to_string())) + .await + .unwrap(), + 1 + ); + assert_eq!( + unfilled(&view).await, + 4, + "rewritten and new rows are unfilled" + ); + } + + /// A SQL declaration is filled by `refresh_column` on the view, which + /// commits a data replacement; the next refresh continues from its + /// watermark and keeps what the fill wrote, and only rows the view added + /// since come back unfilled. + #[tokio::test] + async fn test_a_sql_fill_is_not_drift() { + use crate::materialized_view::tests::{people, sql_field}; + let conn = connect("memory://").execute().await.unwrap(); + let source = people(&conn).await; + let view = crate::materialized_view::prepare_declaration( + &source, + Some(&[("id".to_string(), "id".to_string())]), + None, + None, + ) + .await + .unwrap() + .with_computed_columns( + vec![( + 1, + sql_field("next", arrow_schema::DataType::Int32, "id + 1", r#"["id"]"#), + )], + &[], + ) + .unwrap() + .create("v") + .await + .unwrap(); + let filled = || async { + view.table() + .count_rows(Some("next = id + 1".to_string())) + .await + .unwrap() + }; + assert_eq!( + view.refresh().execute().await.unwrap().mode, + RefreshMode::Rebuild + ); + assert_eq!( + view.table() + .refresh_column("next") + .await + .unwrap() + .rows_filled, + 3 + ); + assert_eq!(filled().await, 3); + assert_eq!( + view.refresh().execute().await.unwrap().mode, + RefreshMode::NoOp + ); + assert_eq!(filled().await, 3); + + append_people(&conn, vec![4], vec!["d"]).await; + assert_eq!( + view.refresh().execute().await.unwrap().mode, + RefreshMode::Incremental + ); + assert_eq!(filled().await, 3); + assert_eq!( + view.table() + .refresh_column("next") + .await + .unwrap() + .rows_filled, + 1 + ); + assert_eq!(filled().await, 4); + assert_eq!( + view.refresh().execute().await.unwrap().mode, + RefreshMode::NoOp + ); + } + + /// A fill of a nested computed column writes its child fields; that is + /// still a fill, not drift. + #[tokio::test] + async fn test_a_nested_sql_fill_is_not_drift() { + use crate::materialized_view::tests::{people, sql_field}; + let conn = connect("memory://").execute().await.unwrap(); + let source = people(&conn).await; + let payload = sql_field( + "payload", + arrow_schema::DataType::Struct( + vec![arrow_schema::Field::new( + "value", + arrow_schema::DataType::Utf8, + true, + )] + .into(), + ), + "named_struct('value', name)", + r#"["name"]"#, + ); + let view = crate::materialized_view::prepare_declaration( + &source, + Some(&[("name".to_string(), "name".to_string())]), + None, + None, + ) + .await + .unwrap() + .with_computed_columns(vec![(1, payload)], &[]) + .unwrap() + .create("v") + .await + .unwrap(); + view.refresh().execute().await.unwrap(); + assert_eq!( + view.table() + .refresh_column("payload") + .await + .unwrap() + .rows_filled, + 3 + ); + assert_eq!( + view.refresh().execute().await.unwrap().mode, + RefreshMode::NoOp + ); + assert_eq!( + view.table() + .count_rows(Some("payload.value = name".to_string())) + .await + .unwrap(), + 3 + ); + } } diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs index 44e12d8ad..56d7ce518 100644 --- a/rust/lancedb/src/table.rs +++ b/rust/lancedb/src/table.rs @@ -2804,7 +2804,7 @@ impl NativeTable { namespace_client: Option>, pushdown_operations: HashSet, ) -> Result { - computed_columns::ensure_no_foreign_declarations(batches.arrow_schema().fields())?; + let batches = computed_columns::admit_create_source(batches)?; // Default params uses format v1. let params = params.unwrap_or(WriteParams { ..Default::default() @@ -2904,6 +2904,7 @@ impl NativeTable { pushdown_operations: HashSet, session: Option>, ) -> Result { + let batches = computed_columns::admit_create_source(batches)?; // Build table_id from namespace + name for the storage options provider let mut table_id = namespace.clone(); table_id.push(name.to_string()); diff --git a/rust/lancedb/src/table/computed_columns.rs b/rust/lancedb/src/table/computed_columns.rs index 21d4f3016..f1ec75213 100644 --- a/rust/lancedb/src/table/computed_columns.rs +++ b/rust/lancedb/src/table/computed_columns.rs @@ -21,6 +21,7 @@ //! [`computed_columns`] and [`computed_column_from_field`] read declarations //! back off a schema. +use futures::StreamExt; use std::collections::{BTreeSet, HashMap, HashSet}; use std::sync::Arc; @@ -1338,6 +1339,106 @@ pub(crate) fn ensure_batch_writes_no_computed_values( Ok(()) } +/// Validate every computed-column declaration `schema` carries against the +/// schema itself: every field with declaration metadata is a complete +/// declaration, a SQL declaration re-plans to the field it declares, a +/// Function declaration satisfies the binding contract, and no declaration +/// reads another computed column. What passes here is what `refresh_column` +/// can execute. +pub(crate) fn ensure_declarations_are_planned(schema: &ArrowSchema) -> Result<()> { + let invalid = |message: String| Error::InvalidInput { message }; + // A field with any declaration key is a declaration; a partial one is + // not "no declaration", it is a broken one. + for field in schema.fields() { + if field.metadata().keys().any(|k| is_declaration_key(k)) + && computed_column_from_field(field).is_none() + { + return Err(invalid(format!( + "field '{}' carries an incomplete computed-column declaration", + field.name() + ))); + } + } + let declared: HashSet = computed_columns(schema) + .into_iter() + .map(|c| c.name) + .collect(); + for column in computed_columns(schema) { + let field = schema.field_with_name(&column.name)?; + if !field.is_nullable() { + return Err(invalid(format!( + "computed column '{}' must be nullable until a refresh fills it", + column.name + ))); + } + match &column.kind { + ComputedColumnKind::Sql { expression } => { + let others: Vec = schema + .fields() + .iter() + .filter(|f| f.name() != &column.name) + .map(|f| f.as_ref().clone()) + .collect(); + let bound = bind(Arc::new(ArrowSchema::new(others)), &column.name, expression)?; + if let Some(input) = bound.roots.iter().find(|r| declared.contains(*r)) { + return Err(invalid(format!( + "computed column '{}' reads computed column '{input}'", + column.name + ))); + } + if &bound.data_type != field.data_type() { + return Err(invalid(format!( + "computed column '{}' is declared as {} but its expression yields {}", + column.name, + field.data_type(), + bound.data_type + ))); + } + let mut declared_inputs = column.inputs.clone(); + declared_inputs.sort(); + if declared_inputs != bound.inputs { + return Err(invalid(format!( + "computed column '{}' declares inputs {:?} but its expression reads {:?}", + column.name, declared_inputs, bound.inputs + ))); + } + } + ComputedColumnKind::Function { binding_id, .. } => { + // The binding validator resolves each input's leaf; the + // no-computed-input rule is about the root it hangs from. + let bindings = function_bindings(schema)?; + let Some(binding) = bindings.iter().find(|b| b.binding_id() == binding_id) else { + continue; // reported by the binding validator below + }; + // Roots come from the canonical path parser: a quoted + // top-level name may itself contain a dot. + if let Some(input) = binding + .inputs() + .iter() + .filter_map(|input| resolve_field_path(schema, &input.field_path).ok()) + .map(|resolved| resolved.root.name().as_str()) + .find(|r| declared.contains(*r)) + { + return Err(invalid(format!( + "computed column '{}' reads computed column '{input}'", + column.name + ))); + } + } + ComputedColumnKind::Unrecognized { kind } => { + return Err(Error::NotSupported { + message: format!( + "computed column '{}' is defined by '{kind}', which this version \ + of lancedb cannot fill", + column.name + ), + }); + } + } + } + ensure_supported_function_metadata(schema) +} + /// Reject fields carrying declaration metadata that did not come through /// [`plan`]. One authority for creation, overwrite and raw transforms. pub(crate) fn ensure_no_foreign_declarations<'a>( @@ -1796,6 +1897,54 @@ pub(super) async fn add_foreign_kind(table: &crate::Table, name: &str, kind: &st .unwrap(); } +/// Admit a table's initial data: every declaration it carries is validated, +/// and the stream refuses any batch with values in a computed column, whose +/// values come from refresh alone. One boundary for every way a table is +/// created. +pub(crate) fn admit_create_source( + batches: S, +) -> Result> { + let schema = batches.arrow_schema(); + ensure_declarations_are_planned(&schema)?; + let declared = computed_columns(&schema) + .into_iter() + .map(|c| c.name) + .collect(); + Ok(UnfilledDeclarations { + inner: batches, + declared, + }) +} + +/// A write source whose computed columns must arrive unfilled. +pub(crate) struct UnfilledDeclarations { + inner: S, + declared: Vec, +} + +impl lance_datafusion::utils::StreamingWriteSource + for UnfilledDeclarations +{ + fn arrow_schema(&self) -> SchemaRef { + self.inner.arrow_schema() + } + + fn into_stream(self) -> datafusion_physical_plan::SendableRecordBatchStream { + if self.declared.is_empty() { + return self.inner.into_stream(); + } + let schema = self.inner.arrow_schema(); + let declared = self.declared; + let stream = self.inner.into_stream().map(move |batch| { + let batch = batch?; + ensure_batch_writes_no_computed_values(&declared, &batch) + .map_err(|e| datafusion_common::DataFusionError::External(Box::new(e)))?; + Ok(batch) + }); + Box::pin(datafusion_physical_plan::stream::RecordBatchStreamAdapter::new(schema, stream)) + } +} + #[cfg(test)] mod tests { /// The gate's reproducer: the validator applies the same schema-level @@ -2646,6 +2795,8 @@ mod tests { ); } + /// A create carries a declaration only if it re-plans completely; this + /// one lacks its inputs and is refused before its forged value matters. #[tokio::test] async fn test_create_table_cannot_inject_a_declaration() { let conn = connect("memory://").execute().await.unwrap(); @@ -2673,7 +2824,7 @@ mod tests { .await .unwrap_err(); assert!( - matches!(&err, Error::InvalidInput { message } if message.contains("computed()")), + matches!(&err, Error::InvalidInput { message } if message.contains("computed column 'doubled'")), "{err:?}" ); }