micromegas/servers/
view_ddl.rs

1//! `CREATE [OR REPLACE] MATERIALIZED VIEW ... WITH (...)` / `DROP MATERIALIZED VIEW ...` --
2//! the DDL front end for DDL-defined eagerly materialized views.
3//!
4//! All three queries (`extract_query`, `count_src_query`, `merge_partitions_query`) are `WITH`
5//! options, so none of them is privileged the way a statement body would be -- that requires
6//! parsing the statement directly instead of going through `sqlparser`'s own `parse_create_view`,
7//! which unconditionally expects an `AS` body immediately after the view name and so cannot parse
8//! a `CREATE MATERIALIZED VIEW ... WITH (...)` with no `AS` body at all. [`parse_view_ddl`]
9//! therefore hand-rolls the parse off a single `Parser::try_with_sql`.
10
11use anyhow::{Context, Result};
12use datafusion::sql::sqlparser::{
13    ast::{Expr as SqlExpr, SqlOption, Value},
14    dialect::GenericDialect,
15    keywords::Keyword,
16    parser::Parser,
17    tokenizer::Token,
18};
19use micromegas_analytics::lakehouse::read_scope::CallerContext;
20use micromegas_analytics::lakehouse::view_definition::{
21    ViewDefinition, ViewOptions, check_view_set_name_charset, parse_time_delta,
22};
23use tonic::Status;
24
25/// One parsed `CREATE [OR REPLACE] MATERIALIZED VIEW` / `DROP MATERIALIZED VIEW` statement.
26#[derive(Debug, Clone)]
27pub enum ViewDdl {
28    Create {
29        name: String,
30        or_replace: bool,
31        // Boxed: `ViewDefinition` is far larger than `Drop`'s variant, and clippy's
32        // `large_enum_variant` flags the resulting size gap.
33        definition: Box<ViewDefinition>,
34        /// The verbatim statement text, stored as `lakehouse_view_set_definitions.definition_sql`.
35        sql: String,
36    },
37    Drop {
38        name: String,
39        if_exists: bool,
40    },
41}
42
43/// The same admin gate as the nine admin-gated lakehouse UDTFs/UDFs
44/// (`query.rs::register_lakehouse_functions`'s `if lakehouse_admin` block), pulled out as a
45/// standalone function since this path never builds a caller session context to register a UDTF
46/// against in the first place. Also load-bearing beyond the usual reason: a DDL view's queries
47/// are planned and materialized under `CallerContext::maintenance()` (`ReadScope::All`), so an
48/// author sees every audience regardless of their own read scope.
49pub fn authorize_view_ddl(caller: &CallerContext) -> Result<(), Status> {
50    if !caller.is_admin {
51        return Err(Status::permission_denied(
52            "CREATE/DROP MATERIALIZED VIEW requires admin privileges",
53        ));
54    }
55    Ok(())
56}
57
58/// `Ok(None)` when `sql` is not view DDL -- the overwhelmingly common case, decided by a cheap
59/// keyword peek before any full parse. `Err` when the hand-rolled parse does not reach EOF right
60/// after `<name> WITH (...)` (or, on `DROP`, right after the single name): an unmodelled clause,
61/// an `AS` body, a second `DROP` name, or a second statement each would otherwise silently do less
62/// than the statement says.
63///
64/// Returns a plain `anyhow::Result` rather than a typed error enum: this crate's convention
65/// (`rust/CLAUDE.md`) is to use `anyhow` unless a caller needs to branch on the error kind, and
66/// `execute_query`'s only caller here never does -- it always maps the error into a client-facing
67/// `invalid_argument` `Status`.
68pub fn parse_view_ddl(sql: &str) -> Result<Option<ViewDdl>> {
69    let first_word = sql
70        .trim_start()
71        .split(|c: char| c.is_whitespace() || c == '(')
72        .next()
73        .unwrap_or("")
74        .to_lowercase();
75    // A leading `--` or `/*` comment (routine from BI/notebook front ends) makes the naive
76    // first-word peek useless -- fall through to the real tokenizing parse below, which skips
77    // comments, instead of bailing out here.
78    if first_word != "create"
79        && first_word != "drop"
80        && !first_word.starts_with("--")
81        && !first_word.starts_with("/*")
82    {
83        return Ok(None);
84    }
85
86    let mut parser = Parser::new(&GenericDialect {})
87        .try_with_sql(sql)
88        .with_context(|| "tokenizing statement")?;
89
90    if parser.parse_keyword(Keyword::CREATE) {
91        let or_replace = parser.parse_keywords(&[Keyword::OR, Keyword::REPLACE]);
92        if !parser.parse_keyword(Keyword::MATERIALIZED) {
93            // CREATE VIEW / CREATE TABLE / ... -- not our DDL, let ctx.sql handle it (or reject
94            // it) as usual.
95            return Ok(None);
96        }
97        parser
98            .expect_keyword_is(Keyword::VIEW)
99            .with_context(|| "expected VIEW after CREATE [OR REPLACE] MATERIALIZED")?;
100        let name = parser
101            .parse_object_name(false)
102            .with_context(|| "expected a view set name")?
103            .to_string();
104        let options = parser
105            .parse_options(Keyword::WITH)
106            .with_context(|| "expected WITH (...) options")?;
107        require_eof(
108            &mut parser,
109            "CREATE [OR REPLACE] MATERIALIZED VIEW <name> WITH (...)",
110        )?;
111        let definition = view_definition_from_options(name.clone(), options)
112            .with_context(|| format!("CREATE MATERIALIZED VIEW {name}"))?;
113        return Ok(Some(ViewDdl::Create {
114            name,
115            or_replace,
116            definition: Box::new(definition),
117            sql: sql.to_string(),
118        }));
119    }
120
121    if parser.parse_keyword(Keyword::DROP) {
122        if !parser.parse_keyword(Keyword::MATERIALIZED) {
123            return Ok(None);
124        }
125        parser
126            .expect_keyword_is(Keyword::VIEW)
127            .with_context(|| "expected VIEW after DROP MATERIALIZED")?;
128        let if_exists = parser.parse_keywords(&[Keyword::IF, Keyword::EXISTS]);
129        let name = parser
130            .parse_object_name(false)
131            .with_context(|| "expected a view set name")?
132            .to_string();
133        require_eof(&mut parser, "DROP MATERIALIZED VIEW [IF EXISTS] <name>")?;
134        return Ok(Some(ViewDdl::Drop { name, if_exists }));
135    }
136
137    Ok(None)
138}
139
140/// Consumes an optional trailing `;` and requires EOF immediately after -- preserving the
141/// single-statement rule `SessionContext::sql`'s `sql_to_statement` enforces for every non-DDL
142/// query; intercepting DDL ahead of `ctx.sql` would otherwise bypass that guard and silently
143/// execute only the first of several statements.
144fn require_eof(parser: &mut Parser, context: &str) -> Result<()> {
145    let _ = parser.consume_token(&Token::SemiColon);
146    let trailing = parser.peek_token_ref().token.clone();
147    if trailing != Token::EOF {
148        anyhow::bail!("unexpected trailing input after {context}: {trailing}");
149    }
150    Ok(())
151}
152
153fn value_to_string(expr: &SqlExpr) -> Result<String> {
154    match expr {
155        SqlExpr::Value(v) => match &v.value {
156            Value::SingleQuotedString(s) => Ok(s.clone()),
157            Value::DollarQuotedString(d) => Ok(d.value.clone()),
158            other => anyhow::bail!("expected a string literal, got {other}"),
159        },
160        other => anyhow::bail!("expected a string literal, got {other}"),
161    }
162}
163
164fn value_to_i32(expr: &SqlExpr) -> Result<i32> {
165    match expr {
166        SqlExpr::Value(v) => match &v.value {
167            Value::Number(s, _) => s
168                .parse::<i32>()
169                .with_context(|| format!("invalid integer {s:?}")),
170            Value::SingleQuotedString(s) => s
171                .parse::<i32>()
172                .with_context(|| format!("invalid integer {s:?}")),
173            other => anyhow::bail!("expected an integer literal, got {other}"),
174        },
175        other => anyhow::bail!("expected an integer literal, got {other}"),
176    }
177}
178
179/// Assembles the normalized `ViewDefinition` from a `WITH (...)` option list. Option semantics:
180/// `extract_query`, `count_src_query`, `merge_partitions_query`, and `update_group` are required;
181/// `time_column` is required unless
182/// both `min_time_column` and `max_time_column` are given individually; `source_partition_delta`
183/// defaults to `'1 day'`; `merge_partition_delta` defaults to `source_partition_delta`;
184/// `merge_sort_order` is an optional comma-separated column list. Anything else is a hard error.
185fn view_definition_from_options(name: String, options: Vec<SqlOption>) -> Result<ViewDefinition> {
186    let mut extract_query = None;
187    let mut count_src_query = None;
188    let mut merge_partitions_query = None;
189    let mut update_group = None;
190    let mut time_column = None;
191    let mut min_time_column = None;
192    let mut max_time_column = None;
193    let mut source_partition_delta: Option<String> = None;
194    let mut merge_partition_delta: Option<String> = None;
195    let mut merge_sort_order: Option<Vec<String>> = None;
196
197    for opt in options {
198        let SqlOption::KeyValue { key, value } = opt else {
199            anyhow::bail!("unsupported view option (expected `key = value`)");
200        };
201        let key_name = key.value.to_lowercase();
202        match key_name.as_str() {
203            "extract_query" => extract_query = Some(value_to_string(&value)?),
204            "count_src_query" => count_src_query = Some(value_to_string(&value)?),
205            "merge_partitions_query" => merge_partitions_query = Some(value_to_string(&value)?),
206            "update_group" => update_group = Some(value_to_i32(&value)?),
207            "time_column" => time_column = Some(value_to_string(&value)?),
208            "min_time_column" => min_time_column = Some(value_to_string(&value)?),
209            "max_time_column" => max_time_column = Some(value_to_string(&value)?),
210            "source_partition_delta" => {
211                let s = value_to_string(&value)?;
212                parse_time_delta(&s).with_context(|| "source_partition_delta")?;
213                source_partition_delta = Some(s);
214            }
215            "merge_partition_delta" => {
216                let s = value_to_string(&value)?;
217                parse_time_delta(&s).with_context(|| "merge_partition_delta")?;
218                merge_partition_delta = Some(s);
219            }
220            "merge_sort_order" => {
221                let s = value_to_string(&value)?;
222                let cols: Vec<String> = s
223                    .split(',')
224                    .map(|c| c.trim().to_string())
225                    .filter(|c| !c.is_empty())
226                    .collect();
227                if cols.is_empty() {
228                    anyhow::bail!(
229                        "merge_sort_order must be a non-empty, comma-separated column list"
230                    );
231                }
232                merge_sort_order = Some(cols);
233            }
234            other => anyhow::bail!("unknown view option '{other}'"),
235        }
236    }
237
238    let extract_query = extract_query.context("missing required option 'extract_query'")?;
239    let count_src_query = count_src_query.context("missing required option 'count_src_query'")?;
240    let merge_partitions_query =
241        merge_partitions_query.context("missing required option 'merge_partitions_query'")?;
242    let update_group = update_group.context("missing required option 'update_group'")?;
243
244    let (resolved_min, resolved_max) = match (time_column, min_time_column, max_time_column) {
245        (Some(t), min, max) => (min.unwrap_or_else(|| t.clone()), max.unwrap_or(t)),
246        (None, Some(min), Some(max)) => (min, max),
247        _ => anyhow::bail!(
248            "missing required option 'time_column' (or both 'min_time_column' and \
249             'max_time_column')"
250        ),
251    };
252
253    let source_partition_delta = source_partition_delta.unwrap_or_else(|| "1 day".to_string());
254    let merge_partition_delta =
255        merge_partition_delta.unwrap_or_else(|| source_partition_delta.clone());
256
257    check_view_set_name_charset(&name)?;
258
259    Ok(ViewDefinition {
260        view_set_name: name,
261        extract_query,
262        count_src_query,
263        merge_partitions_query,
264        update_group,
265        options: ViewOptions {
266            min_time_column: resolved_min,
267            max_time_column: resolved_max,
268            source_partition_delta,
269            merge_partition_delta,
270            merge_sort_order,
271        },
272    })
273}
micromegas/servers/view_ddl.rs

micromegas/servers/
view_ddl.rs
