reddb_rql/parser/
timeseries.rs

1//! Parser for CREATE/DROP TIMESERIES
2
3use super::error::ParseError;
4use super::Parser;
5use crate::ast::{
6    CreateSloQuery, CreateTableQuery, CreateTimeSeriesQuery, DropTimeSeriesQuery, HypertableDdl,
7    QueryExpr,
8};
9use crate::lexer::Token;
10use reddb_types::catalog::CollectionModel;
11
12impl<'a> Parser<'a> {
13    /// Parse CREATE TIMESERIES body (after CREATE TIMESERIES consumed)
14    pub fn parse_create_timeseries_body(&mut self) -> Result<QueryExpr, ParseError> {
15        let if_not_exists = self.match_if_not_exists()?;
16        let name = self.expect_ident()?;
17
18        let mut retention_ms = None;
19        let mut chunk_size = None;
20        let mut downsample_policies = Vec::new();
21        let mut session_key: Option<String> = None;
22        let mut session_gap_ms: Option<u64> = None;
23        let mut columnar = true;
24
25        // Parse optional clauses in any order
26        loop {
27            if self.consume(&Token::Retention)? {
28                let value = self.parse_float()?;
29                let unit = self.parse_duration_unit()?;
30                retention_ms = Some((value * unit) as u64);
31            } else if self.consume_ident_ci("CHUNK_SIZE")? || self.consume_ident_ci("CHUNKSIZE")? {
32                chunk_size = Some(self.parse_integer()? as usize);
33            } else if self.consume_ident_ci("NO")? {
34                if !self.consume_ident_ci("COLUMNAR")? {
35                    return Err(ParseError::expected(
36                        vec!["COLUMNAR"],
37                        self.peek(),
38                        self.position(),
39                    ));
40                }
41                columnar = false;
42            } else if self.consume_ident_ci("COLUMNAR")? {
43                return Err(retired_columnar_keyword_error(self.position()));
44            } else if self.consume_ident_ci("DOWNSAMPLE")? {
45                downsample_policies.push(self.parse_downsample_policy_spec()?);
46                while self.consume(&Token::Comma)? {
47                    downsample_policies.push(self.parse_downsample_policy_spec()?);
48                }
49            } else if self.consume(&Token::With)? {
50                // `WITH SESSION_KEY <col> SESSION_GAP <duration>` — both
51                // clauses are paired so the SESSIONIZE operator (slice
52                // 2+) has a complete default. Order is fixed
53                // (SESSION_KEY first) to keep the grammar simple; one
54                // without the other is a parse error.
55                self.parse_with_session_clause(&mut session_key, &mut session_gap_ms)?;
56            } else {
57                break;
58            }
59        }
60
61        Ok(QueryExpr::CreateTimeSeries(CreateTimeSeriesQuery {
62            name,
63            retention_ms,
64            chunk_size,
65            downsample_policies,
66            if_not_exists,
67            hypertable: None,
68            session_key,
69            session_gap_ms,
70            columnar,
71        }))
72    }
73
74    /// Parse `SESSION_KEY <ident> SESSION_GAP <duration>` after a
75    /// `WITH` token has been consumed. Both clauses are required; a
76    /// SESSION_KEY without a SESSION_GAP (or vice-versa) is rejected
77    /// at parse time so the descriptor never carries half a pairing.
78    fn parse_with_session_clause(
79        &mut self,
80        session_key: &mut Option<String>,
81        session_gap_ms: &mut Option<u64>,
82    ) -> Result<(), ParseError> {
83        if !self.consume_ident_ci("SESSION_KEY")? {
84            return Err(ParseError::new(
85                "expected SESSION_KEY after WITH on CREATE TIMESERIES".to_string(),
86                self.position(),
87            ));
88        }
89        let key = self.expect_ident()?;
90        if !self.consume_ident_ci("SESSION_GAP")? {
91            return Err(ParseError::new(
92                "WITH SESSION_KEY requires a paired SESSION_GAP <duration>".to_string(),
93                self.position(),
94            ));
95        }
96        let value = self.parse_float()?;
97        let unit = self.parse_duration_unit()?;
98        *session_key = Some(key);
99        *session_gap_ms = Some((value * unit) as u64);
100        Ok(())
101    }
102
103    /// Parse CREATE METRICS body (after CREATE METRICS consumed).
104    ///
105    /// v0 intentionally establishes only the collection contract. Ingestion,
106    /// series registry, and Prometheus adapter slices build on this metadata.
107    pub fn parse_create_metrics_body(&mut self) -> Result<QueryExpr, ParseError> {
108        let if_not_exists = self.match_if_not_exists()?;
109        let name = self.expect_ident()?;
110
111        let mut raw_retention_ms = None;
112        let mut tenant_by = None;
113        let mut downsample_policies = Vec::new();
114
115        loop {
116            if self.consume(&Token::Retention)? {
117                let value = self.parse_float()?;
118                let unit = self.parse_duration_unit()?;
119                raw_retention_ms = Some((value * unit) as u64);
120            } else if self.consume_ident_ci("DOWNSAMPLE")? {
121                downsample_policies.push(self.parse_downsample_policy_spec()?);
122                while self.consume(&Token::Comma)? {
123                    downsample_policies.push(self.parse_downsample_policy_spec()?);
124                }
125            } else if tenant_by.is_none() && self.consume_ident_ci("TENANT")? {
126                self.expect(Token::By)?;
127                self.expect(Token::LParen)?;
128                let mut path = self.expect_ident_or_keyword()?;
129                while self.consume(&Token::Dot)? {
130                    let next = self.expect_ident_or_keyword()?;
131                    path = format!("{path}.{next}");
132                }
133                self.expect(Token::RParen)?;
134                tenant_by = Some(path);
135            } else {
136                break;
137            }
138        }
139
140        Ok(QueryExpr::CreateTable(CreateTableQuery {
141            collection_model: CollectionModel::Metrics,
142            name,
143            columns: Vec::new(),
144            if_not_exists,
145            default_ttl_ms: raw_retention_ms,
146            metrics_rollup_policies: downsample_policies,
147            context_index_fields: Vec::new(),
148            context_index_enabled: false,
149            timestamps: false,
150            partition_by: None,
151            tenant_by,
152            append_only: true,
153            subscriptions: Vec::new(),
154            analytics_config: Vec::new(),
155            vault_own_master_key: false,
156            ai_policy: None,
157        }))
158    }
159
160    /// Parse CREATE METRIC body (after CREATE METRIC consumed).
161    pub fn parse_create_metric_body(&mut self) -> Result<QueryExpr, ParseError> {
162        let mut path = self.expect_ident_or_keyword()?.to_ascii_lowercase();
163        while self.consume(&Token::Dot)? {
164            let next = self.expect_ident_or_keyword()?.to_ascii_lowercase();
165            path = format!("{path}.{next}");
166        }
167
168        let mut kind = None;
169        let mut role = None;
170        let mut source: Option<String> = None;
171        let mut query: Option<String> = None;
172        let mut window_ms: Option<u64> = None;
173        let mut time_field: Option<String> = None;
174        loop {
175            if self.consume_ident_ci("TYPE")? || self.consume_ident_ci("KIND")? {
176                kind = Some(self.expect_ident_or_keyword()?.to_ascii_lowercase());
177            } else if self.consume_ident_ci("ROLE")? {
178                role = Some(self.expect_ident_or_keyword()?.to_ascii_lowercase());
179            } else if self.consume_ident_ci("SOURCE")? {
180                source = Some(self.expect_ident_or_keyword()?);
181            } else if self.consume_ident_ci("QUERY")? {
182                let value = self.parse_literal_value()?;
183                match value {
184                    reddb_types::types::Value::Text(s) => query = Some(s.to_string()),
185                    other => {
186                        return Err(ParseError::new(
187                            format!("derived metric QUERY expects a string literal, got {other:?}"),
188                            self.position(),
189                        ));
190                    }
191                }
192            } else if self.consume_ident_ci("WINDOW")? {
193                let value = self.parse_float()?;
194                let unit = self.parse_duration_unit()?;
195                window_ms = Some((value * unit) as u64);
196            } else if self.consume_ident_ci("TIME_FIELD")? {
197                time_field = Some(self.expect_ident_or_keyword()?);
198            } else {
199                break;
200            }
201        }
202
203        Ok(QueryExpr::CreateMetric(crate::ast::CreateMetricQuery {
204            path,
205            kind: kind.ok_or_else(|| {
206                ParseError::new(
207                    "metric descriptor requires TYPE or KIND".to_string(),
208                    self.position(),
209                )
210            })?,
211            role: role.ok_or_else(|| {
212                ParseError::new(
213                    "metric descriptor requires ROLE".to_string(),
214                    self.position(),
215                )
216            })?,
217            source,
218            query,
219            window_ms,
220            time_field,
221        }))
222    }
223
224    /// Parse ALTER METRIC body (after ALTER METRIC consumed).
225    ///
226    /// Grammar:
227    ///   ALTER METRIC <dotted.path> SET ROLE <ident>
228    ///   ALTER METRIC <dotted.path> SET KIND <ident>      -- rejected at runtime
229    ///   ALTER METRIC <dotted.path> SET TYPE <ident>      -- rejected at runtime
230    ///   ALTER METRIC <dotted.path> SET PATH <dotted>     -- rejected at runtime
231    ///
232    /// Immutable-field attempts parse so the runtime can return a
233    /// structured "field X cannot be changed" error explaining *why*.
234    pub fn parse_alter_metric_body(&mut self) -> Result<QueryExpr, ParseError> {
235        let mut path = self.expect_ident_or_keyword()?.to_ascii_lowercase();
236        while self.consume(&Token::Dot)? {
237            let next = self.expect_ident_or_keyword()?.to_ascii_lowercase();
238            path = format!("{path}.{next}");
239        }
240
241        if !self.consume(&Token::Set)? && !self.consume_ident_ci("SET")? {
242            return Err(ParseError::expected(
243                vec!["SET"],
244                self.peek(),
245                self.position(),
246            ));
247        }
248
249        let mut set_role = None;
250        let mut attempted_kind = None;
251        let mut attempted_path = None;
252
253        if self.consume_ident_ci("ROLE")? {
254            set_role = Some(self.expect_ident_or_keyword()?.to_ascii_lowercase());
255        } else if self.consume_ident_ci("KIND")? || self.consume_ident_ci("TYPE")? {
256            attempted_kind = Some(self.expect_ident_or_keyword()?.to_ascii_lowercase());
257        } else if self.consume(&Token::Path)? || self.consume_ident_ci("PATH")? {
258            let mut new_path = self.expect_ident_or_keyword()?.to_ascii_lowercase();
259            while self.consume(&Token::Dot)? {
260                let next = self.expect_ident_or_keyword()?.to_ascii_lowercase();
261                new_path = format!("{new_path}.{next}");
262            }
263            attempted_path = Some(new_path);
264        } else {
265            return Err(ParseError::expected(
266                vec!["ROLE", "KIND", "TYPE", "PATH"],
267                self.peek(),
268                self.position(),
269            ));
270        }
271
272        Ok(QueryExpr::AlterMetric(crate::ast::AlterMetricQuery {
273            path,
274            set_role,
275            attempted_kind,
276            attempted_path,
277        }))
278    }
279
280    /// Parse CREATE SLO body (after CREATE SLO consumed).
281    ///
282    /// Grammar:
283    ///   CREATE SLO <dotted.path>
284    ///     ON <metric.dotted.path>
285    ///     TARGET <number>
286    ///     WINDOW <number> <duration_unit>
287    ///
288    /// Clauses are positional after the SLO path so the grammar stays
289    /// tight; the parser leaves semantic validation (metric exists, role
290    /// = sli, target in range) to the runtime catalog where the error
291    /// wording can reference the live catalog state.
292    pub fn parse_create_slo_body(&mut self) -> Result<QueryExpr, ParseError> {
293        let mut path = self.expect_ident_or_keyword()?.to_ascii_lowercase();
294        while self.consume(&Token::Dot)? {
295            let next = self.expect_ident_or_keyword()?.to_ascii_lowercase();
296            path = format!("{path}.{next}");
297        }
298
299        if !self.consume(&Token::On)? {
300            return Err(ParseError::expected(
301                vec!["ON"],
302                self.peek(),
303                self.position(),
304            ));
305        }
306
307        let mut metric_path = self.expect_ident_or_keyword()?.to_ascii_lowercase();
308        while self.consume(&Token::Dot)? {
309            let next = self.expect_ident_or_keyword()?.to_ascii_lowercase();
310            metric_path = format!("{metric_path}.{next}");
311        }
312
313        let mut target: Option<f64> = None;
314        let mut window_ms: Option<u64> = None;
315
316        loop {
317            if self.consume_ident_ci("TARGET")? {
318                target = Some(self.parse_float()?);
319            } else if self.consume_ident_ci("WINDOW")? {
320                let value = self.parse_float()?;
321                let unit = self.parse_duration_unit()?;
322                window_ms = Some((value * unit) as u64);
323            } else {
324                break;
325            }
326        }
327
328        Ok(QueryExpr::CreateSlo(CreateSloQuery {
329            path,
330            metric_path,
331            target: target.ok_or_else(|| {
332                ParseError::new(
333                    "SLO descriptor requires TARGET <number>".to_string(),
334                    self.position(),
335                )
336            })?,
337            window_ms: window_ms.ok_or_else(|| {
338                ParseError::new(
339                    "SLO descriptor requires WINDOW <duration>".to_string(),
340                    self.position(),
341                )
342            })?,
343        }))
344    }
345
346    /// Parse CREATE HYPERTABLE body — TimescaleDB-style.
347    ///
348    ///   CREATE HYPERTABLE metrics
349    ///     TIME_COLUMN ts
350    ///     CHUNK_INTERVAL '1d'
351    ///     [TTL '90d']
352    ///     [RETENTION 90 DAYS]          -- collection-level TTL (ms)
353    ///
354    /// Produces the same `CreateTimeSeriesQuery` AST as `CREATE
355    /// TIMESERIES`, with the `hypertable` field populated. The
356    /// runtime dispatcher registers the spec on the RedDB-wide
357    /// `HypertableRegistry` alongside creating the collection.
358    pub fn parse_create_hypertable_body(&mut self) -> Result<QueryExpr, ParseError> {
359        let if_not_exists = self.match_if_not_exists()?;
360        let name = self.expect_ident()?;
361
362        let mut time_column: Option<String> = None;
363        let mut chunk_interval_ns: Option<u64> = None;
364        let mut ttl_ns: Option<u64> = None;
365        let mut retention_ms = None;
366        let mut columnar = true;
367
368        loop {
369            if self.consume_ident_ci("TIME_COLUMN")? {
370                time_column = Some(self.expect_ident()?);
371            } else if self.consume_ident_ci("CHUNK_INTERVAL")? {
372                chunk_interval_ns = Some(self.parse_duration_ns_literal("CHUNK_INTERVAL")?);
373            } else if self.consume_ident_ci("NO")? {
374                if !self.consume_ident_ci("COLUMNAR")? {
375                    return Err(ParseError::expected(
376                        vec!["COLUMNAR"],
377                        self.peek(),
378                        self.position(),
379                    ));
380                }
381                columnar = false;
382            } else if self.consume_ident_ci("COLUMNAR")? {
383                return Err(retired_columnar_keyword_error(self.position()));
384            } else if self.consume_ident_ci("TTL")? {
385                ttl_ns = Some(self.parse_duration_ns_literal("TTL")?);
386            } else if self.consume(&Token::Retention)? {
387                let value = self.parse_float()?;
388                let unit = self.parse_duration_unit()?;
389                retention_ms = Some((value * unit) as u64);
390            } else {
391                break;
392            }
393        }
394
395        let time_column = time_column.ok_or_else(|| {
396            ParseError::new(
397                "CREATE HYPERTABLE requires TIME_COLUMN <ident>".to_string(),
398                self.position(),
399            )
400        })?;
401        let chunk_interval_ns = chunk_interval_ns.ok_or_else(|| {
402            ParseError::new(
403                "CREATE HYPERTABLE requires CHUNK_INTERVAL '<duration>' (e.g. '1d')".to_string(),
404                self.position(),
405            )
406        })?;
407
408        Ok(QueryExpr::CreateTimeSeries(CreateTimeSeriesQuery {
409            name,
410            retention_ms,
411            chunk_size: None,
412            downsample_policies: Vec::new(),
413            if_not_exists,
414            hypertable: Some(HypertableDdl {
415                time_column,
416                chunk_interval_ns,
417                default_ttl_ns: ttl_ns,
418            }),
419            session_key: None,
420            session_gap_ms: None,
421            columnar,
422        }))
423    }
424
425    /// Accept a string-literal duration (`'1d'`, `'5m'`, `'30s'`, …) and
426    /// resolve it to nanoseconds using the shared retention grammar.
427    fn parse_duration_ns_literal(&mut self, clause: &str) -> Result<u64, ParseError> {
428        let pos = self.position();
429        let value = self.parse_literal_value()?;
430        match value {
431            reddb_types::types::Value::Text(s) => {
432                reddb_types::duration::parse_duration_ns(&s).ok_or_else(|| {
433                    ParseError::new(
434                        // F-05: `s` is caller-controlled string-literal bytes.
435                        // Render via `{:?}` so CR/LF/NUL/quotes are escaped
436                        // before reaching downstream serialization sinks.
437                        // `clause` is a static internal label and stays bare.
438                        format!("{clause} duration {s:?} is not a valid duration literal"),
439                        pos,
440                    )
441                })
442            }
443            other => Err(ParseError::new(
444                format!("{clause} expects a string duration literal, got {other:?}"),
445                pos,
446            )),
447        }
448    }
449
450    /// Parse DROP TIMESERIES body (after DROP TIMESERIES consumed)
451    pub fn parse_drop_timeseries_body(&mut self) -> Result<QueryExpr, ParseError> {
452        let if_exists = self.match_if_exists()?;
453        let name = self.parse_drop_collection_name()?;
454        Ok(QueryExpr::DropTimeSeries(DropTimeSeriesQuery {
455            name,
456            if_exists,
457        }))
458    }
459
460    /// Parse a duration unit and return the multiplier in milliseconds
461    pub fn parse_duration_unit(&mut self) -> Result<f64, ParseError> {
462        // Aggregate-function keywords (`MIN`, `MAX`, `AVG`) lex as
463        // dedicated tokens, not `Token::Ident`, so they need their
464        // own arms. `MIN` is the minute alias; `MAX` and `AVG` have
465        // no canonical duration meaning today but were silently
466        // falling through to the seconds default — surface a clear
467        // error instead.
468        match self.peek().clone() {
469            Token::Ident(ref unit) => {
470                let mult = match unit.to_ascii_lowercase().as_str() {
471                    "ms" | "msec" | "millisecond" | "milliseconds" => 1.0,
472                    "s" | "sec" | "secs" | "second" | "seconds" => 1_000.0,
473                    "m" | "min" | "mins" | "minute" | "minutes" => 60_000.0,
474                    "h" | "hr" | "hrs" | "hour" | "hours" => 3_600_000.0,
475                    "d" | "day" | "days" => 86_400_000.0,
476                    other => {
477                        return Err(ParseError::new(
478                            // F-05: `other` is caller-controlled identifier
479                            // text. Render via `{:?}` so embedded CR/LF/NUL/
480                            // quotes are escaped before the message reaches
481                            // downstream serialization sinks.
482                            format!("unknown duration unit {other:?}, expected s/m/h/d"),
483                            self.position(),
484                        ));
485                    }
486                };
487                self.advance()?;
488                Ok(mult)
489            }
490            Token::Min => {
491                // `MIN` keyword used as the minute alias.
492                self.advance()?;
493                Ok(60_000.0)
494            }
495            Token::Max | Token::Avg => {
496                // These keywords have no duration semantics; reject
497                // explicitly so a stray aggregate keyword does not
498                // silently default to seconds.
499                let kw = self.peek().clone();
500                Err(ParseError::new(
501                    format!("unknown duration unit '{}', expected s/m/h/d", kw),
502                    self.position(),
503                ))
504            }
505            _ => Ok(1_000.0), // default: seconds
506        }
507    }
508
509    fn parse_downsample_policy_spec(&mut self) -> Result<String, ParseError> {
510        let target = self.parse_resolution_spec()?;
511        self.expect(Token::Colon)?;
512        let source = self.parse_resolution_spec()?;
513        let aggregation = if self.consume(&Token::Colon)? {
514            self.expect_ident_or_keyword()?.to_ascii_lowercase()
515        } else {
516            "avg".to_string()
517        };
518        Ok(format!("{target}:{source}:{aggregation}"))
519    }
520
521    fn parse_resolution_spec(&mut self) -> Result<String, ParseError> {
522        match self.peek().clone() {
523            Token::Ident(value) if value.eq_ignore_ascii_case("raw") => {
524                self.advance()?;
525                Ok(value.to_ascii_lowercase())
526            }
527            Token::Integer(value) => {
528                self.advance()?;
529                let unit = self.expect_ident_or_keyword()?.to_ascii_lowercase();
530                Ok(format!("{value}{unit}"))
531            }
532            Token::Float(value) => {
533                self.advance()?;
534                let unit = self.expect_ident_or_keyword()?.to_ascii_lowercase();
535                let number = if value.fract().abs() < f64::EPSILON {
536                    format!("{}", value as i64)
537                } else {
538                    value.to_string()
539                };
540                Ok(format!("{number}{unit}"))
541            }
542            other => Err(ParseError::new(
543                format!(
544                    "expected duration literal for downsample policy, got {}",
545                    other
546                ),
547                self.position(),
548            )),
549        }
550    }
551}
552
553fn retired_columnar_keyword_error(position: crate::lexer::Position) -> ParseError {
554    ParseError::new(
555        "COLUMNAR is no longer accepted; columnar projection is automatic for in-scope \
556         collections, use NO COLUMNAR to opt out"
557            .to_string(),
558        position,
559    )
560}
561
562#[cfg(test)]
563mod tests {
564    use super::*;
565    use reddb_types::catalog::CollectionModel;
566
567    fn parse_query(input: &str) -> Result<QueryExpr, ParseError> {
568        crate::parser::parse(input).map(|query| query.query)
569    }
570
571    #[test]
572    fn create_timeseries_accepts_clause_order_and_defaults_to_columnar() {
573        let query = parse_query(
574            "CREATE TIMESERIES IF NOT EXISTS readings DOWNSAMPLE 1h:raw \
575             RETENTION 2 h CHUNKSIZE 64",
576        )
577        .unwrap();
578
579        let QueryExpr::CreateTimeSeries(timeseries) = query else {
580            panic!("expected create timeseries");
581        };
582        assert_eq!(timeseries.name, "readings");
583        assert!(timeseries.if_not_exists);
584        assert!(timeseries.columnar);
585        assert_eq!(timeseries.retention_ms, Some(2 * 3_600_000));
586        assert_eq!(timeseries.chunk_size, Some(64));
587        assert_eq!(timeseries.downsample_policies, vec!["1h:raw:avg"]);
588        assert_eq!(timeseries.session_key, None);
589        assert_eq!(timeseries.session_gap_ms, None);
590        assert!(timeseries.hypertable.is_none());
591    }
592
593    #[test]
594    fn create_timeseries_accepts_no_columnar_opt_out() {
595        let query = parse_query("CREATE TIMESERIES readings NO COLUMNAR").unwrap();
596        let QueryExpr::CreateTimeSeries(timeseries) = query else {
597            panic!("expected create timeseries");
598        };
599        assert!(!timeseries.columnar);
600    }
601
602    #[test]
603    fn create_timeseries_rejects_retired_columnar_keyword() {
604        for sql in [
605            "CREATE TIMESERIES readings COLUMNAR",
606            "CREATE HYPERTABLE readings TIME_COLUMN ts CHUNK_INTERVAL '1h' COLUMNAR",
607        ] {
608            let err = parse_query(sql)
609                .expect_err("COLUMNAR is retired")
610                .to_string();
611            assert!(
612                err.contains("COLUMNAR is no longer accepted"),
613                "{sql}: {err}"
614            );
615            assert!(err.contains("automatic"), "{sql}: {err}");
616        }
617    }
618
619    #[test]
620    fn create_metrics_sets_collection_defaults_and_optional_clauses() {
621        let query = parse_query(
622            "CREATE METRICS IF NOT EXISTS telemetry RETENTION 30 m \
623             DOWNSAMPLE 5m:raw:max TENANT BY (ctx.tenant)",
624        )
625        .unwrap();
626
627        let QueryExpr::CreateTable(metrics) = query else {
628            panic!("expected metrics collection");
629        };
630        assert_eq!(metrics.collection_model, CollectionModel::Metrics);
631        assert_eq!(metrics.name, "telemetry");
632        assert!(metrics.if_not_exists);
633        assert_eq!(metrics.default_ttl_ms, Some(30 * 60_000));
634        assert_eq!(metrics.metrics_rollup_policies, vec!["5m:raw:max"]);
635        assert_eq!(metrics.tenant_by.as_deref(), Some("ctx.tenant"));
636        assert!(metrics.append_only);
637        assert!(metrics.columns.is_empty());
638    }
639
640    #[test]
641    fn create_metric_alter_metric_and_slo_parse_descriptor_forms() {
642        let query = parse_query(
643            "CREATE METRIC Svc.Latency.P99 TYPE gauge ROLE sli SOURCE rollups \
644             QUERY 'SELECT p99 FROM rollups' WINDOW 5 min TIME_FIELD observed_at",
645        )
646        .unwrap();
647        let QueryExpr::CreateMetric(metric) = query else {
648            panic!("expected create metric");
649        };
650        assert_eq!(metric.path, "svc.latency.p99");
651        assert_eq!(metric.kind, "gauge");
652        assert_eq!(metric.role, "sli");
653        assert_eq!(metric.source.as_deref(), Some("rollups"));
654        assert_eq!(metric.query.as_deref(), Some("SELECT p99 FROM rollups"));
655        assert_eq!(metric.window_ms, Some(5 * 60_000));
656        assert_eq!(metric.time_field.as_deref(), Some("observed_at"));
657
658        let query = parse_query("ALTER METRIC Svc.Latency.P99 SET PATH svc.latency.p95").unwrap();
659        let QueryExpr::AlterMetric(alter) = query else {
660            panic!("expected alter metric");
661        };
662        assert_eq!(alter.path, "svc.latency.p99");
663        assert_eq!(alter.set_role, None);
664        assert_eq!(alter.attempted_kind, None);
665        assert_eq!(alter.attempted_path.as_deref(), Some("svc.latency.p95"));
666
667        let query =
668            parse_query("CREATE SLO Api.Availability ON Svc.Latency.P99 TARGET 0.999 WINDOW 28 d")
669                .unwrap();
670        let QueryExpr::CreateSlo(slo) = query else {
671            panic!("expected create slo");
672        };
673        assert_eq!(slo.path, "api.availability");
674        assert_eq!(slo.metric_path, "svc.latency.p99");
675        assert!((slo.target - 0.999).abs() < f64::EPSILON);
676        assert_eq!(slo.window_ms, 28 * 86_400_000);
677    }
678
679    #[test]
680    fn create_hypertable_and_drop_timeseries_parse_variants() {
681        let query = parse_query(
682            "CREATE HYPERTABLE IF NOT EXISTS events TIME_COLUMN ts \
683             CHUNK_INTERVAL '30m' TTL '10s' RETENTION 1 h",
684        )
685        .unwrap();
686        let QueryExpr::CreateTimeSeries(timeseries) = query else {
687            panic!("expected hypertable as timeseries");
688        };
689        let hypertable = timeseries.hypertable.expect("hypertable ddl");
690        assert_eq!(timeseries.name, "events");
691        assert!(timeseries.if_not_exists);
692        assert!(timeseries.columnar);
693        assert_eq!(timeseries.retention_ms, Some(3_600_000));
694        assert_eq!(hypertable.time_column, "ts");
695        assert_eq!(hypertable.chunk_interval_ns, 30 * 60 * 1_000_000_000);
696        assert_eq!(hypertable.default_ttl_ns, Some(10 * 1_000_000_000));
697
698        let query = parse_query(
699            "CREATE HYPERTABLE cold_events TIME_COLUMN ts CHUNK_INTERVAL '1h' NO COLUMNAR",
700        )
701        .unwrap();
702        let QueryExpr::CreateTimeSeries(timeseries) = query else {
703            panic!("expected hypertable as timeseries");
704        };
705        assert!(!timeseries.columnar);
706
707        let query = parse_query("DROP TIMESERIES IF EXISTS tenant.metrics.*").unwrap();
708        assert!(matches!(
709            query,
710            QueryExpr::DropTimeSeries(drop) if drop.name == "tenant.metrics.*" && drop.if_exists
711        ));
712    }
713
714    #[test]
715    fn timeseries_metric_and_slo_errors_are_reported() {
716        for sql in [
717            "CREATE TIMESERIES events WITH RETENTION 1 d",
718            "CREATE METRICS telemetry TENANT (ctx.tenant)",
719            "CREATE METRIC svc.latency TYPE gauge",
720            "CREATE METRIC svc.latency TYPE gauge ROLE sli QUERY 42",
721            "ALTER METRIC svc.latency ROLE sli",
722            "CREATE SLO api.availability ON svc.latency WINDOW 1 h",
723            "CREATE HYPERTABLE events TIME_COLUMN ts CHUNK_INTERVAL 'not-duration'",
724        ] {
725            assert!(parse_query(sql).is_err(), "{sql} should not parse");
726        }
727    }
728}
reddb_rql/parser/timeseries.rs

reddb_rql/parser/
timeseries.rs
