mirror of
https://github.com/lancedb/lancedb.git
synced 2026-09-21 12:35:42 +00:00
feat: define a materialized view by its query, with Functions in FROM position (#4190)
A view definition was a structured record under a `kind` tag, one kind per query shape, and a Function returning `list<struct>` 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: ```sql SELECT columns FROM [ns.]table [, function(args) AS alias | , 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. A Function in FROM position yields one row per element it returns, as a table function does in any dialect. The server stages its list output in a hidden table and records that binding beside the query, which stays as the user wrote it; refresh scans the staging and unnests the column, the same operator as UNNEST over a list column the table already holds. Row ids repeat per element, so eviction and incremental append are unchanged. A local database refuses a Function in FROM position, since it has no executor.
This commit is contained in:
Wyatt Alt
@@ -6,64 +6,17 @@
|
||||
|
||||
# Interface: MaterializedViewDefinition
|
||||
|
||||
The query that defines a materialized view.
|
||||
The query that defines a materialized view, as stored:
|
||||
`SELECT columns FROM [ns.]table [, function(args) AS alias | , UNNEST(column) AS alias]
|
||||
[WHERE predicate] [LIMIT n]`. A Function in `FROM` position yields one row
|
||||
per element it returns.
|
||||
|
||||
## 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.
|
||||
|
||||
@@ -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 () => {
|
||||
|
||||
@@ -7,20 +7,18 @@ 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": "<SQL>"}`. */
|
||||
export const DEFINITION_FORMAT = 1;
|
||||
|
||||
/**
|
||||
* The query that defines a materialized view, as stored:
|
||||
* `SELECT columns FROM [ns.]table [, function(args) AS alias | , UNNEST(column) AS alias]
|
||||
* [WHERE predicate] [LIMIT n]`. A Function in `FROM` position yields one row
|
||||
* per element it returns.
|
||||
*/
|
||||
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 +86,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 +112,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;
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
+1
-1
@@ -479,7 +479,7 @@ impl Table {
|
||||
let view = lancedb::MaterializedView::from_table(inner)
|
||||
.await
|
||||
.default_error()?;
|
||||
serde_json::to_string(view.definition()).map_err(|err| {
|
||||
view.definition().to_json().map_err(|err| {
|
||||
napi::Error::from_reason(format!(
|
||||
"failed to serialize materialized-view definition: {err}"
|
||||
))
|
||||
|
||||
@@ -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,24 @@ SelectArg = Union[
|
||||
]
|
||||
|
||||
|
||||
DEFINITION_FORMAT = 1
|
||||
"""The stored layout this version reads: ``{"format": 1, "query": "<SQL>"}``.
|
||||
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 [, function(args) AS alias | , UNNEST(column) AS alias]
|
||||
[WHERE predicate] [LIMIT n]
|
||||
|
||||
A Function in ``FROM`` position yields one row per element it returns.
|
||||
"""
|
||||
|
||||
query: str
|
||||
"""The defining query, in the canonical spelling the server stores."""
|
||||
|
||||
|
||||
def _definition_from_schema(
|
||||
@@ -54,38 +56,64 @@ def _definition_from_schema(
|
||||
raw = metadata.get(DEFINITION_META_KEY)
|
||||
if raw is None:
|
||||
raise ValueError(f"Table '{name}' is not a materialized view")
|
||||
value = json.loads(raw)
|
||||
return _definition_from_value(json.loads(raw), name)
|
||||
|
||||
|
||||
def _definition_from_json(raw: str, name: str = "") -> MaterializedViewDefinition:
|
||||
"""Parse the definition native code hands over, in its stored layout."""
|
||||
return _definition_from_value(json.loads(raw), name)
|
||||
|
||||
|
||||
def _definition_from_value(value: dict, name: str) -> MaterializedViewDefinition:
|
||||
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:
|
||||
|
||||
@@ -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)
|
||||
|
||||
+1
-1
@@ -1814,7 +1814,7 @@ impl Table {
|
||||
let view = lancedb::MaterializedView::from_table(inner)
|
||||
.await
|
||||
.infer_error()?;
|
||||
serde_json::to_string(view.definition()).map_err(|err| {
|
||||
view.definition().to_json().map_err(|err| {
|
||||
PyRuntimeError::new_err(format!(
|
||||
"failed to serialize materialized-view definition: {err}"
|
||||
))
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,633 @@
|
||||
// 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 <column | expr AS name | *>, ...
|
||||
//! FROM [ns.]table [, function(args) AS alias | , UNNEST(column) AS alias]
|
||||
//! [WHERE predicate] [LIMIT n]
|
||||
//! ```
|
||||
//!
|
||||
//! A Function in `FROM` position yields one row per element it returns;
|
||||
//! `UNNEST` does the same for a list column the table already holds.
|
||||
//!
|
||||
//! 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, FunctionArg, FunctionArgExpr, JoinOperator, LimitClause, ObjectName, ObjectNamePart,
|
||||
Query, SelectItem, SetExpr, Statement, TableFactor, TableFunctionArgs, 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::{LateralSource, MaterializedViewDefinition, ViewLateral, ViewProjection};
|
||||
use crate::{Error, Result};
|
||||
|
||||
fn invalid(message: impl Into<String>) -> Error {
|
||||
Error::InvalidInput {
|
||||
message: message.into(),
|
||||
}
|
||||
}
|
||||
|
||||
const SHAPE: &str = "a materialized view is defined by `SELECT columns FROM table \
|
||||
[, function(args) AS alias | , 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<String> {
|
||||
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<String> {
|
||||
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<String> {
|
||||
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<MaterializedViewDefinition> {
|
||||
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(_)
|
||||
)
|
||||
&& is_lateral_item(&select.from[0].joins[0].relation)
|
||||
{
|
||||
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());
|
||||
}
|
||||
}
|
||||
if let Some(item) = select.from.get_mut(1) {
|
||||
// `LATERAL f(x) AS c` and `f(x) AS c` are one relation: a function
|
||||
// in FROM position is lateral by nature.
|
||||
if let TableFactor::Function {
|
||||
name, args, alias, ..
|
||||
} = &item.relation
|
||||
{
|
||||
item.relation = TableFactor::Table {
|
||||
name: name.clone(),
|
||||
alias: alias.clone(),
|
||||
args: Some(TableFunctionArgs {
|
||||
args: args.clone(),
|
||||
settings: None,
|
||||
}),
|
||||
with_hints: Vec::new(),
|
||||
version: None,
|
||||
with_ordinality: false,
|
||||
partitions: Vec::new(),
|
||||
json_path: None,
|
||||
sample: None,
|
||||
index_hints: Vec::new(),
|
||||
};
|
||||
}
|
||||
// `UNNEST(c) e` and `f(x) e` take `AS`.
|
||||
match &mut item.relation {
|
||||
TableFactor::UNNEST {
|
||||
alias: Some(alias), ..
|
||||
}
|
||||
| TableFactor::Table {
|
||||
alias: Some(alias), ..
|
||||
} => alias.explicit = true,
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
}
|
||||
query
|
||||
}
|
||||
|
||||
fn is_lateral_item(factor: &TableFactor) -> bool {
|
||||
matches!(
|
||||
factor,
|
||||
TableFactor::UNNEST { .. }
|
||||
| TableFactor::Function { .. }
|
||||
| TableFactor::Table { args: Some(_), .. }
|
||||
)
|
||||
}
|
||||
|
||||
fn single_name(name: &ObjectName, what: &str) -> Result<String> {
|
||||
match name.0.as_slice() {
|
||||
[ObjectNamePart::Identifier(ident)] => Ok(ident.value.clone()),
|
||||
_ => Err(invalid(format!(
|
||||
"{what} must be a single name, not '{name}'"
|
||||
))),
|
||||
}
|
||||
}
|
||||
|
||||
fn extract(query: &Query) -> Result<MaterializedViewDefinition> {
|
||||
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 { args: Some(_), .. }) => {
|
||||
return Err(invalid(
|
||||
"a view reads a table; a Function in FROM position follows it: \
|
||||
`FROM table, function(args) AS alias`",
|
||||
));
|
||||
}
|
||||
Some(TableFactor::Table {
|
||||
alias: Some(alias), ..
|
||||
}) => {
|
||||
return Err(invalid(format!(
|
||||
"table aliases are not supported (`AS {}`); refer to columns unqualified",
|
||||
alias.name
|
||||
)));
|
||||
}
|
||||
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 lateral = 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(ViewLateral {
|
||||
source: LateralSource::Unnest { column },
|
||||
alias: alias.name.value.clone(),
|
||||
})
|
||||
}
|
||||
Some(TableFactor::Table {
|
||||
name,
|
||||
args: Some(TableFunctionArgs { args, .. }),
|
||||
alias,
|
||||
..
|
||||
}) => {
|
||||
let function = single_name(name, "a Function in FROM position")?;
|
||||
let mut rendered = Vec::with_capacity(args.len());
|
||||
for arg in args {
|
||||
match arg {
|
||||
FunctionArg::Unnamed(FunctionArgExpr::Expr(expr)) => {
|
||||
rendered.push(expr.to_string())
|
||||
}
|
||||
other => {
|
||||
return Err(invalid(format!(
|
||||
"'{function}' takes positional expression arguments, not '{other}'"
|
||||
)));
|
||||
}
|
||||
}
|
||||
}
|
||||
let alias = alias.as_ref().ok_or_else(|| {
|
||||
invalid(format!(
|
||||
"'{function}' in FROM position needs an alias: `{function}(...) AS alias`"
|
||||
))
|
||||
})?;
|
||||
Some(ViewLateral {
|
||||
source: LateralSource::Function {
|
||||
name: function,
|
||||
args: rendered,
|
||||
},
|
||||
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 => {
|
||||
projections.push(ViewProjection::star());
|
||||
}
|
||||
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::<u64>().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,
|
||||
lateral,
|
||||
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.selects_star() {
|
||||
sql.push('*');
|
||||
} else {
|
||||
let items: Vec<String> = 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<String> = 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(lateral) = &definition.lateral {
|
||||
match &lateral.source {
|
||||
LateralSource::Unnest { column } => {
|
||||
sql.push_str(&format!(", UNNEST({})", ident_sql(column)))
|
||||
}
|
||||
LateralSource::Function { name, args } => {
|
||||
sql.push_str(&format!(", {}({})", ident_sql(name), args.join(", ")))
|
||||
}
|
||||
}
|
||||
sql.push_str(&format!(" AS {}", ident_sql(&lateral.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 d.id, c.text from docs, chunk(body, 512) c where c.ordinal < 3",
|
||||
"SELECT d.id, c.text FROM docs, chunk(body, 512) AS c WHERE c.ordinal < 3",
|
||||
),
|
||||
(
|
||||
"SELECT id, c.text FROM docs CROSS JOIN LATERAL chunk(body) AS c",
|
||||
"SELECT id, c.text FROM docs, chunk(body) AS c",
|
||||
),
|
||||
(
|
||||
"SELECT id, c.text FROM docs, LATERAL chunk(upper(body)) AS c",
|
||||
"SELECT id, c.text FROM docs, chunk(upper(body)) 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.lateral,
|
||||
Some(ViewLateral {
|
||||
source: LateralSource::Unnest {
|
||||
column: "chunks".into()
|
||||
},
|
||||
alias: "c".into()
|
||||
})
|
||||
);
|
||||
let function = parse("SELECT id, c.text FROM docs, chunk(body, 512) AS c").unwrap();
|
||||
assert_eq!(
|
||||
function.lateral,
|
||||
Some(ViewLateral {
|
||||
source: LateralSource::Function {
|
||||
name: "chunk".into(),
|
||||
args: vec!["body".into(), "512".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().selects_star());
|
||||
}
|
||||
|
||||
/// 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 id FROM t, chunk(body)",
|
||||
"SELECT id FROM t, ns.chunk(body) AS c",
|
||||
"SELECT id FROM t, chunk(size => 5) AS c",
|
||||
"SELECT id FROM t AS d, chunk(d.body) AS c",
|
||||
"SELECT id FROM chunk(body) AS c",
|
||||
"SELECT id FROM t, chunk(body) AS c, UNNEST(x) AS u",
|
||||
"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");
|
||||
}
|
||||
}
|
||||
@@ -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()
|
||||
),
|
||||
@@ -149,9 +152,10 @@ pub(crate) async fn execute_refresh(
|
||||
}
|
||||
};
|
||||
let definition = &definition;
|
||||
let staging = super::read_staging(&view_ds.schema().metadata)?;
|
||||
ensure_no_mem_wal(&view_ds, "materialized view", view.name()).await?;
|
||||
|
||||
let source_ds = open_source(view, definition).await?;
|
||||
let source_ds = open_source(view, definition, staging.as_ref()).await?;
|
||||
let source_ds = match pinned {
|
||||
Some(version) => source_ds.checkout_version(version).await?,
|
||||
None => source_ds,
|
||||
@@ -164,20 +168,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, staging.as_ref()).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 +238,19 @@ 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 unnest = super::physical_unnest(&replanned, staging.as_ref())?;
|
||||
let definition_changed = !same_meaning(&source_schema, definition, &replanned, unnest.as_ref());
|
||||
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 +259,8 @@ pub(crate) async fn execute_refresh(
|
||||
source_version,
|
||||
source_ts,
|
||||
definition,
|
||||
&inputs,
|
||||
unnest.as_ref(),
|
||||
true,
|
||||
expected_incarnation,
|
||||
)
|
||||
@@ -284,6 +305,7 @@ pub(crate) async fn execute_refresh(
|
||||
recorded_ts,
|
||||
full,
|
||||
definition,
|
||||
&inputs,
|
||||
)
|
||||
.await
|
||||
{
|
||||
@@ -296,6 +318,9 @@ pub(crate) async fn execute_refresh(
|
||||
source_ts,
|
||||
increment,
|
||||
definition,
|
||||
&inputs,
|
||||
unnest.as_ref(),
|
||||
persist,
|
||||
watermark,
|
||||
expected_incarnation,
|
||||
)
|
||||
@@ -311,7 +336,9 @@ pub(crate) async fn execute_refresh(
|
||||
source_version,
|
||||
source_ts,
|
||||
definition,
|
||||
false,
|
||||
&inputs,
|
||||
unnest.as_ref(),
|
||||
persist.is_some(),
|
||||
expected_incarnation,
|
||||
)
|
||||
.await
|
||||
@@ -326,7 +353,9 @@ pub(crate) async fn execute_refresh(
|
||||
source_version,
|
||||
source_ts,
|
||||
definition,
|
||||
false,
|
||||
&inputs,
|
||||
unnest.as_ref(),
|
||||
persist.is_some(),
|
||||
expected_incarnation,
|
||||
)
|
||||
.await
|
||||
@@ -345,6 +374,7 @@ async fn plan_increment(
|
||||
recorded_ts: Option<u128>,
|
||||
full: bool,
|
||||
definition: &MaterializedViewDefinition,
|
||||
inputs: &[String],
|
||||
) -> Option<Increment> {
|
||||
if full {
|
||||
return None;
|
||||
@@ -414,7 +444,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 +597,7 @@ fn fragment_signature(metadata: &Fragment, relevant: &HashSet<i32>) -> (u64, Str
|
||||
}
|
||||
|
||||
/// Field ids (with struct descendants) of the source columns the view reads.
|
||||
fn relevant_field_ids(source: &Dataset, definition: &MaterializedViewDefinition) -> HashSet<i32> {
|
||||
fn relevant_field_ids(source: &Dataset, inputs: &[String]) -> HashSet<i32> {
|
||||
fn collect(field: &lance_core::datatypes::Field, ids: &mut HashSet<i32>) {
|
||||
ids.insert(field.id);
|
||||
for child in &field.children {
|
||||
@@ -575,7 +605,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 +613,46 @@ 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,
|
||||
unnest: Option<&super::ViewUnnest>,
|
||||
) -> bool {
|
||||
if stored.source_table != replanned.source_table
|
||||
|| stored.source_namespace != replanned.source_namespace
|
||||
|| stored.lateral != replanned.lateral
|
||||
|| stored.limit != replanned.limit
|
||||
|| stored.projections.len() != replanned.projections.len()
|
||||
|| stored.filter.is_some() != replanned.filter.is_some()
|
||||
{
|
||||
return false;
|
||||
}
|
||||
Ok(())
|
||||
let schema = match 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
|
||||
@@ -616,14 +672,27 @@ pub(crate) async fn ensure_no_mem_wal(dataset: &Dataset, role: &str, name: &str)
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn open_source(view: &Table, definition: &MaterializedViewDefinition) -> Result<Dataset> {
|
||||
/// The table refresh scans: the staging table when the query calls a
|
||||
/// Function in FROM position, otherwise the query's source.
|
||||
async fn open_source(
|
||||
view: &Table,
|
||||
definition: &MaterializedViewDefinition,
|
||||
staging: Option<&super::StagingBinding>,
|
||||
) -> Result<Dataset> {
|
||||
let database = view.database_opt().ok_or_else(|| Error::InvalidInput {
|
||||
message: "the view was not opened through a database connection".into(),
|
||||
})?;
|
||||
let (name, namespace_path) = match staging {
|
||||
Some(staging) => (staging.table.clone(), staging.namespace.clone()),
|
||||
None => (
|
||||
definition.source_table.clone(),
|
||||
definition.source_namespace.clone(),
|
||||
),
|
||||
};
|
||||
let source = database
|
||||
.open_table(OpenTableRequest {
|
||||
name: definition.source_table.clone(),
|
||||
namespace_path: definition.source_namespace.clone(),
|
||||
name,
|
||||
namespace_path,
|
||||
index_cache_size: None,
|
||||
lance_read_params: None,
|
||||
location: None,
|
||||
@@ -656,6 +725,9 @@ async fn incremental(
|
||||
source_ts: u128,
|
||||
increment: Increment,
|
||||
definition: &MaterializedViewDefinition,
|
||||
inputs: &[String],
|
||||
unnest: Option<&super::ViewUnnest>,
|
||||
persist: Option<&MaterializedViewDefinition>,
|
||||
watermark: Option<u64>,
|
||||
expected_incarnation: Option<&str>,
|
||||
) -> Result<Option<RefreshMaterializedViewResult>> {
|
||||
@@ -739,7 +811,7 @@ async fn incremental(
|
||||
view_ds.clone(),
|
||||
source_version,
|
||||
source_ts,
|
||||
None,
|
||||
persist,
|
||||
expected_incarnation,
|
||||
)
|
||||
.await?;
|
||||
@@ -761,7 +833,7 @@ async fn incremental(
|
||||
published,
|
||||
source_version,
|
||||
source_ts,
|
||||
None,
|
||||
persist,
|
||||
expected_incarnation,
|
||||
)
|
||||
.await?;
|
||||
@@ -778,6 +850,8 @@ async fn incremental(
|
||||
let mut stream = compute_stream(
|
||||
source_ds,
|
||||
definition,
|
||||
inputs,
|
||||
unnest,
|
||||
RowScope {
|
||||
fragments: Some(new_fragments),
|
||||
// An update rewrites whole fragments, so a fragment new at head
|
||||
@@ -799,6 +873,8 @@ async fn incremental(
|
||||
let recomputed = compute_stream(
|
||||
source_ds,
|
||||
definition,
|
||||
inputs,
|
||||
unnest,
|
||||
RowScope {
|
||||
updated_between: Some((watermark_version, source_version)),
|
||||
..Default::default()
|
||||
@@ -833,7 +909,7 @@ async fn incremental(
|
||||
published,
|
||||
source_version,
|
||||
source_ts,
|
||||
None,
|
||||
persist,
|
||||
expected_incarnation,
|
||||
)
|
||||
.await?;
|
||||
@@ -883,7 +959,7 @@ async fn incremental(
|
||||
appended,
|
||||
source_version,
|
||||
source_ts,
|
||||
None,
|
||||
persist,
|
||||
expected_incarnation,
|
||||
)
|
||||
.await?;
|
||||
@@ -898,6 +974,8 @@ async fn rebuild(
|
||||
source_version: u64,
|
||||
source_ts: u128,
|
||||
definition: &MaterializedViewDefinition,
|
||||
inputs: &[String],
|
||||
unnest: Option<&super::ViewUnnest>,
|
||||
persist_definition: bool,
|
||||
expected_incarnation: Option<&str>,
|
||||
) -> Result<RefreshMaterializedViewResult> {
|
||||
@@ -906,6 +984,8 @@ async fn rebuild(
|
||||
let stream = compute_stream(
|
||||
source_ds,
|
||||
definition,
|
||||
inputs,
|
||||
unnest,
|
||||
RowScope {
|
||||
limit: definition.limit,
|
||||
..Default::default()
|
||||
@@ -1202,6 +1282,8 @@ async fn only_computed_rewrites_since(view_ds: &Dataset, recorded: u64) -> Resul
|
||||
async fn compute_stream(
|
||||
source: &Dataset,
|
||||
definition: &MaterializedViewDefinition,
|
||||
inputs: &[String],
|
||||
unnest: Option<&super::ViewUnnest>,
|
||||
scope: RowScope,
|
||||
schema: SchemaRef,
|
||||
rows_written: Arc<AtomicU64>,
|
||||
@@ -1233,6 +1315,7 @@ async fn compute_stream(
|
||||
let clauses: Vec<String> = definition
|
||||
.filter
|
||||
.clone()
|
||||
.filter(|_| unnest.is_none())
|
||||
.map(|f| format!("({f})"))
|
||||
.into_iter()
|
||||
.chain(updated_filter)
|
||||
@@ -1241,12 +1324,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 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 +1359,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 +1388,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<String>,
|
||||
read_schema: SchemaRef,
|
||||
projections: Vec<(String, Arc<dyn PhysicalExpr>)>,
|
||||
filter: Option<Arc<dyn PhysicalExpr>>,
|
||||
}
|
||||
|
||||
impl UnnestPlan {
|
||||
fn new(
|
||||
source: &Dataset,
|
||||
definition: &MaterializedViewDefinition,
|
||||
inputs: &[String],
|
||||
unnest: &super::ViewUnnest,
|
||||
) -> Result<Self> {
|
||||
// Whole root columns: a nested input is projected by the expression.
|
||||
let mut raw_inputs: Vec<String> = 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<Arc<dyn PhysicalExpr>> {
|
||||
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<RecordBatch> {
|
||||
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<RecordBatch> {
|
||||
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::<ListArray>()
|
||||
.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 +1956,247 @@ 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<struct<chunk, ordinal>>` 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::<StringBuilder>(0).unwrap().append_value(c);
|
||||
s.field_builder::<Int32Builder>(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<i64> = out["id"]
|
||||
.as_primitive::<arrow_array::types::Int64Type>()
|
||||
.values()
|
||||
.to_vec();
|
||||
assert_eq!(ids, [1, 1, 4]);
|
||||
let element = out["c"].as_any().downcast_ref::<StructArray>().unwrap();
|
||||
let chunks: Vec<&str> = element
|
||||
.column(0)
|
||||
.as_any()
|
||||
.downcast_ref::<StringArray>()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.flatten()
|
||||
.collect();
|
||||
assert_eq!(chunks, ["a", "b", "c"]);
|
||||
let ordinals: Vec<i32> = element
|
||||
.column(1)
|
||||
.as_primitive::<arrow_array::types::Int32Type>()
|
||||
.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)
|
||||
);
|
||||
}
|
||||
|
||||
/// A Function in FROM position is refreshed from its staging table: the
|
||||
/// query on the view stays the one the user wrote, the staging binding
|
||||
/// says which table and list column refresh reads, and rows come out one
|
||||
/// per element as with UNNEST. Without a staging, a local database
|
||||
/// refuses the query rather than guess.
|
||||
#[tokio::test]
|
||||
async fn a_function_in_from_position_refreshes_from_its_staging() {
|
||||
use arrow_array::StringArray;
|
||||
|
||||
let conn = connect("memory://").execute().await.unwrap();
|
||||
let staging = conn
|
||||
.create_table("docs__chunk", chunked_batch())
|
||||
.write_options(crate::materialized_view::tests::stable_row_ids())
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
let query =
|
||||
"SELECT id AS doc, e.chunk AS text, e.ordinal FROM docs, chunk(meta.title, 2) AS e";
|
||||
let definition = MaterializedViewDefinition::from_sql(query).unwrap();
|
||||
|
||||
let err = crate::materialized_view::prepare_definition(&staging, definition.clone())
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert!(
|
||||
matches!(&err, Error::InvalidInput { message } if message.contains("reads 'docs'")),
|
||||
"{err:?}"
|
||||
);
|
||||
|
||||
let view = crate::materialized_view::prepare_staged_definition(&staging, definition, "c")
|
||||
.await
|
||||
.unwrap()
|
||||
.create("chunks")
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(view.definition().to_sql(), query);
|
||||
let metadata = view.table().schema().await.unwrap().metadata().clone();
|
||||
assert_eq!(
|
||||
crate::materialized_view::read_staging(&metadata).unwrap(),
|
||||
Some(crate::materialized_view::StagingBinding {
|
||||
table: "docs__chunk".into(),
|
||||
namespace: Vec::new(),
|
||||
column: "c".into(),
|
||||
})
|
||||
);
|
||||
|
||||
view.refresh().execute().await.unwrap();
|
||||
assert_eq!(read(view.table(), "ordinal").await, [0, 0, 1]);
|
||||
let batches = view
|
||||
.table()
|
||||
.query()
|
||||
.select(Select::columns(&["text"]))
|
||||
.execute()
|
||||
.await
|
||||
.unwrap()
|
||||
.try_collect::<Vec<_>>()
|
||||
.await
|
||||
.unwrap();
|
||||
let out = arrow_select::concat::concat_batches(&batches[0].schema(), &batches).unwrap();
|
||||
let texts: Vec<&str> = out["text"]
|
||||
.as_any()
|
||||
.downcast_ref::<StringArray>()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.flatten()
|
||||
.collect();
|
||||
assert_eq!(texts, ["a", "b", "c"]);
|
||||
|
||||
// The reopened view reads the same logical query and refreshes
|
||||
// incrementally from the staging.
|
||||
staging.add(chunked_batch()).execute().await.unwrap();
|
||||
let reopened = conn.open_materialized_view("chunks").await.unwrap();
|
||||
assert_eq!(reopened.definition().to_sql(), query);
|
||||
let result = reopened.refresh().execute().await.unwrap();
|
||||
assert_eq!(result.mode, RefreshMode::Incremental);
|
||||
assert_eq!(read(reopened.table(), "ordinal").await, [0, 0, 0, 0, 1, 1]);
|
||||
}
|
||||
|
||||
/// 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.lateral.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::<Vec<_>>()
|
||||
.await
|
||||
.unwrap();
|
||||
let out = arrow_select::concat::concat_batches(&batches[0].schema(), &batches).unwrap();
|
||||
let docs: Vec<i64> = out["doc"]
|
||||
.as_any()
|
||||
.downcast_ref::<Int64Array>()
|
||||
.unwrap()
|
||||
.values()
|
||||
.to_vec();
|
||||
let strings = |column: &str| -> Vec<String> {
|
||||
out[column]
|
||||
.as_any()
|
||||
.downcast_ref::<StringArray>()
|
||||
.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<i32> {
|
||||
let batches = table
|
||||
.query()
|
||||
@@ -1765,13 +2290,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 +2371,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 +2407,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 +3333,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 +3642,7 @@ mod tests {
|
||||
],
|
||||
filter: None,
|
||||
limit: None,
|
||||
inputs: vec!["x".into()],
|
||||
lateral: None,
|
||||
};
|
||||
let mut metadata = HashMap::new();
|
||||
metadata.insert(
|
||||
@@ -3097,7 +3676,7 @@ mod tests {
|
||||
}],
|
||||
filter: None,
|
||||
limit: None,
|
||||
inputs: vec!["x".into()],
|
||||
lateral: None,
|
||||
};
|
||||
let mut metadata = HashMap::new();
|
||||
metadata.insert(
|
||||
|
||||
@@ -2099,8 +2099,10 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
|
||||
filter: Option<String>,
|
||||
#[serde(default)]
|
||||
limit: Option<u64>,
|
||||
/// The defining query, once the server describes a view by it;
|
||||
/// takes precedence over the structured fields.
|
||||
#[serde(default)]
|
||||
inputs: Vec<String>,
|
||||
query: Option<String>,
|
||||
#[serde(default)]
|
||||
incarnation: Option<String>,
|
||||
}
|
||||
@@ -2113,10 +2115,12 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
|
||||
let response = self.check_table_response(&request_id, response).await?;
|
||||
let response: DescribeMaterializedViewResponse =
|
||||
response.json().await.err_to_http(request_id)?;
|
||||
Ok(MaterializedViewInfo {
|
||||
definition: MaterializedViewDefinition {
|
||||
let definition = match response.query {
|
||||
Some(query) => MaterializedViewDefinition::from_sql(&query)?,
|
||||
None => MaterializedViewDefinition {
|
||||
source_table: response.source_table,
|
||||
source_namespace: response.source_namespace,
|
||||
lateral: None,
|
||||
projections: response
|
||||
.projections
|
||||
.into_iter()
|
||||
@@ -2127,8 +2131,10 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
|
||||
.collect(),
|
||||
filter: response.filter,
|
||||
limit: response.limit,
|
||||
inputs: response.inputs,
|
||||
},
|
||||
};
|
||||
Ok(MaterializedViewInfo {
|
||||
definition,
|
||||
incarnation: response.incarnation,
|
||||
})
|
||||
}
|
||||
@@ -12448,7 +12454,6 @@ 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"]);
|
||||
assert_eq!(view.incarnation(), Some("inc-1"));
|
||||
|
||||
let result = view
|
||||
|
||||
@@ -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"
|
||||
|
||||
Reference in New Issue
Block a user