fix: refuse direct writes to materialized views (#4471)

Refuse user writes (add, insert, update, delete, merge_insert, column
changes) to a table carrying `mv.definition`, so a materialized view is
written only by refresh. Index, optimize, and computed-column operations
stay allowed. Sophon's REST write paths are handled separately in
ENT-2878.

Fixes ENT-2876.

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")

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<String, String>) -> Result<Option<Stor
     })))
 }
 
+/// Refuse a user write to `name` when it is a materialized view, whose rows
+/// only refresh writes. Any stored definition counts, even one this version
+/// cannot read.
+pub(crate) fn ensure_not_a_view(
+    name: &str,
+    metadata: &HashMap<String, String>,
+    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<String, String>,
@@ -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<dyn arrow_array::RecordBatchReader + Send> {
+        let schema = batch.schema();
+        Box::new(arrow_array::RecordBatchIterator::new(
+            vec![Ok(batch)],
+            schema,
+        ))
+    }
+
+    fn assert_refused<T: std::fmt::Debug>(result: Result<T>, 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);
+    }
 }

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<dyn RecordBatchReader + Send>,
     ) -> Result<MergeResult> {
         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<Arc<dyn datafusion_physical_plan::ExecutionPlan>> {
         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(),

rust/lancedb/src/table/delete.rs

@@ -29,6 +29,7 @@ pub(crate) async fn execute_delete(
     predicate: Predicate<'_>,
 ) -> Result<DeleteResult> {
     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)?;

rust/lancedb/src/table/schema_evolution.rs

@@ -104,6 +104,7 @@ pub(crate) async fn execute_add_columns(
     transforms: NewColumnTransform,
     read_columns: Option<Vec<String>>,
 ) -> Result<AddColumnsResult> {
+    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::<Vec<_>>();
@@ -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<Item = &'a str>,
+) -> 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

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",