From f5e589bbe001768865eb28d191fb57862fb31b21 Mon Sep 17 00:00:00 2001 From: Wyatt Alt Date: Tue, 15 Sep 2026 20:55:27 +0000 Subject: [PATCH] feat: define a materialized view by its query, with UNNEST for table-valued Functions A view definition was a structured record under a `kind` tag, one kind per query shape, and a Function returning `list` was about to add a third. That names shapes instead of describing a relation. A materialized view is now a relation defined by a query, stored as one canonical SQL string under a format number: SELECT columns FROM [ns.]table [, UNNEST(column) AS alias] [WHERE predicate] [LIMIT n] Any other clause is refused at parse time. Older readers report the view as unrefreshable, the pre-format layouts still read, and a legacy view is rewritten on its next refresh that commits, rebuilt only where its raw text meant something else under lance's parser. UNNEST is the first operator past project and filter: projections plan against the source with the list column replaced by its element under the alias, and refresh unnests, filters and projects each batch itself. Row ids repeat per element, so eviction and incremental append are unchanged. A limit is refused with UNNEST: a scan limit counts source rows. --- .../interfaces/MaterializedViewDefinition.md | 59 +- nodejs/__test__/materialized_view.test.ts | 73 +- nodejs/lancedb/materialized_view.ts | 91 ++- nodejs/src/table.rs | 2 +- python/python/lancedb/materialized_view.py | 101 ++- .../python/tests/test_materialized_views.py | 58 +- python/src/table.rs | 2 +- rust/lancedb/src/database.rs | 2 +- rust/lancedb/src/materialized_view.rs | 701 +++++++++++++----- rust/lancedb/src/materialized_view/query.rs | 503 +++++++++++++ rust/lancedb/src/materialized_view/refresh.rs | 606 +++++++++++++-- rust/lancedb/src/remote/table.rs | 9 +- rust/lancedb/src/table/merge/lsm.rs | 2 +- 13 files changed, 1751 insertions(+), 458 deletions(-) create mode 100644 rust/lancedb/src/materialized_view/query.rs diff --git a/docs/src/js/interfaces/MaterializedViewDefinition.md b/docs/src/js/interfaces/MaterializedViewDefinition.md index 607563de5..37382ab46 100644 --- a/docs/src/js/interfaces/MaterializedViewDefinition.md +++ b/docs/src/js/interfaces/MaterializedViewDefinition.md @@ -6,64 +6,15 @@ # Interface: MaterializedViewDefinition -The query that defines a materialized view. +The query that defines a materialized view, as stored: +`SELECT columns FROM [ns.]table [, UNNEST(column) AS alias] [WHERE predicate] [LIMIT n]`. ## Properties -### filter? +### query ```ts -optional filter: string; +query: string; ``` -SQL predicate selecting the source rows the view holds. - -*** - -### inputs - -```ts -inputs: string[]; -``` - -Source columns the projections and filter read. - -*** - -### limit? - -```ts -optional limit: number; -``` - -Cap on the number of rows the view holds. - -*** - -### projections - -```ts -projections: [string, string][]; -``` - -`[output column, SQL expression]` pairs, in view schema order. - -*** - -### sourceNamespace - -```ts -sourceNamespace: string[]; -``` - -Namespace holding the source table; empty is the root namespace. - -*** - -### sourceTable - -```ts -sourceTable: string; -``` - -Name of the source table, in the same database as the view. +The defining query, in the canonical spelling the server stores. diff --git a/nodejs/__test__/materialized_view.test.ts b/nodejs/__test__/materialized_view.test.ts index 7d0e5ebb3..352628208 100644 --- a/nodejs/__test__/materialized_view.test.ts +++ b/nodejs/__test__/materialized_view.test.ts @@ -28,6 +28,39 @@ describe("materialized views", () => { }); afterEach(() => tmpDir.removeCallback()); + it("reads stored queries and legacy layouts", () => { + const read = (stored: string) => + definitionFromMetadata(new Map([[DEFINITION_META_KEY, stored]]), "v"); + const query = + "SELECT id, c.chunk FROM ns.docs, UNNEST(chunks) AS c WHERE id > 1"; + expect(read(`{"format":1,"query":${JSON.stringify(query)}}`).query).toBe( + query, + ); + + // The structured layout written before the format number reads as the + // query it described, under either of its kind tags. + expect( + read( + '{"kind":"namespaced_select","source_table":"people","source_namespace":["ns"],' + + '"projections":[{"output":"name","expression":"`name`"},' + + '{"output":"Shout","expression":"upper(name)"}],"filter":"age >= 18","limit":42}', + ).query, + ).toBe( + "SELECT `name`, upper(name) AS `Shout` FROM ns.people WHERE age >= 18 LIMIT 42", + ); + expect(read('{"kind":"select","source_table":"people"}').query).toBe( + "SELECT * FROM people", + ); + + // A newer writer's layout is reported, never guessed at. + for (const newer of [ + `{"format":2,"query":${JSON.stringify(query)}}`, + '{"kind":"select_v3","source_table":"people"}', + ]) { + expect(() => read(newer)).toThrow(/cannot refresh/); + } + }); + it("rejects a stored limit a number cannot carry", () => { const big = new Map([ [ @@ -38,36 +71,6 @@ describe("materialized views", () => { expect(() => definitionFromMetadata(big, "v")).toThrow( /too large to represent exactly/, ); - - const safe = new Map([ - [ - DEFINITION_META_KEY, - '{"kind":"select","source_table":"people","limit":42}', - ], - ]); - 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 () => { @@ -88,13 +91,9 @@ describe("materialized views", () => { }); const view = await db.openMaterializedView("adults"); const definition = await view.definition(); - expect(definition.sourceTable).toBe("people"); - expect(definition.filter).toBe("age >= 18"); - expect(definition.projections).toEqual([ - ["name", "`name`"], - ["age", "`age`"], - ]); - expect(definition.inputs).toEqual(["age", "name"]); + expect(definition.query).toBe( + "SELECT name, age FROM people WHERE age >= 18", + ); }); it("refreshes incrementally after an append", async () => { diff --git a/nodejs/lancedb/materialized_view.ts b/nodejs/lancedb/materialized_view.ts index e729c960b..750baff66 100644 --- a/nodejs/lancedb/materialized_view.ts +++ b/nodejs/lancedb/materialized_view.ts @@ -7,20 +7,16 @@ import { Table } from "./table"; /** Schema metadata key holding a materialized view's definition. */ export const DEFINITION_META_KEY = "mv.definition"; -/** The query that defines a materialized view. */ +/** The stored layout this version reads: `{"format": 1, "query": ""}`. */ +export const DEFINITION_FORMAT = 1; + +/** + * The query that defines a materialized view, as stored: + * `SELECT columns FROM [ns.]table [, UNNEST(column) AS alias] [WHERE predicate] [LIMIT n]`. + */ export interface MaterializedViewDefinition { - /** Name of the source table, in the same database as the view. */ - sourceTable: string; - /** `[output column, SQL expression]` pairs, in view schema order. */ - projections: [string, string][]; - /** SQL predicate selecting the source rows the view holds. */ - filter?: string; - /** Cap on the number of rows the view holds. */ - limit?: number; - /** Source columns the projections and filter read. */ - inputs: string[]; - /** Namespace holding the source table; empty is the root namespace. */ - sourceNamespace: string[]; + /** The defining query, in the canonical spelling the server stores. */ + query: string; } /** @@ -88,15 +84,21 @@ export function definitionFromJson( ): MaterializedViewDefinition { // biome-ignore lint/suspicious/noExplicitAny: raw JSON const value: any = JSON.parse(raw); - // "namespaced_select" keeps older readers from resolving the source at root. - if ( - value.kind !== undefined && - value.kind !== "select" && - value.kind !== "namespaced_select" - ) { + if (value.format !== undefined) { + // A newer writer's layout is reported, never guessed at. + if (!Number.isInteger(value.format) || value.format > DEFINITION_FORMAT) { + throw new Error( + `materialized view '${name}' is stored in format ${value.format}, ` + + "which this version of lancedb cannot refresh", + ); + } + return { query: value.query }; + } + // The structured layout written before the format number. + 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", + `materialized view '${name}' is stored in format kind '${value.kind}', ` + + "which this version of lancedb cannot refresh", ); } const limit = value.limit ?? undefined; @@ -108,18 +110,41 @@ export function definitionFromJson( `materialized view '${name}' has a stored limit too large to represent exactly`, ); } - return { - sourceTable: value.source_table, - // biome-ignore lint/suspicious/noExplicitAny: raw JSON - projections: (value.projections ?? []).map((p: any) => [ - p.output, - p.expression, - ]), - filter: value.filter ?? undefined, - limit, - inputs: value.inputs ?? [], - sourceNamespace: value.source_namespace ?? [], - }; + return { query: legacyQuery(value, limit) }; +} + +function legacyIdent(name: string): string { + return /^[a-z_][a-z0-9_]*$/.test(name) + ? name + : `\`${name.replace(/`/g, "``")}\``; +} + +/** Render the pre-format structured layout as the query it described. */ +// biome-ignore lint/suspicious/noExplicitAny: raw JSON +function legacyQuery(value: any, limit: number | undefined): string { + // biome-ignore lint/suspicious/noExplicitAny: raw JSON + const projections: any[] = value.projections ?? []; + const columns = + projections.length === 0 + ? "*" + : projections + .map((p) => + p.expression === p.output || p.expression === `\`${p.output}\`` + ? p.expression + : `${p.expression} AS ${legacyIdent(p.output)}`, + ) + .join(", "); + const table = [...(value.source_namespace ?? []), value.source_table] + .map(legacyIdent) + .join("."); + let query = `SELECT ${columns} FROM ${table}`; + if (value.filter !== undefined && value.filter !== null) { + query += ` WHERE ${value.filter}`; + } + if (limit !== undefined) { + query += ` LIMIT ${limit}`; + } + return query; } /** diff --git a/nodejs/src/table.rs b/nodejs/src/table.rs index 344017d08..14592545f 100644 --- a/nodejs/src/table.rs +++ b/nodejs/src/table.rs @@ -453,7 +453,7 @@ impl Table { let view = lancedb::MaterializedView::from_table(inner) .await .default_error()?; - serde_json::to_string(view.definition()).map_err(|err| { + lancedb::materialized_view::definition_to_metadata(view.definition()).map_err(|err| { napi::Error::from_reason(format!( "failed to serialize materialized-view definition: {err}" )) diff --git a/python/python/lancedb/materialized_view.py b/python/python/lancedb/materialized_view.py index 384683c9f..08d19256c 100644 --- a/python/python/lancedb/materialized_view.py +++ b/python/python/lancedb/materialized_view.py @@ -7,7 +7,7 @@ maintained by refresh. See ``DBConnection.create_materialized_view``.""" from __future__ import annotations import json -from dataclasses import dataclass, field +from dataclasses import dataclass from typing import TYPE_CHECKING, Dict, List, Optional, Sequence, Tuple, Union from .background_loop import LOOP @@ -29,22 +29,21 @@ SelectArg = Union[ ] +DEFINITION_FORMAT = 1 +"""The stored layout this version reads: ``{"format": 1, "query": ""}``. +A ``kind`` key beside it is for readers older than the format number.""" + + @dataclass class MaterializedViewDefinition: - """The query that defines a materialized view.""" + """The query that defines a materialized view, as stored:: - source_table: str - """Name of the source table, in the same database as the view.""" - projections: List[Tuple[str, str]] - """``(output column, SQL expression)`` pairs, in view schema order.""" - filter: Optional[str] = None - """SQL predicate selecting the source rows the view holds.""" - limit: Optional[int] = None - """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.""" + SELECT columns FROM [ns.]table [, UNNEST(column) AS alias] + [WHERE predicate] [LIMIT n] + """ + + query: str + """The defining query, in the canonical spelling the server stores.""" def _definition_from_schema( @@ -54,38 +53,62 @@ def _definition_from_schema( raw = metadata.get(DEFINITION_META_KEY) if raw is None: raise ValueError(f"Table '{name}' is not a materialized view") + return _definition_from_json(raw, name) + + +def _definition_from_json(raw: str, name: str) -> MaterializedViewDefinition: + """Parse the stored layout, which native code also returns for a view on + any connection.""" value = json.loads(raw) + fmt = value.get("format") + if fmt is not None: + # A newer writer's layout is reported, never guessed at. + if not isinstance(fmt, int) or fmt > DEFINITION_FORMAT: + raise NotImplementedError( + f"materialized view '{name}' is stored in format {fmt}, which " + "this version of lancedb cannot refresh" + ) + return MaterializedViewDefinition(query=value["query"]) + # The structured layout written before the format number. kind = value.get("kind") - # "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" + f"materialized view '{name}' is stored in format kind '{kind}', " + "which this version of lancedb cannot refresh" ) - return MaterializedViewDefinition( - source_table=value["source_table"], - projections=[ - (p["output"], p["expression"]) for p in value.get("projections", []) - ], - filter=value.get("filter"), - limit=value.get("limit"), - inputs=value.get("inputs", []), - source_namespace=value.get("source_namespace", []), - ) + return MaterializedViewDefinition(query=_legacy_query(value)) -def _definition_from_json(raw: str) -> MaterializedViewDefinition: - value = json.loads(raw) - return MaterializedViewDefinition( - source_table=value["source_table"], - projections=[ - (p["output"], p["expression"]) for p in value.get("projections", []) - ], - filter=value.get("filter"), - limit=value.get("limit"), - inputs=value.get("inputs", []), - source_namespace=value.get("source_namespace", []), +def _legacy_ident(name: str) -> str: + if name and all(c == "_" or c.islower() or c.isdigit() for c in name): + return name + return _quote_identifier(name) + + +def _legacy_query(value: dict) -> str: + """Render the pre-format structured layout as the query it described.""" + projections = value.get("projections", []) + if projections: + items = [] + for p in projections: + output, expression = p["output"], p["expression"] + if expression in (output, _quote_identifier(output)): + items.append(expression) + else: + items.append(f"{expression} AS {_legacy_ident(output)}") + columns = ", ".join(items) + else: + columns = "*" + table = ".".join( + _legacy_ident(part) + for part in [*value.get("source_namespace", []), value["source_table"]] ) + query = f"SELECT {columns} FROM {table}" + if value.get("filter") is not None: + query += f" WHERE {value['filter']}" + if value.get("limit") is not None: + query += f" LIMIT {value['limit']}" + return query def _quote_identifier(name: str) -> str: @@ -143,7 +166,7 @@ class AsyncMaterializedView: async def definition(self) -> MaterializedViewDefinition: """The query that defines the view.""" raw = await self._table._inner.materialized_view_definition() - return _definition_from_json(raw) + return _definition_from_json(raw, self.name) async def refresh( self, *, full: bool = False, source_version: Optional[int] = None diff --git a/python/python/tests/test_materialized_views.py b/python/python/tests/test_materialized_views.py index 695ef142a..cfe2405a4 100644 --- a/python/python/tests/test_materialized_views.py +++ b/python/python/tests/test_materialized_views.py @@ -217,10 +217,7 @@ def test_definition_round_trips(tmp_path): view = db.open_materialized_view("adults") assert view.definition == MaterializedViewDefinition( - source_table="people", - projections=[("name", "`name`"), ("age", "`age`")], - filter="age >= 18", - inputs=["age", "name"], + query="SELECT name, age FROM people WHERE age >= 18" ) @@ -305,7 +302,7 @@ async def test_async_create_refresh_and_open(tmp_path): reopened = await db.open_materialized_view("shouts") definition = await reopened.definition() - assert definition.projections == [("shout", "upper(name)")] + assert definition.query == "SELECT upper(name) AS shout FROM people" assert await db.list_materialized_views() == ["shouts"] @@ -423,7 +420,7 @@ def test_namespace_connection_materialized_views(tmp_path): assert db.list_materialized_views() == ["adults"] reopened = db.open_materialized_view("adults") - assert reopened.definition.source_table == "people" + assert reopened.definition.query.startswith("SELECT name, age FROM ") with pytest.raises(ValueError, match="not a materialized view"): db.open_materialized_view("people") @@ -458,7 +455,7 @@ async def test_async_namespace_connection_materialized_views(tmp_path): assert await db.list_materialized_views() == ["adults"] reopened = await db.open_materialized_view("adults") - assert (await reopened.definition()).source_table == "people" + assert (await reopened.definition()).query.startswith("SELECT name, age FROM ") # The view's table came through the namespace, not straight from the # inner connection: a bare inner table carries no namespace context, so @@ -484,36 +481,49 @@ async def test_async_namespace_connection_materialized_views(tmp_path): assert await db.list_materialized_views() == [] -def test_namespaced_select_kind_is_read_and_unknown_kinds_are_refused(): +def test_stored_queries_and_legacy_layouts_are_read(): 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( + def read(definition: dict) -> MaterializedViewDefinition: + schema = pa.schema([pa.field("id", pa.int32())]).with_metadata( {b"mv.definition": json.dumps(definition).encode()} ) + return _definition_from_schema(schema, "v") - # "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( + query = "SELECT id, c.chunk FROM ns.docs, UNNEST(chunks) AS c WHERE id > 1" + assert read({"format": 1, "query": query}).query == query + + # The structured layout written before the format number reads as the + # query it described, under either of its kind tags. + assert ( + read( { "kind": "namespaced_select", "source_table": "people", "source_namespace": ["ns"], - "projections": [{"output": "name", "expression": "name"}], + "projections": [ + {"output": "name", "expression": "`name`"}, + {"output": "Shout", "expression": "upper(name)"}, + ], + "filter": "age >= 18", + "limit": 10, } - ), - "v", + ).query + == "SELECT `name`, upper(name) AS `Shout` FROM ns.people " + "WHERE age >= 18 LIMIT 10" + ) + assert read({"kind": "select", "source_table": "people"}).query == ( + "SELECT * FROM people" ) - 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" - ) + # A newer writer's layout is reported, never guessed at. + for newer in ( + {"format": 2, "query": query}, + {"kind": "select_v3", "source_table": "people"}, + ): + with pytest.raises(NotImplementedError, match="cannot refresh"): + read(newer) diff --git a/python/src/table.rs b/python/src/table.rs index 4cea61543..ff1ae1e13 100644 --- a/python/src/table.rs +++ b/python/src/table.rs @@ -1684,7 +1684,7 @@ impl Table { let view = lancedb::MaterializedView::from_table(inner) .await .infer_error()?; - serde_json::to_string(view.definition()).map_err(|err| { + lancedb::materialized_view::definition_to_metadata(view.definition()).map_err(|err| { PyRuntimeError::new_err(format!( "failed to serialize materialized-view definition: {err}" )) diff --git a/rust/lancedb/src/database.rs b/rust/lancedb/src/database.rs index 392277f63..0adac778c 100644 --- a/rust/lancedb/src/database.rs +++ b/rust/lancedb/src/database.rs @@ -360,7 +360,7 @@ pub trait Database: continue; }; let schema = table.schema().await?; - if crate::materialized_view::materialized_view_kind(schema.metadata())?.is_some() { + if crate::materialized_view::read_definition(schema.metadata())?.is_some() { names.push(name); } } diff --git a/rust/lancedb/src/materialized_view.rs b/rust/lancedb/src/materialized_view.rs index 94fcb5dda..d9aeb3ed5 100644 --- a/rust/lancedb/src/materialized_view.rs +++ b/rust/lancedb/src/materialized_view.rs @@ -5,11 +5,12 @@ //! //! A materialized view is a table whose contents are defined by a query over //! one source table and maintained by refresh rather than by writes. Creation -//! records a kind-tagged definition in schema metadata and populates the view -//! unless creation explicitly requests no data. A kind added later reads back -//! as unrefreshable, not as a plain table. Queries, indexes and search work on -//! the view unchanged. +//! records the defining query in schema metadata and populates the view +//! unless creation explicitly requests no data. A query this version cannot +//! maintain reads back as unrefreshable, not as a plain table. Queries, +//! indexes and search work on the view unchanged. +mod query; pub mod refresh; #[cfg(test)] @@ -35,13 +36,12 @@ use crate::table::computed_columns::{ FUNCTION_BINDINGS_META_KEY, computed_column_from_field, computed_columns, ensure_declarations_are_planned, function_bindings_metadata, }; -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. +/// Schema metadata key holding the view definition; see [`DEFINITION_FORMAT`]. pub const DEFINITION_META_KEY: &str = "mv.definition"; /// Schema metadata key holding the view's incarnation: a token minted at each @@ -80,14 +80,20 @@ const EMBEDDING_FUNCTIONS_META_KEY: &str = "embedding_functions"; /// produces, which is what lets a query embed its own text. 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`]. +/// The layout this version writes under [`DEFINITION_META_KEY`]: +/// `{"format": 1, "query": ""}`, the query as +/// [`MaterializedViewDefinition::to_sql`] renders it. A reader refuses a +/// newer format rather than guess at it. The layout also carries +/// `"kind": "query"`, which readers older than the format number report +/// as an unrefreshable view instead of failing to read the metadata. +pub const DEFINITION_FORMAT: u64 = 1; + +/// Legacy `kind` tag of the structured layout written before +/// [`DEFINITION_FORMAT`] existed; still read, never written. 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. +/// Legacy `kind` tag of the structured layout over a namespaced source; +/// still read, never written. pub const NAMESPACED_SELECT_KIND: &str = "namespaced_select"; /// Which view outputs each source column is projected to directly. A column @@ -104,31 +110,76 @@ pub struct ViewProjection { pub expression: String, } -/// The query that defines a materialized view. -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +/// `UNNEST(column) AS alias`: each source row yields one view row per +/// element of the list column, and projections read the element as `alias`. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ViewUnnest { + /// The source's list column. + pub column: String, + /// The name the element is read through. + pub alias: String, +} + +/// The query that defines a materialized view, in the relational shape +/// refresh maintains. Stored as SQL; see [`MaterializedViewDefinition::from_sql`] +/// for the shape. +#[derive(Debug, Clone, PartialEq, Eq)] 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. + /// The list column the view unnests, if any. + pub unnest: Option, + /// The projected output columns, in view schema order. Empty selects + /// every source column, expanded when the view is planned. pub projections: Vec, - /// SQL predicate selecting the source rows the view holds. - #[serde(default, skip_serializing_if = "Option::is_none")] + /// SQL predicate selecting the rows the view holds. pub filter: Option, /// Cap on the number of rows the view holds, in materialization order. - #[serde(default, skip_serializing_if = "Option::is_none")] pub limit: Option, - /// Source columns the projections and filter read, derived at creation. - #[serde(default)] - pub inputs: Vec, +} + +impl MaterializedViewDefinition { + /// Parse the defining query: + /// + /// ```sql + /// SELECT , ... FROM [ns.]table [, UNNEST(column) AS alias] + /// [WHERE predicate] [LIMIT n] + /// ``` + /// + /// Any other clause is refused: this engine cannot maintain it, and a + /// definition it does not fully understand must not be materialized. + pub fn from_sql(sql: &str) -> Result { + query::parse(sql) + } + + /// The defining query in its canonical spelling, which is what is + /// stored and what [`Self::from_sql`] reads back equal. + pub fn to_sql(&self) -> String { + query::render(self) + } +} + +/// A view definition as read back from schema metadata. Non-exhaustive so +/// a later outcome is additive. +#[derive(Debug, Clone, PartialEq, Eq)] +#[non_exhaustive] +pub enum StoredDefinition { + /// A query this version can maintain. + Query(MaterializedViewDefinition), + /// Written by a newer version, reported so a caller can tell an + /// unrefreshable view apart from a plain table. `format` is the tag as + /// found: a format number, or a legacy `kind`. + Newer { + /// The format tag as stored. + format: String, + }, } /// The backend-independent metadata needed to open a materialized view. #[doc(hidden)] -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq)] pub struct MaterializedViewInfo { /// The parsed view definition. pub definition: MaterializedViewDefinition, @@ -155,47 +206,43 @@ pub struct CreateMaterializedViewRequest { /// [`PreparedDeclaration::input_column`]. pub const INPUT_COLUMN_PREFIX: &str = "__input_"; -/// The internal view column holding a copy of `source_column`. +/// The internal view column holding a copy of `source_column`; a nested +/// path's separators become `__`, since a top-level name cannot hold `.`. pub fn input_column_name(source_column: &str) -> String { - format!("{INPUT_COLUMN_PREFIX}{source_column}") + format!("{INPUT_COLUMN_PREFIX}{}", source_column.replace('.', "__")) } -/// A view definition as read back from schema metadata. Non-exhaustive so a -/// kind added later is additive. -#[derive(Debug, Clone, PartialEq, Eq)] -#[non_exhaustive] -pub enum MaterializedViewKind { - /// The projected `select` form. - Select(MaterializedViewDefinition), - /// A kind written by a newer version, reported so a caller can tell an - /// unrefreshable view apart from a plain table. Nothing produces this. - Unrecognized { - /// The kind as it was found in the metadata. - kind: String, - }, +/// The structured layout written before [`DEFINITION_FORMAT`]. Read only; +/// refresh rewrites such a view in the current layout. +#[derive(Deserialize)] +struct LegacyDefinition { + source_table: String, + #[serde(default)] + source_namespace: Vec, + projections: Vec, + #[serde(default)] + filter: Option, + #[serde(default)] + limit: Option, } -/// Serialize `definition` into the kind-tagged form stored under -/// [`DEFINITION_META_KEY`]. -pub(crate) fn definition_to_metadata(definition: &MaterializedViewDefinition) -> Result { - let mut value = serde_json::to_value(definition).map_err(|e| Error::Runtime { - message: format!("failed to serialize view definition: {e}"), - })?; - 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()) +/// Serialize `definition` into the layout stored under +/// [`DEFINITION_META_KEY`]; the bindings hand this same layout to their +/// language-side readers. +#[doc(hidden)] +pub fn definition_to_metadata(definition: &MaterializedViewDefinition) -> Result { + Ok(serde_json::json!({ + "kind": "query", + "format": DEFINITION_FORMAT, + "query": definition.to_sql(), + }) + .to_string()) } -/// Read a view declaration off a schema metadata map, if it carries one. -/// `Ok(None)` for a plain table; a declaration that does not parse is an +/// Read a view definition off a schema metadata map, if it carries one. +/// `Ok(None)` for a plain table; a definition that does not parse is an /// error, because treating a view as plain would let it be rewritten. -pub fn materialized_view_kind( - metadata: &HashMap, -) -> Result> { +pub fn read_definition(metadata: &HashMap) -> Result> { let Some(raw) = metadata.get(DEFINITION_META_KEY) else { return Ok(None); }; @@ -203,26 +250,47 @@ pub fn materialized_view_kind( message: format!("unreadable materialized view definition: {e}"), }; let value: serde_json::Value = serde_json::from_str(raw).map_err(|e| unreadable(&e))?; + if let Some(format) = value.get("format") { + let Some(format) = format.as_u64() else { + return Err(unreadable(&format!("format tag {format} is not a number"))); + }; + if format > DEFINITION_FORMAT { + return Ok(Some(StoredDefinition::Newer { + format: format.to_string(), + })); + } + let Some(sql) = value.get("query").and_then(|q| q.as_str()) else { + return Err(unreadable(&"missing query")); + }; + let definition = MaterializedViewDefinition::from_sql(sql).map_err(|e| unreadable(&e))?; + return Ok(Some(StoredDefinition::Query(definition))); + } let kind = value .get("kind") .and_then(|k| k.as_str()) - .ok_or_else(|| unreadable(&"missing kind tag"))?; + .ok_or_else(|| unreadable(&"missing format tag"))? + .to_string(); if kind != SELECT_KIND && kind != NAMESPACED_SELECT_KIND { - return Ok(Some(MaterializedViewKind::Unrecognized { - kind: kind.to_string(), + return Ok(Some(StoredDefinition::Newer { + format: format!("kind '{kind}'"), })); } - 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() { + let legacy: LegacyDefinition = serde_json::from_value(value).map_err(|e| unreadable(&e))?; + // No correct writer produced a tag that disagrees with the definition. + if (kind == SELECT_KIND) != legacy.source_namespace.is_empty() { return Err(unreadable(&format!( "kind '{kind}' does not match its source namespace {:?}", - definition.source_namespace + legacy.source_namespace ))); } - Ok(Some(MaterializedViewKind::Select(definition))) + Ok(Some(StoredDefinition::Query(MaterializedViewDefinition { + source_table: legacy.source_table, + source_namespace: legacy.source_namespace, + unnest: None, + projections: legacy.projections, + filter: legacy.filter, + limit: legacy.limit, + }))) } pub(crate) fn materialized_view_info_from_metadata( @@ -230,15 +298,15 @@ pub(crate) fn materialized_view_info_from_metadata( metadata: &HashMap, ) -> Result { let incarnation = metadata.get(INCARNATION_META_KEY).cloned(); - match materialized_view_kind(metadata)? { - Some(MaterializedViewKind::Select(definition)) => Ok(MaterializedViewInfo { + match read_definition(metadata)? { + Some(StoredDefinition::Query(definition)) => Ok(MaterializedViewInfo { definition, incarnation, }), - Some(MaterializedViewKind::Unrecognized { kind }) => Err(Error::NotSupported { + Some(StoredDefinition::Newer { format }) => Err(Error::NotSupported { message: format!( - "materialized view '{name}' is defined by '{kind}', which this version of \ - lancedb cannot refresh" + "materialized view '{name}' is stored in format {format}, which this version \ + of lancedb cannot refresh" ), }), None => Err(Error::NotAMaterializedView { @@ -248,19 +316,32 @@ pub(crate) fn materialized_view_info_from_metadata( } /// Resolve a definition against the source schema into the view's projected -/// fields, with `inputs` filled in. Everything statically checkable is -/// checked here rather than at refresh time. Empty `projections` selects -/// every source column as the schema stands now. +/// fields, with `inputs` filled in and every expression in its canonical +/// spelling. Everything statically checkable is checked here rather than at +/// refresh time. Empty `projections` selects every column as the schema +/// stands now. +#[derive(Debug)] +pub(crate) struct Planned { + /// The definition with every expression in its canonical spelling and + /// `SELECT *` expanded. + pub definition: MaterializedViewDefinition, + /// The view's projected fields, in order. + pub fields: Vec, + pub lineage: Lineage, + /// Source columns the query reads; a read through the unnest alias is + /// recorded as the list column, which is what the source has and what + /// incremental refresh watches. + pub inputs: Vec, +} + pub(crate) fn plan( source_schema: SchemaRef, - source_table: &str, - source_namespace: &[String], - projections: Option<&[(String, String)]>, - filter: Option<&str>, - limit: Option, -) -> Result<(MaterializedViewDefinition, Vec, Lineage)> { - let filter = filter - .map(crate::expr::canonicalize_sql_predicate) + definition: &MaterializedViewDefinition, +) -> Result { + let filter = definition + .filter + .as_deref() + .map(query::canonical_expr) .transpose() .map_err(|err| match err { Error::InvalidInput { message } => Error::InvalidInput { @@ -268,17 +349,45 @@ pub(crate) fn plan( }, err => err, })?; - let projections: Vec<(String, String)> = match projections { - Some(projections) => projections.to_vec(), + // Projections are typed against the source, or for an unnested view + // against the source with the list column replaced by its element + // under the alias, where `c.chunk` is an ordinary nested path. + let source_schema = match &definition.unnest { + None => source_schema, + Some(unnest) => { + // A scan limit counts source rows, not the elements they expand to. + if definition.limit.is_some() { + return Err(Error::InvalidInput { + message: "LIMIT is not supported together with UNNEST".to_string(), + }); + } + flattened_schema(&source_schema, unnest)? + } + }; + let projections: Vec<(String, String)> = if definition.projections.is_empty() { // `SELECT *`. A source that is itself a view carries its own // provenance column; the new view records its own, not a copy. - None => source_schema + source_schema .fields() .iter() .filter(|f| f.name() != SOURCE_ROW_ID_COLUMN) - .map(|f| (f.name().clone(), quote_identifier(f.name()))) - .collect(), + .map(|f| (f.name().clone(), query::ident_sql(f.name()))) + .collect() + } else { + definition + .projections + .iter() + .map(|p| { + let expression = + query::canonical_expr(&p.expression).map_err(|e| Error::InvalidExpression { + column: p.output.clone(), + message: e.to_string(), + })?; + Ok((p.output.clone(), expression)) + }) + .collect::>()? }; + let limit = definition.limit; // A scan takes the cap as i64. Rejecting it here keeps creation and // refresh from disagreeing about whether a view is valid. @@ -409,21 +518,75 @@ pub(crate) fn plan( inputs.extend(filter_inputs); } - inputs.sort(); - inputs.dedup(); - let definition = MaterializedViewDefinition { - source_table: source_table.to_string(), - source_namespace: source_namespace.to_vec(), + source_table: definition.source_table.clone(), + source_namespace: definition.source_namespace.clone(), + unnest: definition.unnest.clone(), projections: projections .into_iter() .map(|(output, expression)| ViewProjection { output, expression }) .collect(), filter, limit, - inputs, }; - Ok((definition, fields, lineage)) + let mut inputs: Vec = inputs + .iter() + .map(|input| recorded_input(&definition, input)) + .collect(); + inputs.sort(); + inputs.dedup(); + Ok(Planned { + definition, + fields, + lineage, + inputs, + }) +} + +/// The schema a projection over an unnested view is planned against: the +/// source's, with the list column replaced by its element type under the +/// alias. +pub(crate) fn flattened_schema( + source_schema: &ArrowSchema, + unnest: &ViewUnnest, +) -> Result { + let field = source_schema + .field_with_name(&unnest.column) + .map_err(|_| Error::InvalidInput { + message: format!( + "UNNEST column '{}' is not a column of the source", + unnest.column + ), + })?; + let DataType::List(element) = field.data_type() else { + return Err(Error::InvalidInput { + message: format!( + "UNNEST column '{}' is {}, not a list", + unnest.column, + field.data_type() + ), + }); + }; + if source_schema.field_with_name(&unnest.alias).is_ok() { + return Err(Error::InvalidInput { + message: format!( + "UNNEST alias '{}' collides with a source column", + unnest.alias + ), + }); + } + let fields: Vec = source_schema + .fields() + .iter() + .map(|f| { + if f.name() == &unnest.column { + ArrowField::new(&unnest.alias, element.data_type().clone(), true) + } else { + f.as_ref().clone() + } + }) + .collect(); + Ok(Arc::new(ArrowSchema::new(fields))) } /// Reject any function that is not immutable: a view definition has to @@ -466,6 +629,28 @@ fn root(path: &str) -> &str { path.split('.').next().unwrap_or(path) } +/// The field a dotted `path` names, walking struct children. +fn field_at_path(schema: &ArrowSchema, path: &str) -> Option { + let mut parts = path.split('.'); + let mut field = schema.field_with_name(parts.next()?).ok()?.clone(); + for part in parts { + let DataType::Struct(children) = field.data_type() else { + return None; + }; + field = children.iter().find(|f| f.name() == part)?.as_ref().clone(); + } + Some(field) +} + +/// The source column recorded as read for `path`: for an unnested view a +/// read through the alias is a read of the list column. +fn recorded_input(definition: &MaterializedViewDefinition, path: &str) -> String { + match &definition.unnest { + Some(unnest) if root(path) == unnest.alias => unnest.column.clone(), + _ => path.to_string(), + } +} + /// The columns `expr` reads, kept as the planner reports them (a nested /// reference stays a dotted path) but resolved by root field. /// Embedding configuration rewritten for the view: entries whose columns the @@ -737,12 +922,11 @@ impl PreparedDeclaration { return Ok(output.clone()); } let name = input_column_name(source_column); - let field = self - .source_schema - .field_with_name(source_column) - .map_err(|_| Error::InvalidInput { + let field = field_at_path(&self.source_schema, source_column).ok_or_else(|| { + Error::InvalidInput { message: format!("the source has no column '{source_column}' to read"), - })?; + } + })?; if self.schema.field_with_name(&name).is_ok() { return Err(Error::ColumnAlreadyExists { name }); } @@ -753,17 +937,15 @@ impl PreparedDeclaration { .iter() .map(|f| f.as_ref().clone()) .collect(); - fields.insert( - row_id, - without_declarations(&field.as_ref().clone().with_name(name.clone())), - ); + fields.insert(row_id, without_declarations(&field.with_name(name.clone()))); self.definition.projections.push(ViewProjection { output: name.clone(), - expression: quote_identifier(source_column), + expression: source_column + .split('.') + .map(query::ident_sql) + .collect::>() + .join("."), }); - self.definition.inputs.push(source_column.to_string()); - self.definition.inputs.sort(); - self.definition.inputs.dedup(); self.lineage .entry(source_column.to_string()) .or_default() @@ -1022,38 +1204,58 @@ fn rewrite_column_definitions( Ok(()) } -/// `projections` of `None` selects every source column, as `SELECT *`; -/// `Some(&[])` declares no projected column, for a view of function -/// columns alone. -/// -/// ```no_run -/// # #![recursion_limit = "256"] -/// # use lancedb::materialized_view::prepare_declaration; -/// # async fn declare(source: &lancedb::Table) -> Result<(), Box> { -/// let prepared = prepare_declaration( -/// source, -/// Some(&[("id".into(), "id".into()), ("double".into(), "value * 2".into())]), -/// Some("value > 0"), -/// None, -/// ) -/// .await?; -/// let view = prepared.create("doubles").await?; -/// # Ok(()) -/// # } -/// ``` +/// Validate a view declaration over `source`: `projections` as +/// `(name, SQL expression)` pairs, `None` selecting every source column. +/// See [`MaterializedViewDefinition::from_sql`] for the query shape; +/// [`prepare_definition`] takes a parsed query directly. pub async fn prepare_declaration( source: &Table, projections: Option<&[(String, String)]>, filter: Option<&str>, limit: Option, ) -> Result { + let definition = MaterializedViewDefinition { + source_table: source.name().to_string(), + source_namespace: source.namespace().to_vec(), + unnest: None, + projections: projections + .unwrap_or_default() + .iter() + .map(|(output, expression)| ViewProjection { + output: output.clone(), + expression: expression.clone(), + }) + .collect(), + filter: filter.map(str::to_string), + limit, + }; + prepare_definition(source, definition).await +} + +/// Validate `definition` over `source`, the table it names. The declaration +/// is planned against the source as refresh will reach it, and the result +/// creates the view with [`PreparedDeclaration::create`]. +pub async fn prepare_definition( + source: &Table, + definition: MaterializedViewDefinition, +) -> Result { + if definition.source_table != source.name() || definition.source_namespace != source.namespace() + { + return Err(Error::InvalidInput { + message: format!( + "the query reads '{}' in namespace {:?}, but the source handle is '{}' in {:?}", + definition.source_table, + definition.source_namespace, + source.name(), + source.namespace() + ), + }); + } let Some(caller_native) = source.as_native() else { return Err(Error::NotSupported { message: "materialized views are supported only on local databases".into(), }); }; - // 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() @@ -1111,25 +1313,29 @@ pub async fn prepare_declaration( .await?; // The internal-input prefix belongs to the declaration alone; the // replan at refresh sees those projections and must accept them. - if let Some(reserved) = projections - .unwrap_or_default() + if let Some(reserved) = definition + .projections .iter() - .find(|(output, _)| output.starts_with(INPUT_COLUMN_PREFIX)) + .find(|p| p.output.starts_with(INPUT_COLUMN_PREFIX)) { return Err(Error::InvalidInput { - message: format!("view column name '{}' is reserved", reserved.0), + message: format!("view column name '{}' is reserved", reserved.output), }); } let source_schema = resolved.schema().await?; let source_metadata = source_schema.metadata().clone(); - let (definition, mut fields, lineage) = plan( - source_schema.clone(), - resolved.name(), - &source_namespace, - projections, - filter, - limit, - )?; + let Planned { + definition, + mut fields, + lineage, + .. + } = plan(source_schema.clone(), &definition)?; + // What later projections (`input_column`) are planned against: for an + // unnested view the flattened schema, where the alias is a column. + let planning_schema = match &definition.unnest { + None => source_schema.clone(), + Some(unnest) => flattened_schema(&source_schema, unnest)?, + }; fields.push(ArrowField::new( SOURCE_ROW_ID_COLUMN, DataType::UInt64, @@ -1155,7 +1361,7 @@ pub async fn prepare_declaration( Ok(PreparedDeclaration { schema: Arc::new(ArrowSchema::new_with_metadata(fields, metadata)), definition, - source_schema, + source_schema: planning_schema, lineage, internal_inputs: 0, database, @@ -1338,7 +1544,7 @@ pub struct MaterializedView { impl MaterializedView { /// Interpret `table` as a materialized view: [`Error::NotAMaterializedView`] - /// for a plain table, [`Error::NotSupported`] for a kind this version + /// for a plain table, [`Error::NotSupported`] for a query this version /// cannot refresh. pub async fn from_table(table: Table) -> Result { let info = table.base_table().materialized_view_info().await?; @@ -1651,7 +1857,7 @@ mod tests { ], filter: Some("age >= 18".into()), limit: Some(10), - inputs: vec!["age".into(), "name".into()], + unnest: None, } ); @@ -1784,7 +1990,6 @@ mod tests { .collect::>(), vec!["name", "age"] ); - assert_eq!(view.definition().inputs, vec!["age", "name"]); } #[tokio::test] @@ -1988,7 +2193,6 @@ mod tests { .execute() .await .unwrap(); - assert_eq!(view.definition().inputs, vec!["metadata.age"]); let schema = view.table().schema().await.unwrap(); assert_eq!( schema.field_with_name("age").unwrap().data_type(), @@ -2685,71 +2889,79 @@ mod tests { 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, + unnest: None, 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:?}"), - } + fn read(stored: impl Into) -> Result> { + read_definition(&HashMap::from([( + DEFINITION_META_KEY.to_string(), + stored.into(), + )])) } - /// 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. + /// What is stored is the query, under a format number; the same query + /// reads back whatever namespace the source sits in. #[test] - fn a_kind_namespace_mismatch_is_refused() { - for (kind, namespace) in [ - (SELECT_KIND, vec!["ns".to_string()]), - (NAMESPACED_SELECT_KIND, Vec::new()), + fn the_stored_layout_is_the_canonical_query() { + for (namespace, query) in [ + (Vec::new(), "SELECT name FROM people"), + (vec!["ns".to_string()], "SELECT name FROM ns.people"), ] { - 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(); + let stored = definition_to_metadata(&definition(namespace.clone())).unwrap(); + let value: serde_json::Value = serde_json::from_str(&stored).unwrap(); + assert_eq!( + value, + serde_json::json!({"kind": "query", "format": 1, "query": query}) + ); + assert_eq!( + read(stored).unwrap(), + Some(StoredDefinition::Query(definition(namespace))) + ); + } + } + + /// The structured layout written before the format number still reads, + /// under both of its kind tags, and a tag that disagrees with its + /// namespace is an error: under `select` old readers resolved it at root. + #[test] + fn legacy_layouts_read_back() { + let legacy = |kind: &str, namespace: Vec<&str>| { + serde_json::json!({ + "kind": kind, + "source_table": "people", + "source_namespace": namespace, + "projections": [{"output": "name", "expression": "name"}], + "inputs": ["name"], + }) + .to_string() + }; + assert_eq!( + read(legacy(SELECT_KIND, vec![])).unwrap(), + Some(StoredDefinition::Query(definition(Vec::new()))) + ); + assert_eq!( + read(legacy(NAMESPACED_SELECT_KIND, vec!["ns"])).unwrap(), + Some(StoredDefinition::Query(definition(vec!["ns".into()]))) + ); + assert!( + read(r#"{"kind":"select","source_table":"people","projections":[]}"#) + .unwrap() + .is_some(), + "a pre-namespace definition carries no namespace key" + ); + for (kind, namespace) in [(SELECT_KIND, vec!["ns"]), (NAMESPACED_SELECT_KIND, vec![])] { + let err = read(legacy(kind, namespace)).unwrap_err(); assert!( err.to_string() .contains("does not match its source namespace"), @@ -2758,6 +2970,100 @@ mod tests { } } + /// A newer writer's definition is reported as such, never guessed at, + /// and a definition that is not a definition at all is an error rather + /// than a plain table. + #[test] + fn a_newer_format_is_reported_not_guessed() { + assert_eq!( + read(r#"{"format":2,"query":"SELECT name FROM people"}"#).unwrap(), + Some(StoredDefinition::Newer { format: "2".into() }) + ); + assert_eq!( + read(r#"{"kind":"join"}"#).unwrap(), + Some(StoredDefinition::Newer { + format: "kind 'join'".into() + }) + ); + for stored in [ + "{}", + r#"{"format":"one"}"#, + r#"{"format":1}"#, + r#"{"format":1,"query":"SELECT name FROM people GROUP BY name"}"#, + ] { + assert!(read(stored).is_err(), "{stored}"); + } + } + + /// Planning records what the query reads of the source: a nested path + /// as itself, a read through an unnest alias as the list column. + #[test] + fn planning_records_the_source_columns_read() { + let element = DataType::Struct( + vec![ + ArrowField::new("chunk", DataType::Utf8, true), + ArrowField::new("ordinal", DataType::Int32, true), + ] + .into(), + ); + let schema = Arc::new(ArrowSchema::new(vec![ + ArrowField::new("id", DataType::Int64, false), + ArrowField::new( + "meta", + DataType::Struct(vec![ArrowField::new("title", DataType::Utf8, true)].into()), + true, + ), + ArrowField::new( + "chunks", + DataType::List(Arc::new(ArrowField::new("item", element, true))), + true, + ), + ])); + let planned = plan( + schema.clone(), + &MaterializedViewDefinition::from_sql( + "SELECT id, meta.title, c.chunk FROM docs, UNNEST(chunks) AS c WHERE c.ordinal < 3", + ) + .unwrap(), + ) + .unwrap(); + assert_eq!(planned.inputs, ["chunks", "id", "meta.title"]); + assert_eq!( + planned + .fields + .iter() + .map(|f| f.name().as_str()) + .collect::>(), + ["id", "title", "chunk"] + ); + assert_eq!( + planned.fields[2].data_type(), + &DataType::Utf8, + "the element's field is read through the alias" + ); + + let star = plan( + schema, + &MaterializedViewDefinition::from_sql("SELECT * FROM docs, UNNEST(chunks) AS c") + .unwrap(), + ) + .unwrap(); + assert_eq!( + star.definition.to_sql(), + "SELECT id, meta, c FROM docs, UNNEST(chunks) AS c" + ); + let err = plan( + Arc::new(ArrowSchema::new(vec![ArrowField::new( + "id", + DataType::Int64, + false, + )])), + &MaterializedViewDefinition::from_sql("SELECT id FROM docs, UNNEST(id) AS c").unwrap(), + ) + .unwrap_err(); + assert!(err.to_string().contains("not a list"), "{err}"); + } + /// A binding as the server records it: one Utf8 input over `input` /// bound to a nullable parameter, one Int32 output named `output`, with /// the exact schemas the durable contract requires. @@ -2876,10 +3182,10 @@ mod tests { let bindings = crate::table::computed_columns::function_bindings(&schema).unwrap(); assert_eq!(bindings.len(), 1); assert_eq!(bindings[0].binding_id(), "fb_1"); - // The stored definition is the plain select it always was. + // The stored definition is the query alone; bindings live beside it. let stored: serde_json::Value = serde_json::from_str(&schema.metadata()[DEFINITION_META_KEY]).unwrap(); - assert_eq!(stored["kind"], SELECT_KIND); + assert_eq!(stored["format"], DEFINITION_FORMAT); assert_eq!(view.definition().projections.len(), 2); assert_eq!(view.table().count_rows(None).await.unwrap(), 0); assert_eq!(conn.open_materialized_view("v").await.unwrap().name(), "v"); @@ -3013,8 +3319,7 @@ mod tests { .iter() .map(|p| (p.output.as_str(), p.expression.as_str())) .collect(); - assert_eq!(projections, [("key", "id"), ("__input_name", "`name`")]); - assert_eq!(view.definition().inputs, ["id", "name"]); + assert_eq!(projections, [("key", "id"), ("__input_name", "name")]); } /// Two outputs of one binding land at consecutive positions: each diff --git a/rust/lancedb/src/materialized_view/query.rs b/rust/lancedb/src/materialized_view/query.rs new file mode 100644 index 000000000..20d47cc46 --- /dev/null +++ b/rust/lancedb/src/materialized_view/query.rs @@ -0,0 +1,503 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The LanceDB Authors + +//! The SQL a materialized view is defined by. A definition is stored as one +//! canonical query, parsed here into the relational shape refresh maintains: +//! +//! ```sql +//! SELECT , ... FROM [ns.]table [, UNNEST(column) AS alias] +//! [WHERE predicate] [LIMIT n] +//! ``` +//! +//! Anything else is a query this engine cannot maintain yet and is refused +//! at parse time, which is also what an older engine does with a newer +//! query: fail closed, never materialize it at the wrong cardinality. + +use datafusion_sql::sqlparser::ast::{ + Expr, JoinOperator, LimitClause, ObjectNamePart, Query, SelectItem, SetExpr, Statement, + TableFactor, TableWithJoins, Value, +}; +use datafusion_sql::sqlparser::dialect::GenericDialect; +use datafusion_sql::sqlparser::keywords::{ + ALL_KEYWORDS, ALL_KEYWORDS_INDEX, RESERVED_FOR_COLUMN_ALIAS, RESERVED_FOR_IDENTIFIER, + RESERVED_FOR_TABLE_ALIAS, +}; +use datafusion_sql::sqlparser::parser::Parser; +use datafusion_sql::sqlparser::tokenizer::{Token, Tokenizer}; +use lance_datafusion::planner::Planner; + +use super::{MaterializedViewDefinition, ViewProjection, ViewUnnest}; +use crate::{Error, Result}; + +fn invalid(message: impl Into) -> Error { + Error::InvalidInput { + message: message.into(), + } +} + +const SHAPE: &str = "a materialized view is defined by \ + `SELECT columns FROM table [, UNNEST(column) AS alias] [WHERE predicate] [LIMIT n]`"; + +/// Whether `name` must be delimited to read back as this identifier: bare, +/// the parser would take it as a keyword, or a different spelling. +fn needs_quote(name: &str) -> bool { + let plain = !name.is_empty() + && name + .chars() + .enumerate() + .all(|(i, c)| c == '_' || c.is_ascii_lowercase() || (i > 0 && c.is_ascii_digit())); + if !plain { + return true; + } + let reserved = ALL_KEYWORDS + .binary_search(&name.to_ascii_uppercase().as_str()) + .is_ok_and(|i| { + let keyword = &ALL_KEYWORDS_INDEX[i]; + RESERVED_FOR_TABLE_ALIAS.contains(keyword) + || RESERVED_FOR_COLUMN_ALIAS.contains(keyword) + || RESERVED_FOR_IDENTIFIER.contains(keyword) + }); + if reserved { + return true; + } + // Then the parsers' own judgement: lance's in an expression, where a + // column name is read, and sqlparser's in a `FROM`, where a table's is. + let schema = std::sync::Arc::new(arrow_schema::Schema::new(vec![arrow_schema::Field::new( + name, + arrow_schema::DataType::Int32, + true, + )])); + let planner = Planner::new(schema); + let column = planner + .parse_expr(&format!("{name} IS NOT NULL")) + .is_ok_and(|expr| Planner::column_names_in_expr(&expr) == [name]); + if !column { + return true; + } + let sql = format!("SELECT 1 FROM {name}"); + !matches!( + Parser::parse_sql(&GenericDialect {}, &sql).as_deref(), + Ok([Statement::Query(query)]) if matches!( + query.body.as_ref(), + SetExpr::Select(select) if matches!( + select.from.as_slice(), + [TableWithJoins { relation: TableFactor::Table { name: table, .. }, .. }] + if table.to_string() == name + ) + ) + ) +} + +/// `name` as a lance SQL identifier: bare when the parser reads it back +/// unchanged, backtick-delimited otherwise. +pub fn ident_sql(name: &str) -> String { + if needs_quote(name) { + format!("`{}`", name.replace('`', "``")) + } else { + name.to_string() + } +} + +/// Rewrite every delimited identifier in `sql` to the form [`ident_sql`] +/// produces, so the same name is spelled one way wherever it appears. +/// Lance's parser delimits with backticks only; a `"name"` is rewritten +/// rather than read as a string. +pub fn canonical_tokens(sql: &str) -> Result { + let tokens = Tokenizer::new(&GenericDialect {}, sql) + .with_unescape(false) + .tokenize() + .map_err(|err| invalid(format!("invalid SQL: {err}")))?; + Ok(tokens + .into_iter() + .map(|token| match token { + Token::Word(word) if word.quote_style == Some('"') => { + ident_sql(&word.value.replace("\"\"", "\"")) + } + Token::Word(word) if word.quote_style == Some('`') => { + ident_sql(&word.value.replace("``", "`")) + } + other => other.to_string(), + }) + .collect()) +} + +/// One expression, in the spelling the stored query uses. +pub fn canonical_expr(sql: &str) -> Result { + let text = canonical_tokens(sql)?; + let expr = Parser::new(&GenericDialect {}) + .try_with_sql(&text) + .and_then(|mut parser| { + let expr = parser.parse_expr()?; + parser.expect_token(&Token::EOF)?; + Ok(expr) + }) + .map_err(|err| invalid(format!("invalid SQL expression '{sql}': {err}")))?; + Ok(expr.to_string()) +} + +/// The column a bare `SELECT` item names: the last part of a plain or +/// compound identifier, `None` for any other expression. +fn column_ref_name(expr: &Expr) -> Option { + match expr { + Expr::Identifier(ident) => Some(ident.value.clone()), + Expr::CompoundIdentifier(parts) => parts.last().map(|p| p.value.clone()), + _ => None, + } +} + +/// Parse `sql` into a definition. The query is re-rendered and compared +/// with what was parsed, so any clause this shape does not carry is +/// refused rather than dropped. +pub fn parse(sql: &str) -> Result { + let text = canonical_tokens(sql)?; + let mut statements = Parser::parse_sql(&GenericDialect {}, &text) + .map_err(|err| invalid(format!("invalid SQL: {err}")))?; + let query = match (statements.pop(), statements.is_empty()) { + (Some(Statement::Query(query)), true) => normalize_from(*query), + _ => { + return Err(invalid(format!( + "expected a single SELECT statement; {SHAPE}" + ))); + } + }; + let definition = extract(&query)?; + let rendered = render(&definition); + if canonical_tokens(&query.to_string())? != rendered { + return Err(invalid(format!( + "unsupported clause in the view query; {SHAPE}" + ))); + } + Ok(definition) +} + +/// One spelling per relation: `FROM t CROSS JOIN UNNEST(..)` is +/// `FROM t, UNNEST(..)`, and the alias always takes `AS`. +fn normalize_from(mut query: Query) -> Query { + if let SetExpr::Select(select) = query.body.as_mut() { + if select.from.len() == 1 + && select.from[0].joins.len() == 1 + && matches!( + select.from[0].joins[0].join_operator, + JoinOperator::CrossJoin(_) + ) + && matches!(select.from[0].joins[0].relation, TableFactor::UNNEST { .. }) + { + let join = select.from[0].joins.pop().expect("checked above"); + select.from.push(TableWithJoins { + relation: join.relation, + joins: Vec::new(), + }); + } + // `meta.title AS title` names what `meta.title` already names. + for item in &mut select.projection { + if let SelectItem::ExprWithAlias { expr, alias } = item + && column_ref_name(expr).as_deref() == Some(alias.value.as_str()) + { + *item = SelectItem::UnnamedExpr(expr.clone()); + } + } + // `UNNEST(c) e` and `UNNEST(c) AS e` are one alias. + if let Some(TableWithJoins { + relation: TableFactor::UNNEST { + alias: Some(alias), .. + }, + .. + }) = select.from.get_mut(1) + { + alias.explicit = true; + } + } + query +} + +fn extract(query: &Query) -> Result { + let SetExpr::Select(select) = query.body.as_ref() else { + return Err(invalid(format!("expected a SELECT; {SHAPE}"))); + }; + + let mut from = select.from.iter(); + let (source_namespace, source_table) = match from.next().map(|f| &f.relation) { + Some(TableFactor::Table { name, .. }) => { + let mut parts = Vec::with_capacity(name.0.len()); + for part in &name.0 { + match part { + ObjectNamePart::Identifier(ident) => parts.push(ident.value.clone()), + other => return Err(invalid(format!("unsupported table name part '{other}'"))), + } + } + let table = parts.pop().ok_or_else(|| invalid("empty table name"))?; + (parts, table) + } + _ => return Err(invalid(format!("the view must read one table; {SHAPE}"))), + }; + let unnest = match from.next().map(|f| &f.relation) { + None => None, + Some(TableFactor::UNNEST { + alias, array_exprs, .. + }) => { + let column = match array_exprs.as_slice() { + [Expr::Identifier(ident)] => ident.value.clone(), + _ => { + return Err(invalid( + "UNNEST takes one top-level list column of the table", + )); + } + }; + let alias = alias + .as_ref() + .ok_or_else(|| invalid("UNNEST needs an alias: `UNNEST(column) AS alias`"))?; + Some(ViewUnnest { + column, + alias: alias.name.value.clone(), + }) + } + Some(_) => return Err(invalid(format!("the view must read one table; {SHAPE}"))), + }; + if from.next().is_some() { + return Err(invalid(format!("the view must read one table; {SHAPE}"))); + } + + let mut projections = Vec::with_capacity(select.projection.len()); + for item in &select.projection { + match item { + SelectItem::Wildcard(_) if select.projection.len() == 1 => {} + SelectItem::Wildcard(_) => { + return Err(invalid("`*` must be the only column selected")); + } + SelectItem::UnnamedExpr(expr) => { + let output = column_ref_name(expr).ok_or_else(|| { + invalid(format!( + "view column `{expr}` needs a name: `{expr} AS name`" + )) + })?; + projections.push(ViewProjection { + output, + expression: expr.to_string(), + }); + } + SelectItem::ExprWithAlias { expr, alias } => projections.push(ViewProjection { + output: alias.value.clone(), + expression: expr.to_string(), + }), + other => return Err(invalid(format!("unsupported select item '{other}'"))), + } + } + + let limit = + match &query.limit_clause { + None => None, + Some(LimitClause::LimitOffset { + limit: Some(Expr::Value(value)), + offset: None, + limit_by, + }) if limit_by.is_empty() => match &value.value { + Value::Number(n, _) => Some(n.parse::().map_err(|_| { + invalid(format!("view limit {n} is not a non-negative integer")) + })?), + _ => return Err(invalid("view limit must be an integer literal")), + }, + Some(_) => return Err(invalid("view limit must be a plain `LIMIT n`")), + }; + + Ok(MaterializedViewDefinition { + source_table, + source_namespace, + unnest, + projections, + filter: select.selection.as_ref().map(|e| e.to_string()), + limit, + }) +} + +/// The canonical query for `definition`; [`parse`] reads it back equal. +pub fn render(definition: &MaterializedViewDefinition) -> String { + let mut sql = String::from("SELECT "); + if definition.projections.is_empty() { + sql.push('*'); + } else { + let items: Vec = definition + .projections + .iter() + .map(|p| { + let bare = Parser::new(&GenericDialect {}) + .try_with_sql(&p.expression) + .and_then(|mut parser| parser.parse_expr()) + .ok() + .and_then(|expr| column_ref_name(&expr)) + .is_some_and(|name| name == p.output); + if bare { + p.expression.clone() + } else { + format!("{} AS {}", p.expression, ident_sql(&p.output)) + } + }) + .collect(); + sql.push_str(&items.join(", ")); + } + sql.push_str(" FROM "); + let table: Vec = definition + .source_namespace + .iter() + .chain(std::iter::once(&definition.source_table)) + .map(|part| ident_sql(part)) + .collect(); + sql.push_str(&table.join(".")); + if let Some(unnest) = &definition.unnest { + sql.push_str(&format!( + ", UNNEST({}) AS {}", + ident_sql(&unnest.column), + ident_sql(&unnest.alias) + )); + } + if let Some(filter) = &definition.filter { + sql.push_str(&format!(" WHERE {filter}")); + } + if let Some(limit) = definition.limit { + sql.push_str(&format!(" LIMIT {limit}")); + } + sql +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_query_round_trips_through_its_canonical_form() { + for (sql, canonical) in [ + ( + r#"select "Name", x*2 as twice from ns.docs where x > 1 limit 5"#, + "SELECT `Name`, x * 2 AS twice FROM ns.docs WHERE x > 1 LIMIT 5", + ), + ( + "SELECT id, c.chunk, c.ordinal + 1 AS nth FROM docs, UNNEST(chunks) AS c WHERE c.ordinal < 5", + "SELECT id, c.chunk, c.ordinal + 1 AS nth FROM docs, UNNEST(chunks) AS c WHERE c.ordinal < 5", + ), + ( + "SELECT id FROM docs CROSS JOIN UNNEST(chunks) c", + "SELECT id FROM docs, UNNEST(chunks) AS c", + ), + ("SELECT * FROM `select`.t", "SELECT * FROM `select`.t"), + ( + "SELECT meta.title AS title, id AS id FROM t", + "SELECT meta.title, id FROM t", + ), + ] { + let definition = parse(sql).unwrap(); + assert_eq!(render(&definition), canonical, "{sql}"); + assert_eq!(parse(canonical).unwrap(), definition, "{sql}"); + } + } + + #[test] + fn parsed_parts_are_the_relational_shape() { + let definition = + parse("SELECT id, c.chunk FROM ns.docs, UNNEST(chunks) AS c WHERE id > 1").unwrap(); + assert_eq!(definition.source_namespace, ["ns"]); + assert_eq!(definition.source_table, "docs"); + assert_eq!( + definition.unnest, + Some(ViewUnnest { + column: "chunks".into(), + alias: "c".into() + }) + ); + assert_eq!( + definition.projections, + [ + ViewProjection { + output: "id".into(), + expression: "id".into() + }, + ViewProjection { + output: "chunk".into(), + expression: "c.chunk".into() + }, + ] + ); + assert_eq!(definition.filter.as_deref(), Some("id > 1")); + assert!(parse("SELECT * FROM t").unwrap().projections.is_empty()); + } + + /// A clause the engine cannot maintain is refused, never dropped. + #[test] + fn unsupported_clauses_are_refused() { + for sql in [ + "SELECT id FROM t GROUP BY id", + "SELECT id FROM t ORDER BY id", + "SELECT DISTINCT id FROM t", + "SELECT id FROM t LIMIT 5 OFFSET 2", + "SELECT id FROM t JOIN u ON t.id = u.id", + "SELECT id FROM t, u", + "SELECT id FROM t, UNNEST(c)", + "SELECT id FROM t, UNNEST(a.b) AS c", + "SELECT count(*) FROM t", + "SELECT x * 2 FROM t", + "SELECT id FROM t; SELECT id FROM t", + "SELECT *, id FROM t", + "WITH q AS (SELECT 1) SELECT id FROM t", + "SELECT id FROM t HAVING id > 1", + ] { + assert!(parse(sql).is_err(), "{sql}"); + } + } + + /// The canonical spelling is what lance's planner reads back, for the + /// expression forms a view is likely to carry. + #[test] + fn canonical_expressions_plan_in_lance() { + use arrow_schema::{DataType, Field, Schema}; + + let planner = Planner::new(std::sync::Arc::new(Schema::new(vec![ + Field::new("x", DataType::Int32, true), + Field::new("Name", DataType::Utf8, true), + Field::new( + "when", + DataType::Timestamp(arrow_schema::TimeUnit::Microsecond, None), + true, + ), + Field::new( + "meta", + DataType::Struct(vec![Field::new("title", DataType::Utf8, true)].into()), + true, + ), + ]))); + for (raw, canonical) in [ + ("x*2+1", "x * 2 + 1"), + (r#"CAST(x as bigint)"#, "CAST(x AS BIGINT)"), + (r#"upper("Name") like 'A%'"#, "upper(`Name`) LIKE 'A%'"), + ( + "x is not null and x between 1 and 3", + "x IS NOT NULL AND x BETWEEN 1 AND 3", + ), + ("meta.title", "meta.title"), + (r#"`Name` = 'it''s'"#, "`Name` = 'it''s'"), + ("x in (1, 2)", "x IN (1, 2)"), + ( + "`when` > timestamp '2024-01-01'", + "when > TIMESTAMP '2024-01-01'", + ), + ("-x", "-x"), + ] { + let text = canonical_expr(raw).unwrap(); + assert_eq!(text, canonical, "{raw}"); + let expr = planner + .parse_expr(&text) + .unwrap_or_else(|e| panic!("{text}: {e}")); + planner + .optimize_expr(expr) + .unwrap_or_else(|e| panic!("{text}: {e}")); + } + } + + #[test] + fn identifiers_are_delimited_only_when_the_parser_needs_it() { + assert_eq!(ident_sql("name"), "name"); + assert_eq!(ident_sql("Name"), "`Name`"); + assert_eq!(ident_sql("select"), "`select`"); + assert_eq!(ident_sql("1st"), "`1st`"); + assert_eq!(ident_sql("a`b"), "`a``b`"); + assert_eq!(canonical_expr(r#""Party" = 'D'"#).unwrap(), "`Party` = 'D'"); + assert_eq!(canonical_expr("`name`").unwrap(), "name"); + } +} diff --git a/rust/lancedb/src/materialized_view/refresh.rs b/rust/lancedb/src/materialized_view/refresh.rs index 62b3db22f..f0cff1768 100644 --- a/rust/lancedb/src/materialized_view/refresh.rs +++ b/rust/lancedb/src/materialized_view/refresh.rs @@ -25,9 +25,10 @@ use std::time::{SystemTime, UNIX_EPOCH}; use arrow_array::cast::AsArray; use arrow_array::types::UInt64Type; use arrow_array::{RecordBatch, UInt64Array, new_null_array}; -use arrow_schema::{FieldRef, Schema as ArrowSchema, SchemaRef}; +use arrow_schema::{DataType, Field as ArrowField, FieldRef, Schema as ArrowSchema, SchemaRef}; use datafusion::common::ScalarValue; use datafusion::error::DataFusionError; +use datafusion::physical_expr::PhysicalExpr; use datafusion::physical_plan::SendableRecordBatchStream; use datafusion::physical_plan::stream::RecordBatchStreamAdapter; use datafusion::prelude::{col, lit}; @@ -41,6 +42,8 @@ use lance::dataset::write::merge_insert::inserted_rows::{ }; use lance::dataset::{CommitBuilder, InsertBuilder, WriteDestination, WriteMode, WriteParams}; use lance_core::{ROW_CREATED_AT_VERSION, ROW_ID, ROW_LAST_UPDATED_AT_VERSION}; + +use lance_datafusion::planner::Planner; use lance_file::version::ConcreteFileVersion; use lance_table::format::Fragment; use serde::{Deserialize, Serialize}; @@ -131,12 +134,12 @@ pub(crate) async fn execute_refresh( // The definition a handle cached at open may since have been replaced; // what refresh executes and what it stamps must be one generation. - let definition = match super::materialized_view_kind(&view_ds.schema().metadata)? { - Some(super::MaterializedViewKind::Select(definition)) => definition, - Some(super::MaterializedViewKind::Unrecognized { kind }) => { + let definition = match super::read_definition(&view_ds.schema().metadata)? { + Some(super::StoredDefinition::Query(definition)) => definition, + Some(super::StoredDefinition::Newer { format }) => { return Err(Error::NotSupported { message: format!( - "materialized view '{}' is defined by '{kind}', which this \ + "materialized view '{}' is stored in format {format}, which this \ version of lancedb cannot refresh", view.name() ), @@ -164,20 +167,30 @@ pub(crate) async fn execute_refresh( // require its planned output to be exactly the view's physical schema: a // definition the stored table cannot represent must not be certified. let source_schema = Arc::new(ArrowSchema::from(source_ds.schema())); - let projections: Vec<(String, String)> = definition - .projections - .iter() - .map(|p| (p.output.clone(), p.expression.clone())) - .collect(); - validate_inputs(&source_ds, definition)?; - let (replanned, planned_fields, _renames) = super::plan( - source_schema, - &definition.source_table, - &definition.source_namespace, - Some(&projections), - definition.filter.as_deref(), - definition.limit, - )?; + let super::Planned { + definition: replanned, + fields: planned_fields, + inputs, + .. + } = super::plan(source_schema.clone(), definition).map_err(|e| match e { + // The stored query planned when the view was declared; what changed + // since is the source. + Error::InvalidExpression { column, message } => Error::Schema { + message: format!( + "view column '{column}' no longer plans against '{}' (a source column \ + was dropped or renamed): {message}", + definition.source_table + ), + }, + Error::InvalidInput { message } => Error::Schema { + message: format!( + "the stored query no longer plans against '{}' (a source column was \ + dropped or renamed): {message}", + definition.source_table + ), + }, + e => e, + })?; let mut planned_fields = planned_fields; planned_fields.push(arrow_schema::Field::new( SOURCE_ROW_ID_COLUMN, @@ -224,14 +237,18 @@ pub(crate) async fn execute_refresh( ), }); } - let definition_changed = - definition.filter != replanned.filter || definition.inputs != replanned.inputs; + // The stored query is rewritten whenever its stored form differs from + // the current one: a legacy layout, or a spelling the canonicalizer no + // longer produces. Whether the rows change is a separate question: a + // legacy raw filter like `"Party" = 'D'` read the double quotes as a + // string literal, so its watermark certifies different rows than the + // canonical predicate, and only a rebuild can replace them. + let current = definition_to_metadata(&replanned)?; + let persist = view_ds.schema().metadata.get(DEFINITION_META_KEY) != Some(¤t); + let definition_changed = !same_meaning(&source_schema, definition, &replanned); let definition = &replanned; + let persist = persist.then_some(definition); - // A watermark written for a legacy raw filter certifies the rows that - // filter produced, not the canonical predicate above. Rebuild instead of - // accepting or advancing it, and persist the migrated definition in the - // same metadata commit that certifies the replacement rows. if definition_changed { return rebuild( view_native, @@ -240,6 +257,7 @@ pub(crate) async fn execute_refresh( source_version, source_ts, definition, + &inputs, true, expected_incarnation, ) @@ -284,6 +302,7 @@ pub(crate) async fn execute_refresh( recorded_ts, full, definition, + &inputs, ) .await { @@ -296,6 +315,8 @@ pub(crate) async fn execute_refresh( source_ts, increment, definition, + &inputs, + persist, watermark, expected_incarnation, ) @@ -311,7 +332,8 @@ pub(crate) async fn execute_refresh( source_version, source_ts, definition, - false, + &inputs, + persist.is_some(), expected_incarnation, ) .await @@ -326,7 +348,8 @@ pub(crate) async fn execute_refresh( source_version, source_ts, definition, - false, + &inputs, + persist.is_some(), expected_incarnation, ) .await @@ -345,6 +368,7 @@ async fn plan_increment( recorded_ts: Option, full: bool, definition: &MaterializedViewDefinition, + inputs: &[String], ) -> Option { if full { return None; @@ -414,7 +438,7 @@ async fn plan_increment( }); } - is_pure_append(&old, source_ds, &relevant_field_ids(source_ds, definition)).then(|| Increment { + is_pure_append(&old, source_ds, &relevant_field_ids(source_ds, inputs)).then(|| Increment { appended: live .into_iter() .filter(|f| !old_ids.contains(&f.id)) @@ -567,7 +591,7 @@ fn fragment_signature(metadata: &Fragment, relevant: &HashSet) -> (u64, Str } /// Field ids (with struct descendants) of the source columns the view reads. -fn relevant_field_ids(source: &Dataset, definition: &MaterializedViewDefinition) -> HashSet { +fn relevant_field_ids(source: &Dataset, inputs: &[String]) -> HashSet { fn collect(field: &lance_core::datatypes::Field, ids: &mut HashSet) { ids.insert(field.id); for child in &field.children { @@ -575,7 +599,7 @@ fn relevant_field_ids(source: &Dataset, definition: &MaterializedViewDefinition) } } let mut ids = HashSet::new(); - for input in &definition.inputs { + for input in inputs { if let Some(field) = source.schema().field(input) { collect(field, &mut ids); } @@ -583,20 +607,45 @@ fn relevant_field_ids(source: &Dataset, definition: &MaterializedViewDefinition) ids } -/// Error if a column the view reads no longer exists in the source. -fn validate_inputs(source: &Dataset, definition: &MaterializedViewDefinition) -> Result<()> { - for input in &definition.inputs { - if source.schema().field(input).is_none() { - return Err(Error::Schema { - message: format!( - "source column '{input}' read by the view no longer exists \ - (dropped or renamed in '{}')", - definition.source_table - ), - }); - } +/// Whether two plannings of a view compute the same rows and columns: the +/// same source, unnest and limit, and expressions the planner reads as the +/// same logical expression, whatever their spelling. A definition that does +/// not plan compares as different. +fn same_meaning( + source_schema: &SchemaRef, + stored: &MaterializedViewDefinition, + replanned: &MaterializedViewDefinition, +) -> bool { + if stored.source_table != replanned.source_table + || stored.source_namespace != replanned.source_namespace + || stored.unnest != replanned.unnest + || stored.limit != replanned.limit + || stored.projections.len() != replanned.projections.len() + || stored.filter.is_some() != replanned.filter.is_some() + { + return false; } - Ok(()) + let schema = match &stored.unnest { + None => source_schema.clone(), + Some(unnest) => match super::flattened_schema(source_schema, unnest) { + Ok(schema) => schema, + Err(_) => return false, + }, + }; + let planner = Planner::new(schema); + let same_expr = |a: &str, b: &str| match (planner.parse_expr(a), planner.parse_expr(b)) { + (Ok(a), Ok(b)) => a == b, + _ => false, + }; + stored + .projections + .iter() + .zip(&replanned.projections) + .all(|(a, b)| a.output == b.output && same_expr(&a.expression, &b.expression)) + && match (&stored.filter, &replanned.filter) { + (Some(a), Some(b)) => same_expr(a, b), + _ => true, + } } /// Reject MemWAL/LSM state on a refresh participant: un-compacted tiers are @@ -656,6 +705,8 @@ async fn incremental( source_ts: u128, increment: Increment, definition: &MaterializedViewDefinition, + inputs: &[String], + persist: Option<&MaterializedViewDefinition>, watermark: Option, expected_incarnation: Option<&str>, ) -> Result> { @@ -739,7 +790,7 @@ async fn incremental( view_ds.clone(), source_version, source_ts, - None, + persist, expected_incarnation, ) .await?; @@ -761,7 +812,7 @@ async fn incremental( published, source_version, source_ts, - None, + persist, expected_incarnation, ) .await?; @@ -778,6 +829,7 @@ async fn incremental( let mut stream = compute_stream( source_ds, definition, + inputs, RowScope { fragments: Some(new_fragments), // An update rewrites whole fragments, so a fragment new at head @@ -799,6 +851,7 @@ async fn incremental( let recomputed = compute_stream( source_ds, definition, + inputs, RowScope { updated_between: Some((watermark_version, source_version)), ..Default::default() @@ -833,7 +886,7 @@ async fn incremental( published, source_version, source_ts, - None, + persist, expected_incarnation, ) .await?; @@ -883,7 +936,7 @@ async fn incremental( appended, source_version, source_ts, - None, + persist, expected_incarnation, ) .await?; @@ -898,6 +951,7 @@ async fn rebuild( source_version: u64, source_ts: u128, definition: &MaterializedViewDefinition, + inputs: &[String], persist_definition: bool, expected_incarnation: Option<&str>, ) -> Result { @@ -906,6 +960,7 @@ async fn rebuild( let stream = compute_stream( source_ds, definition, + inputs, RowScope { limit: definition.limit, ..Default::default() @@ -1202,6 +1257,7 @@ async fn only_computed_rewrites_since(view_ds: &Dataset, recorded: u64) -> Resul async fn compute_stream( source: &Dataset, definition: &MaterializedViewDefinition, + inputs: &[String], scope: RowScope, schema: SchemaRef, rows_written: Arc, @@ -1233,6 +1289,7 @@ async fn compute_stream( let clauses: Vec = definition .filter .clone() + .filter(|_| definition.unnest.is_none()) .map(|f| format!("({f})")) .into_iter() .chain(updated_filter) @@ -1241,12 +1298,23 @@ async fn compute_stream( if !clauses.is_empty() { scanner.filter(&clauses.join(" AND "))?; } - let transforms: Vec<(&str, &str)> = definition - .projections - .iter() - .map(|p| (p.output.as_str(), p.expression.as_str())) - .collect(); - scanner.project_with_transform(&transforms)?; + // An expanded view cannot project or filter in the scan: both read the + // unnested element, which exists only after the per-batch expansion. + let expanded = match &definition.unnest { + Some(unnest) => Some(UnnestPlan::new(source, definition, inputs, unnest)?), + None => { + let transforms: Vec<(&str, &str)> = definition + .projections + .iter() + .map(|p| (p.output.as_str(), p.expression.as_str())) + .collect(); + scanner.project_with_transform(&transforms)?; + None + } + }; + if let Some(expanded) = &expanded { + scanner.project(&expanded.raw_inputs)?; + } // A scan reads a limit of zero as no limit at all, so a view capped at // nothing is answered without one. if limit == Some(0) { @@ -1265,6 +1333,10 @@ async fn compute_stream( let out_schema = schema.clone(); let mapped = scanner.try_into_stream().await?.map(move |batch| { let batch = batch.map_err(|e| DataFusionError::External(Box::new(e)))?; + let batch = match &expanded { + None => batch, + Some(expanded) => expanded.apply(&batch)?, + }; let mut columns = Vec::with_capacity(out_schema.fields().len()); for field in out_schema.fields() { if computed_column_from_field(field).is_some() { @@ -1290,6 +1362,192 @@ async fn compute_stream( Ok(Box::pin(RecordBatchStreamAdapter::new(schema, mapped))) } +/// The post-scan half of an unnested view's refresh: the scan reads +/// `raw_inputs` plus the row id, and each batch is unnested on the list +/// column, filtered, then projected by expressions typed against +/// `read_schema`, where the element sits under the alias. +struct UnnestPlan { + column: String, + raw_inputs: Vec, + read_schema: SchemaRef, + projections: Vec<(String, Arc)>, + filter: Option>, +} + +impl UnnestPlan { + fn new( + source: &Dataset, + definition: &MaterializedViewDefinition, + inputs: &[String], + unnest: &super::ViewUnnest, + ) -> Result { + // Whole root columns: a nested input is projected by the expression. + let mut raw_inputs: Vec = inputs + .iter() + .map(|input| super::root(input).to_string()) + .chain(std::iter::once(unnest.column.clone())) + .collect(); + raw_inputs.sort(); + raw_inputs.dedup(); + let flattened = super::flattened_schema(&ArrowSchema::from(source.schema()), unnest)?; + // Physical expressions index columns by position, so the schema is + // exactly the scan's output: `raw_inputs` in order, then the row id. + let mut read_fields = Vec::with_capacity(raw_inputs.len() + 1); + for name in &raw_inputs { + let name = if *name == unnest.column { + &unnest.alias + } else { + name + }; + let field = flattened + .field_with_name(name) + .map_err(|_| Error::Runtime { + message: format!("source column '{name}' read by the view is missing"), + })?; + read_fields.push(field.clone()); + } + read_fields.push(ArrowField::new(ROW_ID, DataType::UInt64, false)); + let read_schema = Arc::new(ArrowSchema::new(read_fields)); + let planner = Planner::new(read_schema.clone()); + let physical = |what: &str, sql: &str| -> Result> { + let err = |e: lance::Error| Error::Runtime { + message: format!("{what}: {e}"), + }; + let parsed = planner.parse_expr(sql).map_err(err)?; + let optimized = planner.optimize_expr(parsed).map_err(err)?; + planner.create_physical_expr(&optimized).map_err(err) + }; + let mut projections = Vec::with_capacity(definition.projections.len()); + for projection in &definition.projections { + let expr = physical( + &format!("view column '{}'", projection.output), + &projection.expression, + )?; + projections.push((projection.output.clone(), expr)); + } + let filter = definition + .filter + .as_deref() + .map(|sql| physical("view filter", sql)) + .transpose()?; + Ok(Self { + column: unnest.column.clone(), + raw_inputs, + read_schema, + projections, + filter, + }) + } + + fn apply(&self, batch: &RecordBatch) -> datafusion::common::Result { + let unnested = unnest_batch(batch, &self.column) + .map_err(|e| DataFusionError::External(Box::new(e)))?; + // Same columns, renamed: the list column is now the element under the alias. + let unnested = RecordBatch::try_new(self.read_schema.clone(), unnested.columns().to_vec())?; + let unnested = match &self.filter { + None => unnested, + Some(filter) => { + let keep = filter + .evaluate(&unnested)? + .into_array(unnested.num_rows())?; + let keep = keep.as_boolean_opt().ok_or_else(|| { + DataFusionError::Internal("view filter did not evaluate to a boolean".into()) + })?; + arrow_select::filter::filter_record_batch(&unnested, keep)? + } + }; + let mut columns = Vec::with_capacity(self.projections.len() + 1); + for (output, expr) in &self.projections { + let value = expr.evaluate(&unnested)?.into_array(unnested.num_rows())?; + columns.push((output.clone(), value)); + } + let row_id = unnested + .column_by_name(ROW_ID) + .expect("scan carries the row id") + .clone(); + columns.push((ROW_ID.to_string(), row_id)); + Ok(RecordBatch::try_from_iter(columns)?) + } +} + +/// Expand `list_column` one row per element, repeating every other column +/// for each element; an empty or null list contributes no rows. The list +/// column is replaced by its element type, so a projection reads the +/// element's fields as `alias.field` after this. This is the row-cardinality +/// step of an `expanded_select` view, applied per batch on the scan stream. +fn unnest_batch(batch: &RecordBatch, list_column: &str) -> Result { + use arrow_array::{Array, ListArray, UInt32Array}; + use arrow_select::take::take; + + let (list_index, _) = batch + .schema() + .column_with_name(list_column) + .ok_or_else(|| Error::Runtime { + message: format!("expansion column '{list_column}' is not in the batch"), + })?; + let list = batch + .column(list_index) + .as_any() + .downcast_ref::() + .ok_or_else(|| Error::Runtime { + message: format!( + "expansion column '{list_column}' is {}, not a list", + batch.column(list_index).data_type() + ), + })?; + + // One take index per element, naming the source row it came from. A null + // list has no elements; its offsets are equal, so it repeats nothing. + let offsets = list.value_offsets(); + let mut repeat = Vec::with_capacity(list.values().len()); + for row in 0..list.len() { + if list.is_valid(row) { + let count = (offsets[row + 1] - offsets[row]) as usize; + repeat.extend(std::iter::repeat_n(row as u32, count)); + } + } + let repeat = UInt32Array::from(repeat); + // The flattened elements, in the same order as `repeat`: only the ranges + // valid rows cover, so a null row's stale range (if any) is skipped. + let elements = { + let mut ranges = Vec::new(); + for row in 0..list.len() { + if list.is_valid(row) { + ranges.extend((offsets[row] as u32)..(offsets[row + 1] as u32)); + } + } + take(list.values().as_ref(), &UInt32Array::from(ranges), None)? + }; + + let mut fields = Vec::with_capacity(batch.num_columns()); + let mut columns = Vec::with_capacity(batch.num_columns()); + for (index, field) in batch.schema().fields().iter().enumerate() { + if index == list_index { + let element = match field.data_type() { + arrow_schema::DataType::List(element) => element.clone(), + other => { + return Err(Error::Runtime { + message: format!("expansion column '{list_column}' is {other}, not a list"), + }); + } + }; + fields.push(Arc::new( + arrow_schema::Field::new(field.name(), element.data_type().clone(), true) + .with_metadata(field.metadata().clone()), + )); + columns.push(elements.clone()); + } else { + fields.push(field.clone()); + columns.push(take(batch.column(index).as_ref(), &repeat, None)?); + } + } + let schema = Arc::new(ArrowSchema::new_with_metadata( + fields, + batch.schema().metadata().clone(), + )); + Ok(RecordBatch::try_new(schema, columns)?) +} + /// Commit the view's removals and additions as one change, on the exact /// generation the refresh planned from. Lance rejects an overlapping /// provenance key, but an unrelated write to the view is not a key conflict, @@ -1672,6 +1930,170 @@ mod tests { (conn, source, view) } + /// The per-batch expansion behind an `expanded_select` view: every + /// element of the list column becomes a row, the other columns repeat + /// for each, and an empty or null list contributes no rows at all -- + /// which is exactly "zero rows out" for a table-valued function. + /// Four documents with a `list>` column named + /// `c`: doc 1 has two chunks, doc 2 none, doc 3 a null list, doc 4 one. + fn chunked_batch() -> RecordBatch { + use arrow_array::builder::{Int32Builder, ListBuilder, StringBuilder, StructBuilder}; + use arrow_array::{ArrayRef, Int64Array}; + use arrow_schema::{DataType, Field, Fields}; + + let element_fields = Fields::from(vec![ + Field::new("chunk", DataType::Utf8, true), + Field::new("ordinal", DataType::Int32, true), + ]); + let mut list = ListBuilder::new(StructBuilder::new( + element_fields, + vec![ + Box::new(StringBuilder::new()), + Box::new(Int32Builder::new()), + ], + )); + for chunks in [Some(vec!["a", "b"]), Some(vec![]), None, Some(vec!["c"])] { + match chunks { + Some(chunks) => { + for (i, c) in chunks.iter().enumerate() { + let s = list.values(); + s.field_builder::(0).unwrap().append_value(c); + s.field_builder::(1) + .unwrap() + .append_value(i as i32); + s.append(true); + } + list.append(true); + } + None => list.append(false), + } + } + let meta = arrow_array::StructArray::from(vec![( + Arc::new(Field::new("title", DataType::Utf8, true)), + Arc::new(arrow_array::StringArray::from(vec!["t1", "t2", "t3", "t4"])) as ArrayRef, + )]); + RecordBatch::try_from_iter(vec![ + ( + "id", + Arc::new(Int64Array::from(vec![1, 2, 3, 4])) as ArrayRef, + ), + ("meta", Arc::new(meta) as ArrayRef), + ("c", Arc::new(list.finish()) as ArrayRef), + ]) + .unwrap() + } + + #[test] + fn unnest_repeats_siblings_per_element_and_drops_empty_lists() { + use arrow_array::{StringArray, StructArray}; + use arrow_schema::{DataType, Field, Fields}; + + let batch = chunked_batch(); + let element_fields = Fields::from(vec![ + Field::new("chunk", DataType::Utf8, true), + Field::new("ordinal", DataType::Int32, true), + ]); + let out = unnest_batch(&batch, "c").unwrap(); + assert_eq!(out.num_rows(), 3, "{out:?}"); + let ids: Vec = out["id"] + .as_primitive::() + .values() + .to_vec(); + assert_eq!(ids, [1, 1, 4]); + let element = out["c"].as_any().downcast_ref::().unwrap(); + let chunks: Vec<&str> = element + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .iter() + .flatten() + .collect(); + assert_eq!(chunks, ["a", "b", "c"]); + let ordinals: Vec = element + .column(1) + .as_primitive::() + .values() + .to_vec(); + assert_eq!(ordinals, [0, 1, 0]); + // the element column is now the struct itself, not a list of it + assert_eq!( + out.schema().field_with_name("c").unwrap().data_type(), + &DataType::Struct(element_fields) + ); + } + + /// An expanded view materializes one row per list element, with the + /// projections reading the element through the alias and the other + /// source columns repeated alongside; sources with no elements yield + /// no rows. The lineage is the list column, so a change to it is what + /// drives incremental refresh. + #[tokio::test] + async fn an_expanded_view_materializes_one_row_per_element() { + use arrow_array::{Int64Array, StringArray}; + + let conn = connect("memory://").execute().await.unwrap(); + let source = conn + .create_table("docs", chunked_batch()) + .write_options(crate::materialized_view::tests::stable_row_ids()) + .execute() + .await + .unwrap(); + let mut view = crate::materialized_view::prepare_definition( + &source, + MaterializedViewDefinition::from_sql( + "SELECT id AS doc, meta.title AS title, e.ordinal + 1 AS nth \ + FROM docs, UNNEST(c) AS e WHERE e.ordinal < 5", + ) + .unwrap(), + ) + .await + .unwrap(); + // A computed column's input read through the alias is an element + // field; the recorded source input is the list column. + let text = view.input_column("e.chunk").unwrap(); + assert!(view.definition.unnest.is_some()); + let view = view.create("chunks").await.unwrap(); + view.refresh().execute().await.unwrap(); + + let batches = view + .table() + .query() + .select(Select::columns(&["doc", &text, "title"])) + .execute() + .await + .unwrap() + .try_collect::>() + .await + .unwrap(); + let out = arrow_select::concat::concat_batches(&batches[0].schema(), &batches).unwrap(); + let docs: Vec = out["doc"] + .as_any() + .downcast_ref::() + .unwrap() + .values() + .to_vec(); + let strings = |column: &str| -> Vec { + out[column] + .as_any() + .downcast_ref::() + .unwrap() + .iter() + .flatten() + .map(str::to_string) + .collect() + }; + assert_eq!(docs, [1, 1, 4]); + assert_eq!(strings(&text), ["a", "b", "c"]); + assert_eq!(strings("title"), ["t1", "t1", "t4"]); + assert_eq!(read(view.table(), "nth").await, [1, 1, 2]); + + // Appended documents expand incrementally; the existing rows stay. + source.add(chunked_batch()).execute().await.unwrap(); + view.refresh().execute().await.unwrap(); + assert_eq!(read(view.table(), "nth").await, [1, 1, 1, 1, 2, 2]); + } + async fn read(table: &Table, column: &str) -> Vec { let batches = table .query() @@ -1765,13 +2187,62 @@ mod tests { view.definition().filter.as_deref(), Some("`PartyAbbrev` = 'D'") ); - assert_eq!(view.definition().inputs, ["PartyAbbrev", "id"]); let result = view.refresh().execute().await.unwrap(); assert_eq!(result.rows_written, 2); assert_eq!(read(view.table(), "id").await, vec![1, 3]); } + /// A legacy layout whose query means what the canonical one means is + /// rewritten in the current layout on the next refresh, without a + /// rebuild: the rows it certified are the rows the query produces. + #[tokio::test] + async fn a_legacy_layout_with_the_same_meaning_is_rewritten_without_a_rebuild() { + let (conn, source, view) = refreshed_doubled(vec![1]).await; + let legacy = serde_json::json!({ + "kind": "select", + "source_table": "src", + "projections": [ + {"output": "x", "expression": "`x`"}, + {"output": "twice", "expression": "x*2"}, + ], + "inputs": ["x"], + }) + .to_string(); + let native = view.table().as_native().unwrap(); + let mut dataset = native.dataset.get().await.unwrap().as_ref().clone(); + let predicted = dataset.version().version + 1; + dataset + .update_schema_metadata([ + (DEFINITION_META_KEY.to_string(), Some(legacy)), + ( + VIEW_VERSION_META_KEY.to_string(), + Some(predicted.to_string()), + ), + ]) + .await + .unwrap(); + native.dataset.update(dataset); + + append(&source, vec![2]).await; + let reopened = conn.open_materialized_view("doubled").await.unwrap(); + let result = reopened.refresh().execute().await.unwrap(); + assert_eq!(result.mode, RefreshMode::Incremental); + assert_eq!(read(reopened.table(), "twice").await, vec![2, 4]); + let stored: serde_json::Value = serde_json::from_str( + &reopened.table().schema().await.unwrap().metadata()[DEFINITION_META_KEY], + ) + .unwrap(); + assert_eq!( + stored, + serde_json::json!({ + "kind": "query", + "format": 1, + "query": "SELECT x, x * 2 AS twice FROM src", + }) + ); + } + #[tokio::test] async fn test_legacy_raw_filter_rebuilds_and_persists_canonical_definition() { let conn = connect("memory://").execute().await.unwrap(); @@ -1797,18 +2268,20 @@ mod tests { // Model a definition and up-to-date watermark written before filter // canonicalization was applied to materialized views. - let mut legacy = view.definition().clone(); - legacy.filter = Some(r#""PartyAbbrev" = 'D'"#.into()); - legacy.inputs = vec!["id".into()]; + let legacy = serde_json::json!({ + "kind": "select", + "source_table": "legacy_src", + "projections": [{"output": "id", "expression": "id"}], + "filter": r#""PartyAbbrev" = 'D'"#, + "inputs": ["id"], + }) + .to_string(); let native = view.table().as_native().unwrap(); let mut dataset = native.dataset.get().await.unwrap().as_ref().clone(); let predicted = dataset.version().version + 1; dataset .update_schema_metadata([ - ( - DEFINITION_META_KEY.to_string(), - Some(definition_to_metadata(&legacy).unwrap()), - ), + (DEFINITION_META_KEY.to_string(), Some(legacy)), ( VIEW_VERSION_META_KEY.to_string(), Some(predicted.to_string()), @@ -1831,7 +2304,6 @@ mod tests { migrated.definition().filter.as_deref(), Some("`PartyAbbrev` = 'D'") ); - assert_eq!(migrated.definition().inputs, ["PartyAbbrev", "id"]); assert_eq!( migrated.refresh().execute().await.unwrap().mode, RefreshMode::NoOp @@ -2758,7 +3230,11 @@ mod tests { source.drop_columns(&["x"]).await.unwrap(); let err = view.refresh().execute().await.unwrap_err(); - assert!(matches!(err, Error::Schema { message } if message.contains("'x'"))); + assert!( + matches!(&err, Error::Schema { message } + if message.contains("dropped or renamed") && message.contains("No field named x")), + "{err:?}" + ); } /// A pinned refresh materializes the source as of `version`; catching up @@ -3063,7 +3539,7 @@ mod tests { ], filter: None, limit: None, - inputs: vec!["x".into()], + unnest: None, }; let mut metadata = HashMap::new(); metadata.insert( @@ -3097,7 +3573,7 @@ mod tests { }], filter: None, limit: None, - inputs: vec!["x".into()], + unnest: None, }; let mut metadata = HashMap::new(); metadata.insert( diff --git a/rust/lancedb/src/remote/table.rs b/rust/lancedb/src/remote/table.rs index 75cf8cd00..053f9ef8d 100644 --- a/rust/lancedb/src/remote/table.rs +++ b/rust/lancedb/src/remote/table.rs @@ -2100,8 +2100,6 @@ impl BaseTable for RemoteTable { #[serde(default)] limit: Option, #[serde(default)] - inputs: Vec, - #[serde(default)] incarnation: Option, } @@ -2117,6 +2115,9 @@ impl BaseTable for RemoteTable { definition: MaterializedViewDefinition { source_table: response.source_table, source_namespace: response.source_namespace, + // The describe endpoint carries the structured shape, which + // has no unnest. + unnest: None, projections: response .projections .into_iter() @@ -2127,7 +2128,6 @@ impl BaseTable for RemoteTable { .collect(), filter: response.filter, limit: response.limit, - inputs: response.inputs, }, incarnation: response.incarnation, }) @@ -12269,7 +12269,8 @@ mod tests { let view = crate::MaterializedView::from_table(table).await.unwrap(); assert_eq!(view.definition().source_table, "source"); assert_eq!(view.definition().source_namespace, ["analytics"]); - assert_eq!(view.definition().inputs, ["x"]); + // The describe endpoint carries no unnest. + assert_eq!(view.definition().unnest, None); assert_eq!(view.incarnation(), Some("inc-1")); let result = view diff --git a/rust/lancedb/src/table/merge/lsm.rs b/rust/lancedb/src/table/merge/lsm.rs index 1a1f0b28c..fc925c014 100644 --- a/rust/lancedb/src/table/merge/lsm.rs +++ b/rust/lancedb/src/table/merge/lsm.rs @@ -104,7 +104,7 @@ 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() { + if crate::materialized_view::read_definition(&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"