From d96151c3859972a8fac078c06c5aec36d044c07f Mon Sep 17 00:00:00 2001 From: Wyatt Alt Date: Thu, 24 Sep 2026 19:09:12 +0000 Subject: [PATCH] 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 | 6 +- nodejs/__test__/materialized_view.test.ts | 17 ++ nodejs/lancedb/materialized_view.ts | 6 +- nodejs/lancedb/table.ts | 8 +- nodejs/src/table.rs | 6 +- python/python/lancedb/_lancedb.pyi | 10 +- python/python/lancedb/materialized_view.py | 42 ++++- .../python/tests/test_materialized_views.py | 19 +++ python/src/table.rs | 10 +- rust/lancedb/src/materialized_view.rs | 36 ++-- rust/lancedb/src/materialized_view/refresh.rs | 161 ++++++++++++++++++ rust/lancedb/src/remote/table.rs | 8 +- rust/lancedb/src/table.rs | 1 + 13 files changed, 297 insertions(+), 33 deletions(-) diff --git a/docs/src/js/classes/MaterializedView.md b/docs/src/js/classes/MaterializedView.md index 58eaa3c13..d712e0a5c 100644 --- a/docs/src/js/classes/MaterializedView.md +++ b/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` diff --git a/nodejs/__test__/materialized_view.test.ts b/nodejs/__test__/materialized_view.test.ts index 5e5c08745..40281a2d2 100644 --- a/nodejs/__test__/materialized_view.test.ts +++ b/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", diff --git a/nodejs/lancedb/materialized_view.ts b/nodejs/lancedb/materialized_view.ts index 51f8cfead..939eb10d2 100644 --- a/nodejs/lancedb/materialized_view.ts +++ b/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 { validateNonNegativeInteger(options?.sourceVersion, "sourceVersion"); return await this.inner.refreshMaterializedView( options?.full, options?.sourceVersion, + options?.cascade, ); } } diff --git a/nodejs/lancedb/table.ts b/nodejs/lancedb/table.ts index a098322a9..4ca4eb133 100644 --- a/nodejs/lancedb/table.ts +++ b/nodejs/lancedb/table.ts @@ -670,6 +670,7 @@ export abstract class Table { abstract refreshMaterializedView( full?: boolean, sourceVersion?: number, + cascade?: boolean, ): Promise; /** @ignore */ @@ -1397,8 +1398,13 @@ export class LocalTable extends Table { async refreshMaterializedView( full?: boolean, sourceVersion?: number, + cascade?: boolean, ): Promise { - return await this.inner.refreshMaterializedView(full, sourceVersion); + return await this.inner.refreshMaterializedView( + full, + sourceVersion, + cascade, + ); } async materializedViewDefinition(): Promise { diff --git a/nodejs/src/table.rs b/nodejs/src/table.rs index fa74c20dc..a27f0fd2f 100644 --- a/nodejs/src/table.rs +++ b/nodejs/src/table.rs @@ -458,11 +458,15 @@ impl Table { &self, full: Option, source_version: Option, + cascade: Option, ) -> napi::Result { 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") diff --git a/python/python/lancedb/_lancedb.pyi b/python/python/lancedb/_lancedb.pyi index 337b736ce..92fac01f9 100644 --- a/python/python/lancedb/_lancedb.pyi +++ b/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: ... diff --git a/python/python/lancedb/materialized_view.py b/python/python/lancedb/materialized_view.py index a836210cb..fa1ada55b 100644 --- a/python/python/lancedb/materialized_view.py +++ b/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 + ) ) ) diff --git a/python/python/tests/test_materialized_views.py b/python/python/tests/test_materialized_views.py index c030ac1df..82bde88b5 100644 --- a/python/python/tests/test_materialized_views.py +++ b/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") diff --git a/python/src/table.rs b/python/src/table.rs index bf5250732..1145c4323 100644 --- a/python/src/table.rs +++ b/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, + cascade: bool, ) -> PyResult> { 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, + cascade: bool, ) -> PyResult> { 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); } diff --git a/rust/lancedb/src/materialized_view.rs b/rust/lancedb/src/materialized_view.rs index 430750a00..aafaf62ce 100644 --- a/rust/lancedb/src/materialized_view.rs +++ b/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, expected_incarnation: Option, + 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> { 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 { 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 { + 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 { diff --git a/rust/lancedb/src/materialized_view/refresh.rs b/rust/lancedb/src/materialized_view/refresh.rs index 5e9fe13e3..dc2fe62ef 100644 --- a/rust/lancedb/src/materialized_view/refresh.rs +++ b/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, + expected_incarnation: Option<&str>, +) -> Result { + 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, Vec<(Table, Option)>)> { + 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, 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::::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::::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. diff --git a/rust/lancedb/src/remote/table.rs b/rust/lancedb/src/remote/table.rs index 6bbb49adf..e303ab095 100644 --- a/rust/lancedb/src/remote/table.rs +++ b/rust/lancedb/src/remote/table.rs @@ -2159,9 +2159,13 @@ impl BaseTable for RemoteTable { full: bool, source_version: Option, expected_incarnation: Option<&str>, + cascade: bool, ) -> Result> { 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(); diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs index fbec73dfb..cf6144cb5 100644 --- a/rust/lancedb/src/table.rs +++ b/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, _expected_incarnation: Option<&str>, + _cascade: bool, ) -> Result> { Err(Error::NotSupported { message: "remote materialized-view refresh is not supported on this table type".into(),