feat: refresh materialized views (#4010)

A declared view holds no rows; refresh computes them. It pins one source
version, brings the view to exactly the definition's result at that
version,
and records the version as a watermark in the view's schema metadata.

It is incremental when it can reconcile what changed: appended rows are
computed and appended, and rows the source deleted or updated are found
by
the lance delta and evicted by their __source_row_id provenance, the
updated
ones recomputed in the same commit. Compaction rearranges rows without
changing
them, so its outputs cost nothing -- which is what keeps routine
background
compaction from rebuilding the view. A vacuumed watermark, a
delta the transaction-log walk cannot classify, a Legacy-storage source,
or
more staged ids than a fixed cap all fall back to a rebuild; rebuilding
an
indexed view swaps every fragment in one Update, so readers never see it
unindexed or empty.

Concurrent refreshes serialize at commit -- each carries the
same sentinel row id in its inserted-rows filter, so the loser lands
nothing. On the append path the watermark moves in a follow-up commit,
so a
crash between the two re-appends those rows. Bumps lance
to v11.0.0-beta.19 for the delta reader.

rust/lancedb/src/lib.rs

@@ -211,7 +211,9 @@ pub use function::FunctionVersion;
 pub use job::Job;
 use lance_index::vector::ApproxMode as LanceApproxMode;
 use lance_linalg::distance::DistanceType as LanceDistanceType;
-pub use materialized_view::{MaterializedView, MaterializedViewDefinition};
+pub use materialized_view::{
+    MaterializedView, MaterializedViewDefinition, RefreshMaterializedViewResult, RefreshMode,
+};
 /// Re-export of the [`metrics`](https://docs.rs/metrics) crate facade. Enable
 /// the `metrics` feature to publish LanceDB's internal metrics; install any
 /// `metrics`-compatible recorder to collect them. See also [`metrics_otel`] for

rust/lancedb/src/materialized_view.rs

@@ -9,6 +9,7 @@
 //! metadata; a kind added later reads back as unrefreshable, not as a plain
 //! table. Queries, indexes and search work on the view unchanged.
 
+pub mod refresh;
 use std::collections::HashMap;
 use std::sync::Arc;
 
@@ -27,6 +28,8 @@ use crate::table::refresh::quote_identifier;
 use crate::table::{ColumnDefinition, ColumnKind};
 use crate::{Error, Result};
 
+pub use refresh::{RefreshMaterializedViewResult, RefreshMode};
+
 /// Schema metadata key holding the view definition, as kind-tagged JSON.
 pub const DEFINITION_META_KEY: &str = "mv.definition";
 
@@ -736,6 +739,12 @@ pub async fn prepare_declaration(
             ),
         });
     }
+    refresh::ensure_no_mem_wal(
+        native.dataset.get().await?.as_ref(),
+        "source table",
+        resolved.name(),
+    )
+    .await?;
     let source_schema = resolved.schema().await?;
     let source_metadata = source_schema.metadata().clone();
     let (definition, mut fields, lineage) = plan(
@@ -909,6 +918,54 @@ impl MaterializedView {
     pub fn definition(&self) -> &MaterializedViewDefinition {
         &self.definition
     }
+
+    /// Recompute the view from its source.
+    ///
+    /// By default the refresh is incremental when the source's changes can be
+    /// reconciled into the view, and otherwise rebuilds; see
+    /// [`RefreshMaterializedViewBuilder`].
+    ///
+    /// ```no_run
+    /// # #![recursion_limit = "256"]
+    /// # use lancedb::materialized_view::MaterializedView;
+    /// # async fn refresh(view: &MaterializedView) -> Result<(), Box<dyn std::error::Error>> {
+    /// let result = view.refresh().execute().await?;
+    /// println!("{:?}: {} rows", result.mode, result.rows_written);
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub fn refresh(&self) -> RefreshMaterializedViewBuilder {
+        RefreshMaterializedViewBuilder {
+            view: self.clone(),
+            full: false,
+            source_version: None,
+        }
+    }
+}
+
+/// Builds a refresh. Created by [`MaterializedView::refresh`].
+pub struct RefreshMaterializedViewBuilder {
+    view: MaterializedView,
+    full: bool,
+    source_version: Option<u64>,
+}
+
+impl RefreshMaterializedViewBuilder {
+    /// Rebuild the view even where an incremental refresh would do.
+    pub fn full(mut self, full: bool) -> Self {
+        self.full = full;
+        self
+    }
+
+    /// Refresh to this source table version instead of the latest.
+    pub fn source_version(mut self, version: u64) -> Self {
+        self.source_version = Some(version);
+        self
+    }
+
+    pub async fn execute(self) -> Result<RefreshMaterializedViewResult> {
+        refresh::execute_refresh(&self.view.table, self.full, self.source_version).await
+    }
 }
 
 impl Connection {
@@ -918,6 +975,7 @@ impl Connection {
     /// metadata; refresh computes the rows. Local databases only.
     ///
     /// ```no_run
+    /// # #![recursion_limit = "256"]
     /// # use lancedb::Connection;
     /// # async fn create(conn: &Connection) -> Result<(), Box<dyn std::error::Error>> {
     /// let view = conn
@@ -926,7 +984,7 @@ impl Connection {
     ///     .only_if("age >= 18")
     ///     .execute()
     ///     .await?;
-    /// println!("{}", view.definition().source_table);
+    /// view.refresh().execute().await?;
     /// # Ok(())
     /// # }
     /// ```

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 {

rust/lancedb/src/table/refresh.rs

@@ -597,7 +597,7 @@ mod tests {
 
     /// A fragment spanning several scan batches exercises the streamed fill:
     /// the probe buffers only until the first gained value and the rest flows
-    /// through write_column a batch at a time.
+    /// through write_columns a batch at a time.
     #[tokio::test]
     async fn test_refresh_streams_a_multi_batch_fragment() {
         let values: Vec<i32> = (0..20_000).collect();