diff --git a/docs/src/js/interfaces/MaterializedViewDefinition.md b/docs/src/js/interfaces/MaterializedViewDefinition.md index 741bbba31..607563de5 100644 --- a/docs/src/js/interfaces/MaterializedViewDefinition.md +++ b/docs/src/js/interfaces/MaterializedViewDefinition.md @@ -50,6 +50,16 @@ projections: [string, string][]; *** +### sourceNamespace + +```ts +sourceNamespace: string[]; +``` + +Namespace holding the source table; empty is the root namespace. + +*** + ### sourceTable ```ts diff --git a/nodejs/__test__/materialized_view.test.ts b/nodejs/__test__/materialized_view.test.ts index 2e7b2ec4d..b9d8ec911 100644 --- a/nodejs/__test__/materialized_view.test.ts +++ b/nodejs/__test__/materialized_view.test.ts @@ -48,6 +48,28 @@ describe("materialized views", () => { expect(definitionFromMetadata(safe, "v").limit).toBe(42); }); + it("reads the namespaced select kind and refuses unknown kinds", () => { + // "namespaced_select" is the namespaced form of "select": same shape, a + // separate kind so readers that predate it refuse instead of resolving + // the source at the root. + const namespaced = new Map([ + [ + DEFINITION_META_KEY, + '{"kind":"namespaced_select","source_table":"people","source_namespace":["ns"]}', + ], + ]); + const definition = definitionFromMetadata(namespaced, "v"); + expect(definition.sourceTable).toBe("people"); + expect(definition.sourceNamespace).toEqual(["ns"]); + + const unknown = new Map([ + [DEFINITION_META_KEY, '{"kind":"select_v3","source_table":"people"}'], + ]); + expect(() => definitionFromMetadata(unknown, "v")).toThrow( + /cannot refresh/, + ); + }); + it("creates, refreshes and queries a view", async () => { const view = await db.createMaterializedView("adults", "people", { select: ["name", ["shout", "upper(name)"]], diff --git a/nodejs/lancedb/materialized_view.ts b/nodejs/lancedb/materialized_view.ts index b26dee59e..1d47b640a 100644 --- a/nodejs/lancedb/materialized_view.ts +++ b/nodejs/lancedb/materialized_view.ts @@ -19,6 +19,8 @@ export interface MaterializedViewDefinition { limit?: number; /** Source columns the projections and filter read. */ inputs: string[]; + /** Namespace holding the source table; empty is the root namespace. */ + sourceNamespace: string[]; } /** @@ -78,7 +80,8 @@ export function definitionFromMetadata( } // biome-ignore lint/suspicious/noExplicitAny: raw JSON const value: any = JSON.parse(raw); - if (value.kind !== "select") { + // "namespaced_select" keeps older readers from resolving the source at root. + if (value.kind !== "select" && value.kind !== "namespaced_select") { throw new Error( `materialized view '${name}' is defined by '${value.kind}', which this ` + "version of lancedb cannot refresh", @@ -103,6 +106,7 @@ export function definitionFromMetadata( filter: value.filter ?? undefined, limit, inputs: value.inputs ?? [], + sourceNamespace: value.source_namespace ?? [], }; } diff --git a/python/python/lancedb/materialized_view.py b/python/python/lancedb/materialized_view.py index 5abb44dc0..c52a2d7a9 100644 --- a/python/python/lancedb/materialized_view.py +++ b/python/python/lancedb/materialized_view.py @@ -42,6 +42,8 @@ class MaterializedViewDefinition: """Cap on the number of rows the view holds.""" inputs: List[str] = field(default_factory=list) """Source columns the projections and filter read.""" + source_namespace: List[str] = field(default_factory=list) + """Namespace holding the source table; empty is the root namespace.""" def _definition_from_schema( @@ -53,7 +55,8 @@ def _definition_from_schema( raise ValueError(f"Table '{name}' is not a materialized view") value = json.loads(raw) kind = value.get("kind") - if kind != "select": + # "namespaced_select" keeps older readers from resolving the source at root. + if kind not in ("select", "namespaced_select"): raise NotImplementedError( f"materialized view '{name}' is defined by '{kind}', which this " "version of lancedb cannot refresh" @@ -66,6 +69,7 @@ def _definition_from_schema( filter=value.get("filter"), limit=value.get("limit"), inputs=value.get("inputs", []), + source_namespace=value.get("source_namespace", []), ) diff --git a/python/python/tests/test_materialized_views.py b/python/python/tests/test_materialized_views.py index 5fa3aa4fb..5cd7b23d6 100644 --- a/python/python/tests/test_materialized_views.py +++ b/python/python/tests/test_materialized_views.py @@ -266,3 +266,38 @@ async def test_async_namespace_connection_materialized_views(tmp_path): handle._route_pushdown_to_rust == through_namespace._route_pushdown_to_rust ) assert handle._namespace_path == through_namespace._namespace_path + + +def test_namespaced_select_kind_is_read_and_unknown_kinds_are_refused(): + import json + + import pyarrow as pa + + from lancedb.materialized_view import _definition_from_schema + + def schema_with(definition: dict) -> pa.Schema: + return pa.schema([pa.field("id", pa.int32())]).with_metadata( + {b"mv.definition": json.dumps(definition).encode()} + ) + + # "namespaced_select" is the namespaced form of "select": same shape, + # a separate kind so readers that predate it refuse instead of + # resolving the source at the root. + definition = _definition_from_schema( + schema_with( + { + "kind": "namespaced_select", + "source_table": "people", + "source_namespace": ["ns"], + "projections": [{"output": "name", "expression": "name"}], + } + ), + "v", + ) + assert definition.source_table == "people" + assert definition.source_namespace == ["ns"] + + with pytest.raises(NotImplementedError, match="cannot refresh"): + _definition_from_schema( + schema_with({"kind": "select_v3", "source_table": "people"}), "v" + ) diff --git a/rust/lancedb/src/materialized_view.rs b/rust/lancedb/src/materialized_view.rs index 08d6c921e..ec77c1181 100644 --- a/rust/lancedb/src/materialized_view.rs +++ b/rust/lancedb/src/materialized_view.rs @@ -74,8 +74,15 @@ const EMBEDDING_FUNCTIONS_META_KEY: &str = "embedding_functions"; const COLUMN_DEFINITIONS_META_KEY: &str = "lancedb::column_definitions"; /// Value of the definition's `kind` tag for the projected `select` form. +/// Reserved for root-namespace sources; see [`NAMESPACED_SELECT_KIND`]. pub const SELECT_KIND: &str = "select"; +/// The `select` form over a namespaced source: its own kind, because released +/// readers drop unknown fields and resolve a `select` source at the root, so +/// this routes them to the [`MaterializedViewKind::Unrecognized`] refusal +/// instead of a wrong-table refresh. +pub const NAMESPACED_SELECT_KIND: &str = "namespaced_select"; + /// Which view outputs each source column is projected to directly. A column /// may be projected more than once, so each carries every name the view gives /// it, in projection order. @@ -95,6 +102,10 @@ pub struct ViewProjection { pub struct MaterializedViewDefinition { /// Name of the source table, in the same database as the view. pub source_table: String, + /// Namespace path holding the source table; empty is the root namespace. + /// A definition written before namespaced sources reads as root. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub source_namespace: Vec, /// The projected output columns, in view schema order. pub projections: Vec, /// SQL predicate selecting the source rows the view holds. @@ -129,7 +140,12 @@ pub(crate) fn definition_to_metadata(definition: &MaterializedViewDefinition) -> let mut value = serde_json::to_value(definition).map_err(|e| Error::Runtime { message: format!("failed to serialize view definition: {e}"), })?; - value["kind"] = serde_json::Value::String(SELECT_KIND.to_string()); + let kind = if definition.source_namespace.is_empty() { + SELECT_KIND + } else { + NAMESPACED_SELECT_KIND + }; + value["kind"] = serde_json::Value::String(kind.to_string()); Ok(value.to_string()) } @@ -150,12 +166,21 @@ pub fn materialized_view_kind( .get("kind") .and_then(|k| k.as_str()) .ok_or_else(|| unreadable(&"missing kind tag"))?; - if kind != SELECT_KIND { + if kind != SELECT_KIND && kind != NAMESPACED_SELECT_KIND { return Ok(Some(MaterializedViewKind::Unrecognized { kind: kind.to_string(), })); } - let definition = serde_json::from_value(value).map_err(|e| unreadable(&e))?; + let kind = kind.to_string(); + let definition: MaterializedViewDefinition = + serde_json::from_value(value).map_err(|e| unreadable(&e))?; + // No correct writer produces a kind that disagrees with its namespace. + if (kind == SELECT_KIND) != definition.source_namespace.is_empty() { + return Err(unreadable(&format!( + "kind '{kind}' does not match its source namespace {:?}", + definition.source_namespace + ))); + } Ok(Some(MaterializedViewKind::Select(definition))) } @@ -166,6 +191,7 @@ pub fn materialized_view_kind( pub(crate) fn plan( source_schema: SchemaRef, source_table: &str, + source_namespace: &[String], projections: &[(String, String)], filter: Option<&str>, limit: Option, @@ -319,6 +345,7 @@ pub(crate) fn plan( let definition = MaterializedViewDefinition { source_table: source_table.to_string(), + source_namespace: source_namespace.to_vec(), projections: projections .into_iter() .map(|(output, expression)| ViewProjection { output, expression }) @@ -602,7 +629,7 @@ pub struct PreparedDeclaration { 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. + /// resolves the recorded source coordinate through the view's database. database: Arc, } @@ -622,10 +649,21 @@ impl PreparedDeclaration { /// Create the view table and verify it, consuming the declaration. /// - /// The view goes in the source's own database, where refresh resolves the - /// recorded source name. Stable row ids are requested at both levels and - /// verified rather than trusted; nothing is rolled back on failure. + /// The view goes at the root of the source's own database, where refresh + /// resolves the recorded source coordinate. Stable row ids are requested + /// at both levels and verified rather than trusted; nothing is rolled + /// back on failure. pub async fn create(self, name: &str) -> Result { + self.create_in(&[], name).await + } + + /// Create the view in `namespace_path`, empty for the root namespace. + /// Otherwise [`PreparedDeclaration::create`]. + pub async fn create_in( + self, + namespace_path: &[String], + name: &str, + ) -> Result { let empty: Vec> = vec![]; // Minted here, not at preparation: a declaration can be cloned and @@ -640,6 +678,7 @@ impl PreparedDeclaration { let reader: Box = Box::new(arrow_array::RecordBatchIterator::new(empty, schema)); let mut request = CreateTableRequest::new(name.to_string(), Box::new(reader)); + request.namespace_path = namespace_path.to_vec(); let write_params = request .write_options .lance_write_params @@ -680,8 +719,8 @@ impl PreparedDeclaration { /// Validate a view declaration against its live source and hold what its /// creation needs. The declaration is canonicalized through the coordinate a -/// refresh will resolve, so a handle that does not resolve back to itself is -/// rejected, as is a namespaced source. Same creation-time checks as +/// refresh will resolve -- name and namespace both -- so a handle that does +/// not resolve back to itself is rejected. Same creation-time checks as /// [`Connection::create_materialized_view`]. /// /// ```no_run @@ -710,17 +749,9 @@ pub async fn prepare_declaration( 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() - ), - }); - } + // Refresh resolves the source at exactly this coordinate, so the + // definition records the namespace alongside the name. + let source_namespace = source.namespace().to_vec(); let database = source .database_opt() .ok_or_else(|| Error::InvalidInput { @@ -734,7 +765,7 @@ pub async fn prepare_declaration( let resolved = database .open_table(OpenTableRequest { name: source.name().to_string(), - namespace_path: vec![], + namespace_path: source_namespace.clone(), index_cache_size: None, lance_read_params: None, location: None, @@ -780,6 +811,7 @@ pub async fn prepare_declaration( let (definition, mut fields, lineage) = plan( source_schema.clone(), resolved.name(), + &source_namespace, projections, filter, limit, @@ -839,7 +871,9 @@ fn ensure_local(connection: &Connection) -> Result<()> { pub struct CreateMaterializedViewBuilder { connection: Connection, name: String, + namespace: Vec, source: String, + source_namespace: Vec, projections: Vec<(String, String)>, filter: Option, limit: Option, @@ -850,13 +884,28 @@ impl CreateMaterializedViewBuilder { Self { connection, name, + namespace: Vec::new(), source, + source_namespace: Vec::new(), projections: Vec::new(), filter: None, limit: None, } } + /// The namespace to create the view in. Defaults to the root namespace. + pub fn namespace(mut self, namespace_path: Vec) -> Self { + self.namespace = namespace_path; + self + } + + /// The namespace holding the source table; recorded in the definition + /// for refresh to resolve. Defaults to the root namespace. + pub fn source_namespace(mut self, namespace_path: Vec) -> Self { + self.source_namespace = namespace_path; + self + } + /// The view's columns, as `(name, SQL expression)` pairs. Not calling /// this selects every source column, expanded at creation time. pub fn select( @@ -887,7 +936,12 @@ impl CreateMaterializedViewBuilder { /// provenance across compaction, and cannot be enabled later. pub async fn execute(self) -> Result { ensure_local(&self.connection)?; - let source = self.connection.open_table(&self.source).execute().await?; + let source = self + .connection + .open_table(&self.source) + .namespace(self.source_namespace.clone()) + .execute() + .await?; let prepared = prepare_declaration( &source, &self.projections, @@ -895,7 +949,7 @@ impl CreateMaterializedViewBuilder { self.limit, ) .await?; - prepared.create(&self.name).await + prepared.create_in(&self.namespace, &self.name).await } } @@ -1152,6 +1206,7 @@ mod tests { view.definition(), &MaterializedViewDefinition { source_table: "people".into(), + source_namespace: Vec::new(), projections: vec![ ViewProjection { output: "name".into(), @@ -2083,33 +2138,138 @@ mod tests { .await .unwrap_err(); assert!(err.to_string().contains("custom_loc"), "{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(), - ) + /// A view declared over a namespaced source records that namespace, and + /// refresh resolves the source through it -- the coordinate round-trips. + #[tokio::test] + async fn a_namespaced_source_round_trips_through_refresh() { + use lance_namespace::models::CreateNamespaceRequest; + + let tmp = tempfile::tempdir().unwrap(); + let mut properties = std::collections::HashMap::new(); + properties.insert("root".to_string(), tmp.path().to_str().unwrap().to_string()); + let conn = crate::connect_namespace("dir", properties) + .execute() + .await + .unwrap(); + conn.create_namespace(CreateNamespaceRequest { + id: Some(vec!["ns".into()]), + ..Default::default() + }) .await .unwrap(); - let namespaced = Table::new(std::sync::Arc::new(namespaced), conn.database().clone()); - let err = prepare_declaration(&namespaced, &[], None, None) + + let batch = record_batch!( + ("name", Utf8, ["ada", "grace", "alan"]), + ("age", Int32, [36, 85, 41]) + ) + .unwrap(); + conn.create_table("people", batch) + .namespace(vec!["ns".to_string()]) + .write_options(stable_row_ids()) + .execute() .await - .unwrap_err(); - assert!(err.to_string().contains("namespaced source"), "{err}"); + .unwrap(); + + // A decoy of the same name at the root: resolving the source at the + // wrong namespace materializes one row here instead of three. + let decoy = record_batch!(("name", Utf8, ["mallory"]), ("age", Int32, [42])).unwrap(); + conn.create_table("people", decoy) + .write_options(stable_row_ids()) + .execute() + .await + .unwrap(); + + let view = conn + .create_materialized_view("adults", "people") + .namespace(vec!["ns".to_string()]) + .source_namespace(vec!["ns".to_string()]) + .select([("name", "name")]) + .only_if("age >= 18") + .execute() + .await + .unwrap(); + + assert_eq!(view.definition().source_table, "people"); + assert_eq!(view.definition().source_namespace, vec!["ns".to_string()]); + assert_eq!(view.table().namespace(), &["ns"]); + + // Refresh resolves the source at the recorded namespace, not at root. + let result = view.refresh().execute().await.unwrap(); + assert_eq!(result.rows_written, 3); + } + + /// A definition stored before namespaced sources existed carries no + /// namespace key and must read as the root namespace. + #[test] + fn a_definition_without_a_namespace_reads_as_root() { + let stored = + r#"{"source_table":"people","projections":[{"output":"name","expression":"name"}]}"#; + let definition: MaterializedViewDefinition = serde_json::from_str(stored).unwrap(); + assert!(definition.source_namespace.is_empty()); + } + + fn definition(source_namespace: Vec) -> MaterializedViewDefinition { + MaterializedViewDefinition { + source_table: "people".to_string(), + source_namespace, + projections: vec![ViewProjection { + output: "name".to_string(), + expression: "name".to_string(), + }], + filter: None, + limit: None, + inputs: vec!["name".to_string()], + } + } + + /// A root definition keeps the pre-namespace `select` form byte-stably; + /// a namespaced one moves off `select`, which sends pre-namespace readers + /// to the `Unrecognized` refusal instead of a root resolve. + #[test] + fn a_namespaced_definition_is_refused_by_the_pre_namespace_reader() { + let root = definition_to_metadata(&definition(Vec::new())).unwrap(); + let root: serde_json::Value = serde_json::from_str(&root).unwrap(); + assert_eq!(root["kind"], "select"); + assert!( + root.get("source_namespace").is_none(), + "a root definition must not grow new keys: {root}" + ); + + let stored = definition_to_metadata(&definition(vec!["ns".to_string()])).unwrap(); + let value: serde_json::Value = serde_json::from_str(&stored).unwrap(); + // The pre-namespace discriminator is `kind == "select"`; anything + // else lands in its Unrecognized refusal rather than in a root open. + assert_eq!(value["kind"], "namespaced_select"); + + // The current reader round-trips the coordinate. + let metadata = HashMap::from([(DEFINITION_META_KEY.to_string(), stored)]); + match materialized_view_kind(&metadata).unwrap() { + Some(MaterializedViewKind::Select(read)) => { + assert_eq!(read.source_namespace, vec!["ns".to_string()]) + } + other => panic!("expected the namespaced select form, got {other:?}"), + } + } + + /// A kind that disagrees with its namespace is an error, not a view: + /// under `select` it is the shape old readers would resolve at the root. + #[test] + fn a_kind_namespace_mismatch_is_refused() { + for (kind, namespace) in [ + (SELECT_KIND, vec!["ns".to_string()]), + (NAMESPACED_SELECT_KIND, Vec::new()), + ] { + let mut value = serde_json::to_value(definition(namespace)).unwrap(); + value["kind"] = serde_json::Value::String(kind.to_string()); + let metadata = HashMap::from([(DEFINITION_META_KEY.to_string(), value.to_string())]); + let err = materialized_view_kind(&metadata).unwrap_err(); + assert!( + err.to_string() + .contains("does not match its source namespace"), + "kind '{kind}': {err}" + ); + } } } diff --git a/rust/lancedb/src/materialized_view/refresh.rs b/rust/lancedb/src/materialized_view/refresh.rs index b967e81f8..efddd1c7b 100644 --- a/rust/lancedb/src/materialized_view/refresh.rs +++ b/rust/lancedb/src/materialized_view/refresh.rs @@ -170,6 +170,7 @@ pub(crate) async fn execute_refresh( let (replanned, mut planned_fields, _renames) = super::plan( source_schema, &definition.source_table, + &definition.source_namespace, &projections, definition.filter.as_deref(), definition.limit, @@ -590,7 +591,7 @@ async fn open_source(view: &Table, definition: &MaterializedViewDefinition) -> R let source = database .open_table(OpenTableRequest { name: definition.source_table.clone(), - namespace_path: Vec::new(), + namespace_path: definition.source_namespace.clone(), index_cache_size: None, lance_read_params: None, location: None, @@ -2919,6 +2920,7 @@ mod tests { let replacement = crate::materialized_view::MaterializedViewDefinition { source_table: "src".into(), + source_namespace: Vec::new(), projections: vec![ crate::materialized_view::ViewProjection { output: "x".into(), @@ -2958,6 +2960,7 @@ mod tests { let narrower = crate::materialized_view::MaterializedViewDefinition { source_table: "src".into(), + source_namespace: Vec::new(), projections: vec![crate::materialized_view::ViewProjection { output: "x".into(), expression: "x".into(),