laminar_sql/parser/
view_options.rs

1//! Options of `CREATE MATERIALIZED VIEW ... WITH (...)`.
2
3use std::time::Duration;
4
5use super::{tokenizer, window_rewriter::WindowRewriter, ParseError};
6
7/// How long an append-only materialized view keeps its rows.
8#[derive(Debug, Clone, PartialEq, Eq)]
9pub enum MaterializedViewRetention {
10    /// Rows within this interval of the view's event-time frontier, e.g. `'1 hour'`.
11    Interval(Duration),
12    /// The latest rows, e.g. `'10000 rows'`.
13    Rows(u64),
14    /// Every row, e.g. `'unbounded'`.
15    Unbounded,
16}
17
18/// Options declared in a view's trailing `WITH (...)`.
19#[derive(Debug, Clone, Default, PartialEq, Eq)]
20pub(super) struct ViewOptions {
21    pub(super) retention: Option<MaterializedViewRetention>,
22    pub(super) freshness: Option<Duration>,
23}
24
25/// Parse the trailing `WITH (...)` tokens of a view: `retention` and `freshness`.
26pub(super) fn parse_view_options(
27    tokens: Vec<sqlparser::tokenizer::TokenWithSpan>,
28) -> Result<ViewOptions, ParseError> {
29    let dialect = super::dialect::LaminarDialect::default();
30    let mut parser = sqlparser::parser::Parser::new(&dialect).with_tokens_with_locations(tokens);
31    let options = tokenizer::parse_with_options(&mut parser)?;
32    let mut parsed = ViewOptions::default();
33    for (key, value) in options {
34        if key.eq_ignore_ascii_case("retention") {
35            parsed.retention = Some(parse_retention(&value)?);
36        } else if key.eq_ignore_ascii_case("freshness") {
37            parsed.freshness = Some(parse_freshness(&value)?);
38        } else {
39            return Err(ParseError::StreamingError(format!(
40                "unknown CREATE MATERIALIZED VIEW option '{key}' (expected retention or freshness)"
41            )));
42        }
43    }
44    Ok(parsed)
45}
46
47fn parse_freshness(value: &str) -> Result<Duration, ParseError> {
48    let value = value.trim();
49    WindowRewriter::parse_interval_string(value)
50        .ok()
51        .filter(|interval| !interval.is_zero())
52        .ok_or_else(|| {
53            ParseError::StreamingError(format!(
54                "freshness '{value}' must be a nonzero interval such as '5 seconds'"
55            ))
56        })
57}
58
59fn parse_retention(value: &str) -> Result<MaterializedViewRetention, ParseError> {
60    let value = value.trim();
61    if value.eq_ignore_ascii_case("unbounded") {
62        return Ok(MaterializedViewRetention::Unbounded);
63    }
64    let invalid = || {
65        ParseError::StreamingError(format!(
66            "retention '{value}' must be an interval such as '1 hour', a row count such as \
67             '10000 rows', or 'unbounded'"
68        ))
69    };
70    let mut parts = value.split_whitespace();
71    if let (Some(count), Some(unit), None) = (parts.next(), parts.next(), parts.next()) {
72        if unit.eq_ignore_ascii_case("rows") || unit.eq_ignore_ascii_case("row") {
73            let rows: u64 = count.parse().map_err(|_| invalid())?;
74            return if rows == 0 {
75                Err(invalid())
76            } else {
77                Ok(MaterializedViewRetention::Rows(rows))
78            };
79        }
80    }
81    let interval = WindowRewriter::parse_interval_string(value).map_err(|_| invalid())?;
82    if interval.is_zero() {
83        return Err(invalid());
84    }
85    Ok(MaterializedViewRetention::Interval(interval))
86}
87
88#[cfg(test)]
89mod tests {
90    use super::*;
91
92    #[test]
93    fn retention_accepts_intervals_row_counts_and_unbounded() {
94        assert_eq!(
95            parse_retention("1 hour").unwrap(),
96            MaterializedViewRetention::Interval(Duration::from_secs(3600))
97        );
98        assert_eq!(
99            parse_retention("30 MINUTES").unwrap(),
100            MaterializedViewRetention::Interval(Duration::from_secs(1800))
101        );
102        assert_eq!(
103            parse_retention("10000 rows").unwrap(),
104            MaterializedViewRetention::Rows(10_000)
105        );
106        assert_eq!(
107            parse_retention("UNBOUNDED").unwrap(),
108            MaterializedViewRetention::Unbounded
109        );
110        assert_eq!(
111            parse_freshness("500 milliseconds").unwrap(),
112            Duration::from_millis(500)
113        );
114        for invalid in ["", "0 seconds", "10 rows"] {
115            assert!(parse_freshness(invalid).is_err(), "{invalid}");
116        }
117        for invalid in [
118            "",
119            "0 rows",
120            "0 seconds",
121            "ten rows",
122            "1 fortnight",
123            "5 rows extra",
124        ] {
125            assert!(parse_retention(invalid).is_err(), "{invalid}");
126        }
127    }
128}
laminar_sql/parser/view_options.rs

laminar_sql/parser/
view_options.rs
