diff --git a/rust/lancedb/src/materialized_view.rs b/rust/lancedb/src/materialized_view.rs index 78ac06a66..3f556a613 100644 --- a/rust/lancedb/src/materialized_view.rs +++ b/rust/lancedb/src/materialized_view.rs @@ -448,6 +448,12 @@ impl CreateMaterializedViewBuilder { ), }); } + refresh::ensure_no_mem_wal( + native.dataset.get().await?.as_ref(), + "source table", + &self.source, + ) + .await?; let source_schema = source.schema().await?; let (definition, mut fields) = plan( source_schema, diff --git a/rust/lancedb/src/materialized_view/refresh.rs b/rust/lancedb/src/materialized_view/refresh.rs index 4711f1249..a6ddb2ba4 100644 --- a/rust/lancedb/src/materialized_view/refresh.rs +++ b/rust/lancedb/src/materialized_view/refresh.rs @@ -38,6 +38,7 @@ use datafusion::physical_plan::SendableRecordBatchStream; use datafusion::physical_plan::stream::RecordBatchStreamAdapter; use futures::{StreamExt, TryStreamExt}; use lance::Dataset; +use lance::dataset::mem_wal::DatasetMemWalExt; use lance::dataset::transaction::{Operation, Transaction}; use lance::dataset::{CommitBuilder, InsertBuilder, WriteDestination, WriteMode, WriteParams}; use lance_core::ROW_ID; @@ -140,12 +141,14 @@ pub(crate) async fn execute_refresh( } }; let definition = &definition; + ensure_no_mem_wal(&view_ds, "materialized view", view.name()).await?; let source_ds = open_source(view, definition).await?; let source_ds = match pinned { Some(version) => source_ds.checkout_version(version).await?, None => source_ds, }; + ensure_no_mem_wal(&source_ds, "source table", &definition.source_table).await?; let source_version = source_ds.version().version; let source_ts = source_ds.manifest.timestamp_nanos; @@ -458,6 +461,26 @@ fn validate_inputs(source: &Dataset, definition: &MaterializedViewDefinition) -> Ok(()) } +/// Reject MemWAL/LSM state on a refresh participant: rows in un-compacted +/// tiers are visible to ordinary reads but not to the fragment-planned +/// refresh scan, so certifying either side would misrepresent the table. +/// An active write spec and retained rows (the catch-up flag outlives +/// unset) both disqualify. +pub(crate) async fn ensure_no_mem_wal(dataset: &Dataset, role: &str, name: &str) -> Result<()> { + let retained = dataset.manifest.reader_feature_flags + & lance_table::feature_flags::FLAG_MEM_WAL_INDEX_CATCHUP + != 0; + if retained || dataset.mem_wal_index_details().await?.is_some() { + return Err(Error::NotSupported { + message: format!( + "{role} '{name}' has an LSM write spec or retained un-compacted \ + rows: rows in un-compacted tiers are invisible to refresh" + ), + }); + } + Ok(()) +} + async fn open_source(view: &Table, definition: &MaterializedViewDefinition) -> Result { let database = view.database_opt().ok_or_else(|| Error::InvalidInput { message: "the view was not opened through a database connection".into(), @@ -1510,4 +1533,95 @@ mod tests { assert_eq!(result.rows_written, 2); assert_eq!(read(view.table(), "double value").await, vec![2, 4]); } + + /// MemWAL tiers are visible to reads but not to the refresh scan, so LSM + /// state disqualifies every participant: the source at create, either + /// side at refresh, and the view can never accept a spec. Retained + /// un-compacted rows (the catch-up flag outlives unset) count as state. + #[tokio::test] + async fn lsm_state_disqualifies_source_and_view() { + use crate::table::LsmWriteSpec; + use arrow_array::RecordBatchIterator; + + let tmp_dir = tempfile::tempdir().unwrap(); + let conn = connect(tmp_dir.path().to_str().unwrap()) + .execute() + .await + .unwrap(); + // Hand-rolled: the LSM primary key must be non-nullable, which + // record_batch! cannot express. + let schema = Arc::new(ArrowSchema::new(vec![ + arrow_schema::Field::new("id", arrow_schema::DataType::Int64, false), + arrow_schema::Field::new("x", arrow_schema::DataType::Int32, false), + ])); + let batch = RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new(arrow_array::Int64Array::from(vec![1, 2])) as _, + Arc::new(Int32Array::from(vec![1, 2])) as _, + ], + ) + .unwrap(); + let table = conn + .create_table("src", batch.clone()) + .write_options(crate::materialized_view::tests::stable_row_ids()) + .execute() + .await + .unwrap(); + table.set_unenforced_primary_key(["id"]).await.unwrap(); + table + .set_lsm_write_spec(LsmWriteSpec::unsharded()) + .await + .unwrap(); + + // An active-LSM source is refused at create. + let err = conn + .create_materialized_view("v", "src") + .execute() + .await + .unwrap_err(); + assert!(err.to_string().contains("un-compacted"), "{err}"); + + // Unset with nothing written clears the state: the view creates, + // and a spec can never be installed over it. + table.unset_lsm_write_spec().await.unwrap(); + let view = conn + .create_materialized_view("v", "src") + .execute() + .await + .unwrap(); + let err = view + .table() + .set_lsm_write_spec(LsmWriteSpec::unsharded()) + .await + .unwrap_err(); + assert!(err.to_string().contains("materialized view"), "{err}"); + + // A source that acquires a spec after create fails refresh. + table + .set_lsm_write_spec(LsmWriteSpec::unsharded()) + .await + .unwrap(); + let err = view.refresh().execute().await.unwrap_err(); + assert!(err.to_string().contains("source table 'src'"), "{err}"); + + // Retained rows outlive unset: write through the WAL, unset, and + // refresh still refuses on the catch-up flag. + table.require_mem_wal_index_catchup().await.unwrap(); + let mut merge = table.merge_insert(&["id"]); + merge + .when_matched_update_all(None) + .when_not_matched_insert_all() + .use_lsm(true); + merge + .execute(Box::new(RecordBatchIterator::new( + vec![Ok(batch.clone())], + batch.schema(), + ))) + .await + .unwrap(); + table.unset_lsm_write_spec().await.unwrap(); + let err = view.refresh().execute().await.unwrap_err(); + assert!(err.to_string().contains("source table 'src'"), "{err}"); + } } diff --git a/rust/lancedb/src/table/merge/lsm.rs b/rust/lancedb/src/table/merge/lsm.rs index 5751cd916..342f6ff06 100644 --- a/rust/lancedb/src/table/merge/lsm.rs +++ b/rust/lancedb/src/table/merge/lsm.rs @@ -104,6 +104,13 @@ pub(crate) async fn set_lsm_write_spec(table: &NativeTable, spec: LsmWriteSpec) .into(), }); } + if crate::materialized_view::materialized_view_kind(&dataset.schema().metadata)?.is_some() { + return Err(Error::NotSupported { + message: "an LSM write spec cannot be installed on a materialized view: \ + rows in un-compacted tiers are invisible to refresh" + .into(), + }); + } let mut builder = dataset.initialize_mem_wal(); let writer_config_defaults = match spec { LsmWriteSpec::Bucket {