From f0e6235db74a68c9c8c9234ad170bc45ebe9165d Mon Sep 17 00:00:00 2001 From: Wyatt Alt Date: Fri, 14 Aug 2026 19:33:34 -0700 Subject: [PATCH] feat: prepare a declaration as one opaque contract An external surface declaring views (a SQL layer, a service) needs to produce declarations byte-identical to this crate's, but exporting the raw planner and serializer would leave the assembly invariants -- the provenance column, the metadata embedding, the source checks -- to every caller. prepare_declaration canonicalizes the source through the coordinate a refresh will resolve (its own database, root namespace, bare name), plans against the table that resolution reaches, and rejects what the definition cannot record: a namespaced source, or a handle that does not resolve back to itself. Consuming the result through PreparedDeclaration::create is the only way to materialize it, and it creates in the retained source database -- there is no target parameter to misuse -- through the stable-row-id create-and-verify path. The builder itself now flows through both calls. Co-Authored-By: Claude Fable 5 --- rust/lancedb/src/connection.rs | 2 +- rust/lancedb/src/materialized_view.rs | 353 ++++++++++++++++++++++---- rust/lancedb/src/table.rs | 5 + 3 files changed, 304 insertions(+), 56 deletions(-) diff --git a/rust/lancedb/src/connection.rs b/rust/lancedb/src/connection.rs index 12ca306b8..ae37fe7e5 100644 --- a/rust/lancedb/src/connection.rs +++ b/rust/lancedb/src/connection.rs @@ -41,7 +41,7 @@ use lance_io::object_store::{StorageOptionsAccessor, StorageOptionsProvider}; mod create_table; -fn merge_storage_options( +pub(crate) fn merge_storage_options( store_params: &mut ObjectStoreParams, pairs: impl IntoIterator, ) { diff --git a/rust/lancedb/src/materialized_view.rs b/rust/lancedb/src/materialized_view.rs index 673c32f8e..c612fdac0 100644 --- a/rust/lancedb/src/materialized_view.rs +++ b/rust/lancedb/src/materialized_view.rs @@ -29,6 +29,7 @@ use serde::{Deserialize, Serialize}; use crate::connection::Connection; use crate::database::listing::OPT_NEW_TABLE_ENABLE_STABLE_ROW_IDS; +use crate::database::{CreateTableRequest, Database, OpenTableRequest}; use crate::table::Table; use crate::table::refresh::quote_identifier; use crate::{Error, Result}; @@ -351,6 +352,205 @@ fn project_schema(schema: &ArrowSchema, columns: &[String]) -> SchemaRef { Arc::new(ArrowSchema::new(fields)) } +/// A validated view declaration, ready to become a table: the complete +/// physical schema -- the projected fields plus [`SOURCE_ROW_ID_COLUMN`] -- +/// with the definition stamped in its metadata under +/// [`DEFINITION_META_KEY`]. Produced only by [`prepare_declaration`], so the +/// pieces cannot be assembled separately. +#[derive(Clone)] +pub struct PreparedDeclaration { + schema: SchemaRef, + definition: MaterializedViewDefinition, + /// The source's own database: the only place + /// [`PreparedDeclaration::create`] will put the view, because refresh + /// resolves the recorded source name through the view's database. + database: Arc, +} + +impl std::fmt::Debug for PreparedDeclaration { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("PreparedDeclaration") + .field("definition", &self.definition) + .finish_non_exhaustive() + } +} + +impl PreparedDeclaration { + /// The query the declaration records. + pub fn definition(&self) -> &MaterializedViewDefinition { + &self.definition + } + + /// Create the view table and verify it, consuming the declaration. + /// + /// The view goes in the source's own database, held since + /// [`prepare_declaration`]: refresh resolves the recorded source name + /// through the view's database, so there is nowhere else it could go. + /// The table is created empty from the prepared schema with stable row + /// ids (so the view can itself source another view), requested both as + /// write params and as the request-level creation option that outranks + /// database configuration, and verified rather than trusted because an + /// implementation may ignore both; nothing is rolled back on failure. + pub async fn create(self, name: &str) -> Result { + let empty: Vec> = + vec![]; + let reader: Box = + Box::new(arrow_array::RecordBatchIterator::new(empty, self.schema)); + let mut request = CreateTableRequest::new(name.to_string(), Box::new(reader)); + let write_params = request + .write_options + .lance_write_params + .get_or_insert_with(Default::default); + write_params.enable_stable_row_ids = true; + let store_params = write_params + .store_params + .get_or_insert_with(Default::default); + crate::connection::merge_storage_options( + store_params, + [( + OPT_NEW_TABLE_ENABLE_STABLE_ROW_IDS.to_string(), + "true".to_string(), + )], + ); + let table = self.database.clone().create_table(request).await?; + let table = Table::new(table, self.database); + let stable = match table.as_native() { + Some(native) => native.dataset.get().await?.manifest.uses_stable_row_ids(), + None => false, + }; + if !stable { + return Err(Error::Runtime { + message: format!( + "view '{name}' was created without stable row ids: the database \ + ignored the creation option; the table remains and is not \ + usable as a materialized view" + ), + }); + } + Ok(MaterializedView { + table, + definition: self.definition, + }) + } +} + +/// Validate a view declaration against its live source and hold what its +/// creation needs. +/// +/// The caller's handle only nominates the source: the declaration is +/// canonicalized through the coordinate a refresh will resolve -- the +/// source's own database, root namespace, bare name -- and planned against +/// the table that resolution reaches, so a handle that does not resolve +/// back to itself (a location override, a foreign database) is rejected, +/// as is a namespaced source, which the definition cannot record. Runs the +/// same creation-time checks as [`Connection::create_materialized_view`]: +/// the source must be a local table keeping stable row ids. Empty +/// `projections` selects every source column as of now. +/// +/// ```no_run +/// # #![recursion_limit = "256"] +/// # use lancedb::materialized_view::prepare_declaration; +/// # async fn declare(source: &lancedb::Table) -> Result<(), Box> { +/// let prepared = prepare_declaration( +/// source, +/// &[("id".into(), "id".into()), ("double".into(), "value * 2".into())], +/// Some("value > 0"), +/// None, +/// ) +/// .await?; +/// let view = prepared.create("doubles").await?; +/// # Ok(()) +/// # } +/// ``` +pub async fn prepare_declaration( + source: &Table, + projections: &[(String, String)], + filter: Option<&str>, + limit: Option, +) -> Result { + let Some(caller_native) = source.as_native() else { + return Err(Error::NotSupported { + message: "materialized views are supported only on local databases".into(), + }); + }; + // The definition records the source by bare name; any other source + // form would be recorded as a name its refresh cannot resolve. + if !source.namespace().is_empty() { + return Err(Error::NotSupported { + message: format!( + "a namespaced source cannot be recorded in a view definition; \ + '{}' must be a root-namespace table", + source.name() + ), + }); + } + let database = source + .database_opt() + .ok_or_else(|| Error::InvalidInput { + message: "the source was not opened through a database connection".into(), + })? + .clone(); + + // Canonicalize: resolve the recorded coordinate exactly the way a + // refresh will, and plan from what it reaches. A handle that does not + // resolve back to itself must not be declared under this name. + let resolved = database + .open_table(OpenTableRequest { + name: source.name().to_string(), + namespace_path: vec![], + index_cache_size: None, + lance_read_params: None, + location: None, + namespace_client: None, + managed_versioning: None, + }) + .await?; + let resolved = Table::new(resolved, database.clone()); + let Some(native) = resolved.as_native() else { + return Err(Error::NotSupported { + message: "materialized views are supported only on local databases".into(), + }); + }; + let caller_uri = caller_native.dataset.get().await?.uri().to_string(); + let resolved_uri = native.dataset.get().await?.uri().to_string(); + if caller_uri != resolved_uri { + return Err(Error::InvalidInput { + message: format!( + "the source handle does not resolve to itself through its \ + database: '{}' resolves to '{resolved_uri}', but the handle \ + reads '{caller_uri}'", + source.name() + ), + }); + } + if !native.dataset.get().await?.manifest.uses_stable_row_ids() { + return Err(Error::InvalidInput { + message: format!( + "materialized views require stable row ids on the source table; \ + create '{}' with storage option new_table_enable_stable_row_ids=true", + source.name() + ), + }); + } + let source_schema = resolved.schema().await?; + let (definition, mut fields) = + plan(source_schema, resolved.name(), projections, filter, limit)?; + fields.push(ArrowField::new( + SOURCE_ROW_ID_COLUMN, + DataType::UInt64, + false, + )); + let metadata = HashMap::from([( + DEFINITION_META_KEY.to_string(), + definition_to_metadata(&definition)?, + )]); + Ok(PreparedDeclaration { + schema: Arc::new(ArrowSchema::new_with_metadata(fields, metadata)), + definition, + database, + }) +} + /// One row of [`Connection::list_materialized_views`]: a view's name and its /// definition kind, which may be one this version cannot refresh. #[derive(Debug, Clone, PartialEq, Eq)] @@ -431,64 +631,14 @@ impl CreateMaterializedViewBuilder { pub async fn execute(self) -> Result { ensure_local(&self.connection)?; let source = self.connection.open_table(&self.source).execute().await?; - let Some(native) = source.as_native() else { - return Err(Error::NotSupported { - message: "materialized views are supported only on local databases".into(), - }); - }; - if !native.dataset.get().await?.manifest.uses_stable_row_ids() { - return Err(Error::InvalidInput { - message: format!( - "materialized views require stable row ids on the source table; \ - create '{}' with storage option new_table_enable_stable_row_ids=true", - self.source - ), - }); - } - let source_schema = source.schema().await?; - let (definition, mut fields) = plan( - source_schema, - &self.source, + let prepared = prepare_declaration( + &source, &self.projections, self.filter.as_deref(), self.limit, - )?; - fields.push(ArrowField::new( - SOURCE_ROW_ID_COLUMN, - DataType::UInt64, - false, - )); - - let metadata = HashMap::from([( - DEFINITION_META_KEY.to_string(), - definition_to_metadata(&definition)?, - )]); - let schema = Arc::new(ArrowSchema::new_with_metadata(fields, metadata)); - // Stable row ids so the view can itself source another view. The - // request-level option outranks any connection-level override; the - // manifest is still verified because another Database impl may - // ignore the option, but nothing is rolled back on failure. - let table = self - .connection - .create_empty_table(&self.name, schema) - .storage_option(OPT_NEW_TABLE_ENABLE_STABLE_ROW_IDS, "true") - .execute() - .await?; - let stable = match table.as_native() { - Some(native) => native.dataset.get().await?.manifest.uses_stable_row_ids(), - None => false, - }; - if !stable { - return Err(Error::Runtime { - message: format!( - "view '{}' was created without stable row ids: the database \ - ignored the creation option; the table remains and is not \ - usable as a materialized view", - self.name - ), - }); - } - Ok(MaterializedView { table, definition }) + ) + .await?; + prepared.create(&self.name).await } } @@ -1063,4 +1213,97 @@ mod tests { conn.drop_table("v", &[]).await.unwrap(); assert!(conn.list_materialized_views().await.unwrap().is_empty()); } + + /// The public declaration contract: prepare validates the source and + /// create consumes the declaration into a verified view table; a + /// source that cannot anchor refresh and a target outside the + /// source's database are both refused. + #[tokio::test] + async fn prepare_and_create_bind_the_declaration_lifecycle() { + let conn = connect("memory://").execute().await.unwrap(); + let batch = record_batch!(("id", Int32, [1, 2]), ("value", Int32, [3, 4])).unwrap(); + let source = conn + .create_table("src", batch.clone()) + .write_options(stable_row_ids()) + .execute() + .await + .unwrap(); + + let projections = [ + ("id".to_string(), "id".to_string()), + ("double".to_string(), "value * 2".to_string()), + ]; + let prepared = prepare_declaration(&source, &projections, Some("value > 0"), None) + .await + .unwrap(); + assert_eq!(prepared.definition().source_table, "src"); + + let view = prepared.create("v").await.unwrap(); + let schema = view.table().schema().await.unwrap(); + let names: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect(); + assert_eq!(names, ["id", "double", SOURCE_ROW_ID_COLUMN]); + assert!(schema.metadata().contains_key(DEFINITION_META_KEY)); + + // The same call rejects a source without stable row ids, so an + // external creation path cannot skip the check. + conn.create_table("plain", batch).execute().await.unwrap(); + let plain = conn.open_table("plain").execute().await.unwrap(); + let err = prepare_declaration(&plain, &[], None, None) + .await + .unwrap_err(); + assert!(err.to_string().contains("stable row ids"), "{err}"); + + // A handle whose location does not resolve back through its name is + // refused: the definition would record a name reaching other data. + let plain_uri = plain + .as_native() + .unwrap() + .dataset + .get() + .await + .unwrap() + .uri() + .to_string(); + let masquerade = conn + .open_table("src") + .location(plain_uri) + .execute() + .await + .unwrap(); + let err = prepare_declaration(&masquerade, &[], None, None) + .await + .unwrap_err(); + assert!( + err.to_string().contains("does not resolve to itself"), + "{err}" + ); + + // A namespaced source cannot be recorded in the definition: the + // bare name refresh resolves would reach a different table or none. + let namespaced = crate::table::NativeTable::create( + "memory://ns_src", + "ns_src", + vec!["ns".to_string()], + Box::new(arrow_array::RecordBatchIterator::new( + vec![], + std::sync::Arc::new(arrow_schema::Schema::new(vec![arrow_schema::Field::new( + "id", + arrow_schema::DataType::Int32, + true, + )])), + )) as Box, + None, + None, + None, + None, + std::collections::HashSet::new(), + ) + .await + .unwrap(); + let namespaced = Table::new(std::sync::Arc::new(namespaced), conn.database().clone()); + let err = prepare_declaration(&namespaced, &[], None, None) + .await + .unwrap_err(); + assert!(err.to_string().contains("namespaced source"), "{err}"); + } } diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs index 2e16b0940..8c12b2db3 100644 --- a/rust/lancedb/src/table.rs +++ b/rust/lancedb/src/table.rs @@ -1058,6 +1058,11 @@ impl Table { self.database.as_ref().unwrap() } + /// The database this handle was opened through, when it was. + pub fn database_opt(&self) -> Option<&Arc> { + self.database.as_ref() + } + pub fn embedding_registry(&self) -> &Arc { &self.embedding_registry }