diff --git a/python/python/tests/test_materialized_views.py b/python/python/tests/test_materialized_views.py index 1edd10670..bd1b2a0bb 100644 --- a/python/python/tests/test_materialized_views.py +++ b/python/python/tests/test_materialized_views.py @@ -246,6 +246,31 @@ def test_legacy_storage_source_update_rebuilds(tmp_path): assert sorted(row["age"] for row in rows) == [8, 36] +def test_direct_writes_to_a_view_are_refused(tmp_path): + db = make_db(tmp_path) + view = db.create_materialized_view("copy", "people") + table = view.table + row = [{"name": "eve", "age": 1}] + refused = pytest.raises(NotImplementedError, match="is a materialized view") + + with refused: + table.add(row) + with refused: + table.update(where="name = 'kid'", values={"age": 8}) + with refused: + table.delete("age < 18") + with refused: + table.merge_insert("name").when_not_matched_insert_all().execute(row) + with refused: + table.add_columns({"older": "age + 1"}) + with refused: + table.drop_columns(["age"]) + assert table.count_rows() == 3 + + db.open_table("people").add(row) + assert view.refresh().rows_written == 4 + + def test_list_and_not_a_view(tmp_path): db = make_db(tmp_path) db.create_materialized_view("adults", "people", where="age >= 18") diff --git a/rust/lancedb/src/materialized_view.rs b/rust/lancedb/src/materialized_view.rs index 82e6aa802..c04a6cbbe 100644 --- a/rust/lancedb/src/materialized_view.rs +++ b/rust/lancedb/src/materialized_view.rs @@ -7,7 +7,8 @@ //! one source table and maintained by refresh rather than by writes. Creation //! commits an empty table carrying the defining query in schema metadata; a //! query this version cannot maintain reads back as unrefreshable, not as a -//! plain table. Queries, indexes and search work on the view unchanged. +//! plain table. Queries, indexes and search work on the view unchanged; +//! writes to its rows or stored columns are refused. mod grouped; pub use grouped::IVF_PARTITION; @@ -452,6 +453,24 @@ pub fn read_definition(metadata: &HashMap) -> Result, + operation: &str, +) -> Result<()> { + if metadata.contains_key(DEFINITION_META_KEY) { + return Err(Error::NotSupported { + message: format!( + "cannot {operation} '{name}': it is a materialized view, which only refresh writes" + ), + }); + } + Ok(()) +} + pub(crate) fn materialized_view_info_from_metadata( name: &str, metadata: &HashMap, @@ -1679,7 +1698,8 @@ impl MaterializedView { }) } - /// The view, as the table it is. Queries, indexes and search all apply. + /// The view, as the table it is. Queries, indexes and search all apply; + /// writes such as `add`, `update`, `delete` and `merge_insert` do not. pub fn table(&self) -> &Table { &self.table } @@ -1894,7 +1914,7 @@ impl Connection { #[cfg(test)] mod tests { - use arrow_array::record_batch; + use arrow_array::{RecordBatch, record_batch}; use super::*; use crate::connect; @@ -1978,4 +1998,248 @@ mod tests { assert_eq!(emptied.rows_written, 0); assert_eq!(view.table().count_rows(None).await.unwrap(), 0); } + + fn reader(batch: RecordBatch) -> Box { + let schema = batch.schema(); + Box::new(arrow_array::RecordBatchIterator::new( + vec![Ok(batch)], + schema, + )) + } + + fn assert_refused(result: Result, operation: &str) { + match result { + Err(Error::NotSupported { message }) => assert!( + message.contains("is a materialized view, which only refresh writes"), + "{operation}: {message}" + ), + other => panic!("{operation} on a view was not refused: {other:?}"), + } + } + + #[tokio::test] + async fn direct_writes_to_a_view_are_refused() { + use arrow_schema::Schema; + use datafusion::prelude::SessionContext; + use lance::dataset::{ColumnAlteration, NewColumnTransform}; + + use crate::table::AddDataMode; + use crate::table::datafusion::BaseTableAdapter; + + let conn = connect("memory://").execute().await.unwrap(); + let source = conn + .create_table("src", record_batch!(("x", Int32, [1, 2])).unwrap()) + .execute() + .await + .unwrap(); + let view = conn + .create_materialized_view("m", "src") + .execute() + .await + .unwrap(); + let downstream = conn + .create_materialized_view("m2", "m") + .execute() + .await + .unwrap(); + let table = view.table(); + let row = || record_batch!(("x", Int32, [99])).unwrap(); + + assert_refused(table.add(row()).execute().await, "add"); + assert_refused( + table + .add(row()) + .mode(AddDataMode::Overwrite) + .execute() + .await, + "overwrite", + ); + assert_refused( + table.update().column("x", "x + 1").execute().await, + "update", + ); + assert_refused(table.delete("x = 1").await, "delete"); + let mut merge = table.merge_insert(&["x"]); + merge.when_not_matched_insert_all(); + assert_refused(merge.execute(reader(row())).await, "merge_insert"); + assert_refused( + table + .add_columns() + .transform(NewColumnTransform::SqlExpressions(vec![( + "y".into(), + "x * 2".into(), + )])) + .execute() + .await, + "add_columns", + ); + assert_refused( + table + .add_columns() + .transform(NewColumnTransform::AllNulls(Arc::new(Schema::new(vec![ + ArrowField::new("y", DataType::Int32, true), + ])))) + .execute() + .await, + "add_columns (nulls)", + ); + assert_refused( + table + .alter_columns(&[ColumnAlteration::new("x".into()).rename("z".into())]) + .await, + "alter_columns", + ); + assert_refused(table.drop_columns(&["x"]).await, "drop_columns"); + assert_refused( + downstream.table().add(row()).execute().await, + "add to a chained view", + ); + + let ctx = SessionContext::new(); + let adapter = BaseTableAdapter::try_new(table.base_table().clone()) + .await + .unwrap(); + ctx.register_table("m", Arc::new(adapter)).unwrap(); + let insert = match ctx.sql("INSERT INTO m VALUES (99)").await { + Ok(frame) => frame.collect().await.map(|_| ()), + Err(error) => Err(error), + }; + let message = insert.unwrap_err().to_string(); + assert!(message.contains("is a materialized view"), "{message}"); + + assert_eq!(table.count_rows(None).await.unwrap(), 2); + assert_eq!(table.schema().await.unwrap().fields().len(), 1); + + // Refresh, chained, is the writer. + source.add(row()).execute().await.unwrap(); + assert_eq!(view.refresh().execute().await.unwrap().rows_written, 3); + assert_eq!( + downstream.refresh().execute().await.unwrap().rows_written, + 3 + ); + assert_eq!( + downstream + .table() + .count_rows(Some("x = 99".into())) + .await + .unwrap(), + 1 + ); + + // The source is a plain table and still takes every write. + source + .update() + .column("x", "x + 1") + .execute() + .await + .unwrap(); + source.delete("x = 2").await.unwrap(); + let mut merge = source.merge_insert(&["x"]); + merge.when_not_matched_insert_all(); + merge + .execute(reader(record_batch!(("x", Int32, [7])).unwrap())) + .await + .unwrap(); + source + .add_columns() + .transform(NewColumnTransform::SqlExpressions(vec![( + "y".into(), + "x * 2".into(), + )])) + .execute() + .await + .unwrap(); + source.drop_columns(&["y"]).await.unwrap(); + assert_eq!(view.refresh().execute().await.unwrap().rows_written, 3); + } + + #[tokio::test] + async fn indexes_and_optimize_still_apply_to_a_view() { + use crate::index::Index; + use crate::table::{CompactionOptions, OptimizeAction}; + + let conn = connect("memory://").execute().await.unwrap(); + let source = conn + .create_table("src", record_batch!(("x", Int32, [1, 2, 3])).unwrap()) + .execute() + .await + .unwrap(); + let view = conn + .create_materialized_view("m", "src") + .execute() + .await + .unwrap(); + view.table() + .create_index(&["x"], Index::Auto) + .execute() + .await + .unwrap(); + assert_eq!(view.table().list_indices().await.unwrap().len(), 1); + view.table() + .optimize(OptimizeAction::Compact { + options: CompactionOptions::default(), + remap_options: None, + }) + .await + .unwrap(); + view.table().optimize(OptimizeAction::All).await.unwrap(); + + source + .add(record_batch!(("x", Int32, [4])).unwrap()) + .execute() + .await + .unwrap(); + assert_eq!(view.refresh().execute().await.unwrap().rows_written, 4); + view.table().optimize(OptimizeAction::All).await.unwrap(); + assert_eq!( + view.table().count_rows(Some("x > 2".into())).await.unwrap(), + 2 + ); + } + + #[tokio::test] + async fn computed_columns_of_a_view_are_still_filled_and_dropped() { + use crate::table::computed_columns::{ + COMPUTED_COLUMN_META_KEY, EXPRESSION_META_KEY, INPUTS_META_KEY, KIND_META_KEY, SQL_KIND, + }; + + let conn = connect("memory://").execute().await.unwrap(); + let source = conn + .create_table("src", record_batch!(("x", Int32, [1, 2])).unwrap()) + .execute() + .await + .unwrap(); + let prepared = prepare_declaration(&source, Some(&[("x".into(), "x".into())]), None, None) + .await + .unwrap(); + let doubled = + ArrowField::new("doubled", 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(), "x * 2".into()), + (INPUTS_META_KEY.into(), "[\"x\"]".into()), + ])); + let view = prepared + .with_computed_columns(vec![(1, doubled)], &[]) + .unwrap() + .create("m") + .await + .unwrap(); + view.refresh().execute().await.unwrap(); + view.table().refresh_column("doubled").await.unwrap(); + assert_eq!( + view.table() + .count_rows(Some("doubled = x * 2".into())) + .await + .unwrap(), + 2 + ); + assert_refused( + view.table().drop_columns(&["x", "doubled"]).await, + "drop_columns of a stored and a computed column", + ); + view.table().drop_columns(&["doubled"]).await.unwrap(); + view.refresh().execute().await.unwrap(); + assert_eq!(view.table().count_rows(None).await.unwrap(), 2); + } } diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs index 5bc843f42..b296d135d 100644 --- a/rust/lancedb/src/table.rs +++ b/rust/lancedb/src/table.rs @@ -3120,6 +3120,16 @@ impl NativeTable { }) } + /// Refuse `operation` when this table is a materialized view. + pub(crate) async fn ensure_not_a_view(&self, operation: &str) -> Result<()> { + let dataset = self.dataset.get().await?; + crate::materialized_view::ensure_not_a_view( + &self.name, + &dataset.schema().metadata, + operation, + ) + } + /// Merge new data into this table. pub async fn merge( &mut self, @@ -3128,6 +3138,7 @@ impl NativeTable { right_on: &str, ) -> Result<()> { self.dataset.ensure_mutable()?; + self.ensure_not_a_view("merge into").await?; let mut dataset = (*self.dataset.get().await?).clone(); dataset.merge(batches, left_on, right_on).await?; self.dataset.update(dataset); @@ -3430,6 +3441,7 @@ impl BaseTable for NativeTable { self.dataset.ensure_mutable()?; let ds_wrapper = self.dataset.clone(); let ds = self.dataset.get().await?; + crate::materialized_view::ensure_not_a_view(&self.name, &ds.schema().metadata, "add to")?; let table_schema = Schema::from(&ds.schema().clone()); computed_columns::ensure_supported_function_metadata(&table_schema)?; @@ -3604,6 +3616,7 @@ impl BaseTable for NativeTable { new_data: Box, ) -> Result { let source_schema = arrow_array::RecordBatchReader::schema(&new_data); + self.ensure_not_a_view("merge insert into").await?; computed_columns::ensure_not_written( &Schema::from(self.dataset.get().await?.schema()), source_schema.fields().iter().map(|f| f.name().as_str()), @@ -4034,6 +4047,11 @@ impl BaseTable for NativeTable { write_params: WriteParams, ) -> Result> { let ds = self.dataset.get().await?; + crate::materialized_view::ensure_not_a_view( + &self.name, + &ds.schema().metadata, + "insert into", + )?; let dataset = Arc::new((*ds).clone()); Ok(Arc::new(datafusion::insert::InsertExec::new( self.dataset.clone(), diff --git a/rust/lancedb/src/table/delete.rs b/rust/lancedb/src/table/delete.rs index cb1da03ae..b9e598f3b 100644 --- a/rust/lancedb/src/table/delete.rs +++ b/rust/lancedb/src/table/delete.rs @@ -29,6 +29,7 @@ pub(crate) async fn execute_delete( predicate: Predicate<'_>, ) -> Result { table.dataset.ensure_mutable()?; + table.ensure_not_a_view("delete from").await?; match predicate { Predicate::String(s) => { let predicate = crate::expr::canonicalize_sql_predicate(s)?; diff --git a/rust/lancedb/src/table/schema_evolution.rs b/rust/lancedb/src/table/schema_evolution.rs index 4990901ae..4f229835b 100644 --- a/rust/lancedb/src/table/schema_evolution.rs +++ b/rust/lancedb/src/table/schema_evolution.rs @@ -104,6 +104,7 @@ pub(crate) async fn execute_add_columns( transforms: NewColumnTransform, read_columns: Option>, ) -> Result { + table.ensure_not_a_view("add columns to").await?; computed_columns::ensure_not_function_bound( table.schema().await?.as_ref(), "schema evolution", @@ -202,6 +203,15 @@ pub(crate) async fn execute_alter_columns( table.dataset.ensure_mutable()?; let mut dataset = (*table.dataset.get().await?).clone(); let schema = std::sync::Arc::new(ArrowSchema::from(dataset.schema())); + ensure_only_computed_columns_of_a_view( + table, + &dataset, + &schema, + "alter columns of", + alterations + .iter() + .map(|alteration| alteration.path.as_str()), + )?; // A Function binding stores its columns' exact fields, nullability // included, so every alteration of one counts, and a rename's target too. computed_columns::ensure_not_function_bound( @@ -241,6 +251,13 @@ pub(crate) async fn execute_drop_columns( table.dataset.ensure_mutable()?; let mut dataset = (*table.dataset.get().await?).clone(); let schema = std::sync::Arc::new(ArrowSchema::from(dataset.schema())); + ensure_only_computed_columns_of_a_view( + table, + &dataset, + &schema, + "drop columns of", + columns.iter().copied(), + )?; let unbinding = computed_columns::plan_function_unbinding(schema.as_ref(), columns)?; let mut names = columns.iter().map(|c| c.to_string()).collect::>(); @@ -266,6 +283,30 @@ pub(crate) async fn execute_drop_columns( Ok(DropColumnsResult { version }) } +/// A view's stored columns are the definition's output, which refresh checks +/// against the view's schema; only its computed columns may be altered or +/// dropped. +fn ensure_only_computed_columns_of_a_view<'a>( + table: &NativeTable, + dataset: &Dataset, + schema: &ArrowSchema, + operation: &str, + paths: impl IntoIterator, +) -> Result<()> { + let metadata = &dataset.schema().metadata; + if !metadata.contains_key(crate::materialized_view::DEFINITION_META_KEY) { + return Ok(()); + } + let computed = computed_columns::computed_columns(schema); + if paths + .into_iter() + .all(|path| computed.iter().any(|column| column.name == path)) + { + return Ok(()); + } + crate::materialized_view::ensure_not_a_view(&table.name, metadata, operation) +} + /// Retire the bindings a drop covers, in the commit before it. /// /// A binding lives in three places -- the schema-metadata envelope, each diff --git a/rust/lancedb/src/table/update.rs b/rust/lancedb/src/table/update.rs index 5ffcb74a9..cec787ff8 100644 --- a/rust/lancedb/src/table/update.rs +++ b/rust/lancedb/src/table/update.rs @@ -93,6 +93,7 @@ pub(crate) async fn execute_update( // 1. Snapshot the current dataset let dataset = table.dataset.get().await?; + crate::materialized_view::ensure_not_a_view(&table.name, &dataset.schema().metadata, "update")?; super::computed_columns::ensure_no_function_bindings_for_mutation( &arrow_schema::Schema::from(dataset.schema()), "update",