feat: cascade option on materialized view refresh

A view over a view could only be brought fully up to date by refreshing
each view in the lineage by hand, in order. `cascade` on the refresh
builder, Python `refresh`/`refresh_async` and Node `refresh` does that in
one call: every upstream view is refreshed furthest first, each reading
the version the one before it left, then the named view. `full` applies
to the named view only.

Locally the lineage is walked through each view's source, and each hop is
refreshed against the incarnation read in that walk. Remote tables send
`cascade` in the refresh request, only when set, so older servers that
reject unknown fields are unaffected by plain refreshes.

docs/src/js/classes/MaterializedView.md

@@ -68,7 +68,9 @@ Recompute the view from its source.
 The refresh is incremental when the source's changes can be reconciled
 into the view -- rows added, changed or removed since the last one --
 and otherwise rebuilds. `full` forces a rebuild; `sourceVersion`
-refreshes to that source version instead of the latest.
+refreshes to that source version instead of the latest. `cascade` first
+refreshes every upstream view, furthest first, each reading the version
+the one before it left; `full` applies to this view only.
 
 Concurrent refreshes of one view do not duplicate its rows. Two that
 plan the same source rows conflict on commit, and the loser throws
@@ -78,6 +80,8 @@ rather than writing them a second time.
 
 * **options?**
 
+* **options.cascade?**: `boolean`
+
 * **options.full?**: `boolean`
 
 * **options.sourceVersion?**: `number`

nodejs/__test__/materialized_view.test.ts

@@ -112,6 +112,23 @@ describe("materialized views", () => {
     expect((await view.refresh()).mode).toBe("no_op");
   });
 
+  it("cascades a refresh through upstream views", async () => {
+    await db.createMaterializedView("adults", "people", {
+      where: "age >= 18",
+      withNoData: true,
+    });
+    const seniors = await db.createMaterializedView("seniors", "adults", {
+      where: "age >= 65",
+      withNoData: true,
+    });
+    expect(Number((await seniors.refresh()).rowsWritten)).toBe(0);
+
+    expect(Number((await seniors.refresh({ cascade: true })).rowsWritten)).toBe(
+      1,
+    );
+    expect(await (await db.openTable("adults")).countRows()).toBe(2);
+  });
+
   it("lists views and rejects non-views", async () => {
     await db.createMaterializedView("adults", "people", {
       where: "age >= 18",

nodejs/lancedb/materialized_view.ts

@@ -190,7 +190,9 @@ export class MaterializedView {
    * The refresh is incremental when the source's changes can be reconciled
    * into the view -- rows added, changed or removed since the last one --
    * and otherwise rebuilds. `full` forces a rebuild; `sourceVersion`
-   * refreshes to that source version instead of the latest.
+   * refreshes to that source version instead of the latest. `cascade` first
+   * refreshes every upstream view, furthest first, each reading the version
+   * the one before it left; `full` applies to this view only.
    *
    * Concurrent refreshes of one view do not duplicate its rows. Two that
    * plan the same source rows conflict on commit, and the loser throws
@@ -199,11 +201,13 @@ export class MaterializedView {
   async refresh(options?: {
     full?: boolean;
     sourceVersion?: number;
+    cascade?: boolean;
   }): Promise<RefreshMaterializedViewResult> {
     validateNonNegativeInteger(options?.sourceVersion, "sourceVersion");
     return await this.inner.refreshMaterializedView(
       options?.full,
       options?.sourceVersion,
+      options?.cascade,
     );
   }
 }

nodejs/lancedb/table.ts

@@ -670,6 +670,7 @@ export abstract class Table {
   abstract refreshMaterializedView(
     full?: boolean,
     sourceVersion?: number,
+    cascade?: boolean,
   ): Promise<RefreshMaterializedViewResult>;
 
   /** @ignore */
@@ -1397,8 +1398,13 @@ export class LocalTable extends Table {
   async refreshMaterializedView(
     full?: boolean,
     sourceVersion?: number,
+    cascade?: boolean,
   ): Promise<RefreshMaterializedViewResult> {
-    return await this.inner.refreshMaterializedView(full, sourceVersion);
+    return await this.inner.refreshMaterializedView(
+      full,
+      sourceVersion,
+      cascade,
+    );
   }
 
   async materializedViewDefinition(): Promise<string> {

nodejs/src/table.rs

@@ -458,11 +458,15 @@ impl Table {
         &self,
         full: Option<bool>,
         source_version: Option<i64>,
+        cascade: Option<bool>,
     ) -> napi::Result<RefreshMaterializedViewResult> {
         let view = lancedb::MaterializedView::from_table(self.inner_ref()?.clone())
             .await
             .default_error()?;
-        let mut builder = view.refresh().full(full.unwrap_or(false));
+        let mut builder = view
+            .refresh()
+            .full(full.unwrap_or(false))
+            .cascade(cascade.unwrap_or(false));
         if let Some(version) = source_version {
             let version = u64::try_from(version).map_err(|_| {
                 napi::Error::from_reason("sourceVersion must be a non-negative integer")

python/python/lancedb/_lancedb.pyi

@@ -488,10 +488,16 @@ class Table:
         limit: Optional[int] = None,
     ) -> FunctionErrors: ...
     async def refresh_materialized_view(
-        self, full: bool = False, source_version: Optional[int] = None
+        self,
+        full: bool = False,
+        source_version: Optional[int] = None,
+        cascade: bool = False,
     ) -> RefreshMaterializedViewResult: ...
     async def refresh_materialized_view_async(
-        self, full: bool = False, source_version: Optional[int] = None
+        self,
+        full: bool = False,
+        source_version: Optional[int] = None,
+        cascade: bool = False,
     ) -> Job: ...
     async def materialized_view_definition(self) -> str: ...
     async def add_columns_with_schema(self, schema: pa.Schema) -> AddColumnsResult: ...

python/python/lancedb/materialized_view.py

@@ -175,7 +175,11 @@ class AsyncMaterializedView:
         return _definition_from_json(raw)
 
     async def refresh(
-        self, *, full: bool = False, source_version: Optional[int] = None
+        self,
+        *,
+        full: bool = False,
+        source_version: Optional[int] = None,
+        cascade: bool = False,
     ) -> "RefreshMaterializedViewResult":
         """Recompute the view from its source.
 
@@ -185,16 +189,24 @@ class AsyncMaterializedView:
         ``source_version`` refreshes to that source version instead of the
         latest.
 
+        ``cascade=True`` first refreshes every upstream view, furthest first,
+        each reading the version the one before it left; ``full`` applies to
+        this view only.
+
         Concurrent refreshes of one view do not duplicate its rows. Two that
         plan the same source rows conflict on commit, and the loser raises
         rather than writing them a second time.
         """
         return await self._table._inner.refresh_materialized_view(
-            full=full, source_version=source_version
+            full=full, source_version=source_version, cascade=cascade
         )
 
     async def refresh_async(
-        self, *, full: bool = False, source_version: Optional[int] = None
+        self,
+        *,
+        full: bool = False,
+        source_version: Optional[int] = None,
+        cascade: bool = False,
     ) -> "AsyncJob[RefreshMaterializedViewResult]":
         """Submit a refresh and return its job without waiting.
 
@@ -206,7 +218,7 @@ class AsyncMaterializedView:
 
         return _typed_job(
             await self._table._inner.refresh_materialized_view_async(
-                full=full, source_version=source_version
+                full=full, source_version=source_version, cascade=cascade
             ),
             RefreshMaterializedViewResult.from_json,
         )
@@ -238,14 +250,26 @@ class MaterializedView:
         return LOOP.run(self._async.definition())
 
     def refresh(
-        self, *, full: bool = False, source_version: Optional[int] = None
+        self,
+        *,
+        full: bool = False,
+        source_version: Optional[int] = None,
+        cascade: bool = False,
     ) -> "RefreshMaterializedViewResult":
         """Recompute the view from its source. See
         [AsyncMaterializedView.refresh][lancedb.materialized_view.AsyncMaterializedView.refresh]."""
-        return LOOP.run(self._async.refresh(full=full, source_version=source_version))
+        return LOOP.run(
+            self._async.refresh(
+                full=full, source_version=source_version, cascade=cascade
+            )
+        )
 
     def refresh_async(
-        self, *, full: bool = False, source_version: Optional[int] = None
+        self,
+        *,
+        full: bool = False,
+        source_version: Optional[int] = None,
+        cascade: bool = False,
     ) -> "Job[RefreshMaterializedViewResult]":
         """Submit a refresh and return its job without waiting.
 
@@ -254,6 +278,8 @@ class MaterializedView:
         """
         return Job(
             LOOP.run(
-                self._async.refresh_async(full=full, source_version=source_version)
+                self._async.refresh_async(
+                    full=full, source_version=source_version, cascade=cascade
+                )
             )
         )

python/python/tests/test_materialized_views.py

@@ -211,6 +211,25 @@ def test_create_and_refresh_jobs(tmp_path):
     assert "adults" not in db.list_materialized_views()
 
 
+def test_cascade_refreshes_upstream_views(tmp_path):
+    db = make_db(tmp_path)
+    adults = db.create_materialized_view(
+        "adults", "people", where="age >= 18", with_no_data=True
+    )
+    seniors = db.create_materialized_view(
+        "seniors", "adults", where="age >= 65", with_no_data=True
+    )
+    assert seniors.refresh().rows_written == 0
+
+    assert seniors.refresh(cascade=True).rows_written == 1
+    assert db.open_table("adults").count_rows() == 2
+
+    db.open_table("people").add([{"name": "alan", "age": 90}])
+    assert seniors.refresh_async(cascade=True).wait().mode == "incremental"
+    assert seniors.table.count_rows() == 2
+    assert adults.refresh().mode == "no_op"
+
+
 def test_definition_round_trips(tmp_path):
     db = make_db(tmp_path)
     db.create_materialized_view("adults", "people", where="age >= 18")

python/src/table.rs

@@ -1802,18 +1802,19 @@ impl Table {
         })
     }
 
-    #[pyo3(signature = (full=false, source_version=None))]
+    #[pyo3(signature = (full=false, source_version=None, cascade=false))]
     pub fn refresh_materialized_view(
         self_: PyRef<'_, Self>,
         full: bool,
         source_version: Option<u64>,
+        cascade: bool,
     ) -> PyResult<Bound<'_, PyAny>> {
         let inner = self_.inner_ref()?.clone();
         future_into_py(self_.py(), async move {
             let view = lancedb::MaterializedView::from_table(inner)
                 .await
                 .infer_error()?;
-            let mut builder = view.refresh().full(full);
+            let mut builder = view.refresh().full(full).cascade(cascade);
             if let Some(version) = source_version {
                 builder = builder.source_version(version);
             }
@@ -1822,18 +1823,19 @@ impl Table {
         })
     }
 
-    #[pyo3(signature = (full=false, source_version=None))]
+    #[pyo3(signature = (full=false, source_version=None, cascade=false))]
     pub fn refresh_materialized_view_async(
         self_: PyRef<'_, Self>,
         full: bool,
         source_version: Option<u64>,
+        cascade: bool,
     ) -> PyResult<Bound<'_, PyAny>> {
         let inner = self_.inner_ref()?.clone();
         future_into_py(self_.py(), async move {
             let view = lancedb::MaterializedView::from_table(inner)
                 .await
                 .infer_error()?;
-            let mut builder = view.refresh().full(full);
+            let mut builder = view.refresh().full(full).cascade(cascade);
             if let Some(version) = source_version {
                 builder = builder.source_version(version);
             }

rust/lancedb/src/materialized_view.rs

@@ -1940,6 +1940,7 @@ impl MaterializedView {
             full: false,
             source_version: None,
             expected_incarnation: None,
+            cascade: false,
         }
     }
 }
@@ -1950,6 +1951,7 @@ pub struct RefreshMaterializedViewBuilder {
     full: bool,
     source_version: Option<u64>,
     expected_incarnation: Option<String>,
+    cascade: bool,
 }
 
 impl RefreshMaterializedViewBuilder {
@@ -1979,6 +1981,13 @@ impl RefreshMaterializedViewBuilder {
         self
     }
 
+    /// Refresh every upstream view first, furthest first, each reading the
+    /// version the one before it left. [`Self::full`] applies to this view only.
+    pub fn cascade(mut self, cascade: bool) -> Self {
+        self.cascade = cascade;
+        self
+    }
+
     /// Submit the refresh and return a job that settles with its result.
     pub async fn execute_async(self) -> Result<Job<RefreshMaterializedViewResult>> {
         if self.view.table.as_native().is_none() {
@@ -1990,33 +1999,32 @@ impl RefreshMaterializedViewBuilder {
                     self.full,
                     self.source_version,
                     self.expected_incarnation.as_deref(),
+                    self.cascade,
                 )
                 .await;
         }
         Ok(Job::spawned(tokio::spawn(async move {
-            refresh::execute_refresh(
-                &self.view.table,
-                self.full,
-                self.source_version,
-                self.expected_incarnation.as_deref(),
-            )
-            .await
+            self.execute_local().await
         })))
     }
 
     /// Refresh the view, waiting for the job to finish.
     pub async fn execute(self) -> Result<RefreshMaterializedViewResult> {
         if self.view.table.as_native().is_some() {
-            return refresh::execute_refresh(
-                &self.view.table,
-                self.full,
-                self.source_version,
-                self.expected_incarnation.as_deref(),
-            )
-            .await;
+            return self.execute_local().await;
         }
         self.execute_async().await?.wait().await
     }
+
+    async fn execute_local(&self) -> Result<RefreshMaterializedViewResult> {
+        let (table, full, pinned) = (&self.view.table, self.full, self.source_version);
+        let expected = self.expected_incarnation.as_deref();
+        if self.cascade {
+            refresh::execute_cascade(table, full, pinned, expected).await
+        } else {
+            refresh::execute_refresh(table, full, pinned, expected).await
+        }
+    }
 }
 
 impl Connection {

rust/lancedb/src/materialized_view/refresh.rs

@@ -696,6 +696,96 @@ pub(crate) async fn ensure_no_mem_wal(dataset: &Dataset, role: &str, name: &str)
     Ok(())
 }
 
+/// Refresh every upstream view of `view` furthest first, each reading the
+/// version the previous one left, then `view` itself. `full` is `view`'s only.
+pub(crate) async fn execute_cascade(
+    view: &Table,
+    full: bool,
+    pinned: Option<u64>,
+    expected_incarnation: Option<&str>,
+) -> Result<RefreshMaterializedViewResult> {
+    let (incarnation, hops) = upstream_views(view).await?;
+    if let Some(expected) = expected_incarnation
+        && incarnation.as_deref() != Some(expected)
+    {
+        return Err(Error::Runtime {
+            message: format!(
+                "materialized view '{}' is not the incarnation this refresh was \
+                 requested for: it was dropped and recreated",
+                view.name()
+            ),
+        });
+    }
+    let mut settled = None;
+    for (hop, incarnation) in &hops {
+        settled = Some(
+            execute_refresh(hop, false, settled, incarnation.as_deref())
+                .await?
+                .version,
+        );
+    }
+    execute_refresh(view, full, pinned.or(settled), incarnation.as_deref()).await
+}
+
+/// `view`'s incarnation and its upstream views, furthest first, each with the
+/// incarnation read in the same metadata read that yielded its lineage.
+async fn upstream_views(view: &Table) -> Result<(Option<String>, Vec<(Table, Option<String>)>)> {
+    let database = view.database_opt().ok_or_else(|| Error::InvalidInput {
+        message: "the view was not opened through a database connection".into(),
+    })?;
+    let open = |namespace_path: Vec<String>, name: String| {
+        database.open_table(OpenTableRequest {
+            name,
+            namespace_path,
+            index_cache_size: None,
+            lance_read_params: None,
+            location: None,
+            namespace_client: None,
+            managed_versioning: None,
+        })
+    };
+    let mut key = (view.namespace().to_vec(), view.name().to_string());
+    let mut visited = HashSet::new();
+    let mut named_incarnation = None;
+    let mut hops = Vec::new();
+    loop {
+        if !visited.insert(key.clone()) {
+            return Err(Error::InvalidInput {
+                message: format!("materialized view lineage of '{}' is cyclic", view.name()),
+            });
+        }
+        let table = Table::new(open(key.0.clone(), key.1.clone()).await?, database.clone());
+        let metadata = table.schema().await?.metadata().clone();
+        let definition = match super::read_definition(&metadata)? {
+            Some(super::StoredDefinition::Query(definition)) => definition,
+            Some(super::StoredDefinition::Newer { format }) => {
+                return Err(Error::NotSupported {
+                    message: format!(
+                        "materialized view '{}' is stored in format {format}, which this \
+                         version of lancedb cannot refresh",
+                        table.name()
+                    ),
+                });
+            }
+            None if visited.len() == 1 => {
+                return Err(Error::NotAMaterializedView {
+                    name: view.name().to_string(),
+                });
+            }
+            None => break,
+        };
+        let incarnation = metadata.get(INCARNATION_META_KEY).cloned();
+        if visited.len() == 1 {
+            named_incarnation = incarnation;
+        } else {
+            hops.push((table, incarnation));
+        }
+        key = (definition.source_namespace, definition.source_table);
+    }
+    hops.reverse();
+    Ok((named_incarnation, hops))
+}
+
 /// The table refresh scans: the staging table when the query calls a
 /// Function in FROM position, otherwise the query's source.
 pub(super) async fn open_source(
@@ -3443,6 +3533,77 @@ pub(crate) mod tests {
         assert_eq!(read(second.table(), "twice").await, vec![60, 100]);
     }
 
+    async fn view_over_doubled(conn: &Connection) -> MaterializedView {
+        conn.create_materialized_view("second", "doubled")
+            .with_no_data(true)
+            .only_if("twice > 10")
+            .execute()
+            .await
+            .unwrap()
+    }
+
+    #[tokio::test]
+    async fn test_cascade_refreshes_upstream_views_first() {
+        let (conn, source) = db_with_source(vec![1, 30]).await;
+        let first = doubled_view(&conn).await;
+        let second = view_over_doubled(&conn).await;
+
+        second.refresh().execute().await.unwrap();
+        assert_eq!(read(second.table(), "twice").await, Vec::<i32>::new());
+
+        let result = second
+            .refresh()
+            .cascade(true)
+            .full(true)
+            .execute()
+            .await
+            .unwrap();
+        assert_eq!(result.mode, RefreshMode::Rebuild);
+        let doubled = conn.open_table("doubled").execute().await.unwrap();
+        assert_eq!(read(&doubled, "twice").await, vec![2, 60]);
+        assert_eq!(read(second.table(), "twice").await, vec![60]);
+
+        append(&source, vec![50]).await;
+        let job = second
+            .refresh()
+            .cascade(true)
+            .execute_async()
+            .await
+            .unwrap();
+        assert_eq!(job.wait().await.unwrap().mode, RefreshMode::Incremental);
+        assert_eq!(read(second.table(), "twice").await, vec![60, 100]);
+        assert_eq!(
+            first.refresh().execute().await.unwrap().mode,
+            RefreshMode::NoOp
+        );
+    }
+
+    #[tokio::test]
+    async fn test_cascade_checks_the_incarnation_before_any_hop() {
+        let (conn, _) = db_with_source(vec![1, 30]).await;
+        doubled_view(&conn).await;
+        let second = view_over_doubled(&conn).await;
+
+        let err = second
+            .refresh()
+            .cascade(true)
+            .expect_incarnation("stale")
+            .execute()
+            .await
+            .unwrap_err();
+        assert!(err.to_string().contains("dropped and recreated"), "{err}");
+        let doubled = conn.open_table("doubled").execute().await.unwrap();
+        assert_eq!(read(&doubled, "twice").await, Vec::<i32>::new());
+    }
+
+    #[tokio::test]
+    async fn test_cascade_over_a_plain_table_is_a_refresh() {
+        let (conn, _) = db_with_source(vec![1, 2]).await;
+        let view = doubled_view(&conn).await;
+        view.refresh().cascade(true).execute().await.unwrap();
+        assert_eq!(read(view.table(), "twice").await, vec![2, 4]);
+    }
+
     /// The watermark speaks only for the state a refresh left behind: a
     /// direct write to the view is drift, and the next refresh rebuilds
     /// rather than preserving it as current.

rust/lancedb/src/remote/table.rs

@@ -2159,9 +2159,13 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
         full: bool,
         source_version: Option<u64>,
         expected_incarnation: Option<&str>,
+        cascade: bool,
     ) -> Result<Job<RefreshMaterializedViewResult>> {
         self.check_mutable().await?;
         let mut body = serde_json::json!({ "full": full });
+        if cascade {
+            body["cascade"] = true.into();
+        }
         if let Some(source_version) = source_version {
             body["source_version"] = source_version.into();
         }
@@ -12532,7 +12536,8 @@ mod tests {
                     json!({
                         "full": true,
                         "source_version": 7,
-                        "expected_incarnation": "inc-1"
+                        "expected_incarnation": "inc-1",
+                        "cascade": true
                     })
                 );
                 http::Response::builder()
@@ -12568,6 +12573,7 @@ mod tests {
             .full(true)
             .source_version(7)
             .expect_incarnation("inc-1")
+            .cascade(true)
             .execute()
             .await
             .unwrap();

rust/lancedb/src/table.rs

@@ -582,6 +582,7 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
         _full: bool,
         _source_version: Option<u64>,
         _expected_incarnation: Option<&str>,
+        _cascade: bool,
     ) -> Result<Job<crate::materialized_view::RefreshMaterializedViewResult>> {
         Err(Error::NotSupported {
             message: "remote materialized-view refresh is not supported on this table type".into(),